Migrate node removal and port allow runners onto the task runner

Both are mechanical translations of their suspend points into the
handler vocabulary, and neither had a retry counter to hand over — every
gate in them deferred.

Node removal's "incomplete, retry later" pass becomes TaskProgress. It
is progress, not failure: the removal waits on device failure-migration
that can take hours, and treating each poll as a retry would have earned
it an exponential backoff it never had.

Port allow keeps its documented opt-out — no IN_ACTIVATION eligibility
gate, because activation needs exactly these ports open. Two of its
paths returned bare, leaving the task untouched for the loop to revisit;
under the driver a bare return means success, so both are now explicit
defers. Its abort-on-half-open-hublvol path keeps the abort side effect
in the handler and raises TaskAbort for the outcome.

Adds an import-smoke over all thirteen migrated runners that also
asserts no two of them claim the same task function name — the lease
guards against a second host, not against a second spec.

Author: mxsrc

simplyblock_core/services/tasks_runner_node_removal.py

@@ -1,11 +1,12 @@
 # coding=utf-8
-import time
-
-
 from simplyblock_core import db_controller, storage_node_ops, utils, constants
-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.services.task_runner_base import (
+    RunnerSpec,
+    TaskProgress,
+    serve,
+)
 
 
 logger = utils.get_logger(__name__)
@@ -19,76 +20,29 @@ def process_task(task):
 
     node_removal_orchestrate is idempotent and resumable: it returns True only
     when the node is fully REMOVED, and False to mean "incomplete, retry later"
-    (most commonly: device failure-migration still in progress). On False we
-    suspend the task so the outer loop revisits it on the next tick instead of
-    busy-spinning here for what can be hours.
+    (most commonly: device failure-migration still in progress, which can take
+    hours). Incomplete is progress, not failure — it consumes no retry, and the
+    task stays RUNNING so the next tick picks it straight back up.
     """
-    if task.canceled:
-        task.function_result = "canceled"
-        task.status = JobSchedule.STATUS_DONE
-        task.write_to_db(db.kv_store)
-        return True
-
-    cluster = db.get_cluster_by_id(task.cluster_id)
-    if cluster.status == Cluster.STATUS_IN_ACTIVATION:
-        task.function_result = "cluster is in_activation, waiting"
-        task.status = JobSchedule.STATUS_SUSPENDED
-        task.write_to_db(db.kv_store)
-        return False
+    force_remove = bool(task.function_params.get("force_remove", False))
+    if not storage_node_ops.node_removal_orchestrate(task.node_id, force_remove=force_remove):
+        raise TaskProgress("removal in progress, retrying")
 
-    if task.status != JobSchedule.STATUS_RUNNING:
-        task.status = JobSchedule.STATUS_RUNNING
-        task.write_to_db(db.kv_store)
+    task.function_result = "Node removed"
 
-    force_remove = bool(task.function_params.get("force_remove", False))
-    try:
-        done = storage_node_ops.node_removal_orchestrate(task.node_id, force_remove=force_remove)
-    except Exception as e:
-        logger.error(f"Node-removal task {task.uuid} raised: {e}")
-        logger.exception(e)
-        task.function_result = f"error: {e}"
-        task.retry += 1
-        task.status = JobSchedule.STATUS_SUSPENDED
-        task.write_to_db(db.kv_store)
-        return False
 
-    if done:
-        task.function_result = "Node removed"
-        task.status = JobSchedule.STATUS_DONE
-        task.write_to_db(db.kv_store)
-        return True
+SPEC = RunnerSpec(
+    name="tasks-runner-node-removal",
+    function_names=[JobSchedule.FN_NODE_REMOVAL],
+    handler=process_task,
+    is_eligible=lambda task, cluster: cluster.status != Cluster.STATUS_IN_ACTIVATION,
+    interval=constants.TASK_EXEC_INTERVAL_SEC,
+)
 
-    # 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
 
+def main():
+    serve(SPEC)
 
-logger.info("Starting Tasks runner node removal...")
 
-while True:
-    time.sleep(constants.TASK_EXEC_INTERVAL_SEC)
-    clusters = db.get_clusters()
-    if not clusters:
-        logger.error("No clusters found!")
-        continue
-    for cl in clusters:
-        tasks = db.get_job_tasks(cl.get_id(), reverse=False)
-        for task in tasks:
-            if task.function_name != JobSchedule.FN_NODE_REMOVAL:
-                continue
-            if task.status == JobSchedule.STATUS_DONE:
-                continue
-            # get a fresh object: cancel/other writers may have changed it
-            task = db.get_task_by_id(task.uuid)
-            # Lease gate: skip a task another live runner host owns.
-            if not tasks_controller.claim_task(task):
-                logger.info(f"Node-removal task {task.uuid} owned by another runner host; skipping")
-                continue
-            try:
-                process_task(task)
-            except Exception as e:
-                logger.error(f"Node-removal task {task.uuid} processing crashed: {e}")
-                logger.exception(e)
+if __name__ == "__main__":
+    main()