feat(replication): hold cutover_pending per-task for operator preconn…

simplyblock_core/constants.py

@@ -369,6 +369,10 @@ def get_config_var(name, default=None):
 # (snapshot -> wait replicated -> snapshot -> wait) before the final freeze.
 # Two rounds normally complete within 2 replication intervals + transfer time.
 REPL_CUTOVER_SHRINK_TIMEOUT_SEC = 900
+# Safety timeout for the operator preconnect signal. The task suspends indefinitely
+# waiting for POST .../replication/cutover-proceed; this is the fallback deadline
+# if the operator is unavailable. Cutover proceeds regardless after this many seconds.
+REPL_CUTOVER_PROCEED_TIMEOUT_SEC = 120
 
 SPDK_PROXY_MULTI_THREADING_ENABLED=True
 SPDK_PROXY_TIMEOUT=60*5

simplyblock_core/controllers/replication_policy_controller.py

@@ -359,6 +359,24 @@ def _failover_volumes(volumes, what):
     return results
 
 
+def set_cutover_proceed(lvol_id):
+    """Signal that the operator has connected the target NVMe paths.
+
+    Finds the cutover_pending LVolReplication for *lvol_id* (source side) and
+    sets cutover_proceed = True so the task runner advances past the wait.
+
+    Returns the replication ID on success, raises KeyError when no matching
+    cutover_pending record is found.
+    """
+    rep = _active_relationship(lvol_id)
+    if rep is None or rep.state != LVolReplication.STATE_CUTOVER_PENDING:
+        raise KeyError(
+            f"No cutover_pending replication found for volume {lvol_id}")
+    rep.cutover_proceed = True
+    rep.write_to_db(db.kv_store)
+    return rep.get_id()
+
+
 def get_relationship(lvol_id):
     """The replication relationship of *lvol_id*, source or target side.
 

simplyblock_core/models/lvol_model.py

@@ -144,6 +144,10 @@ class LVolReplication(BaseModel):
     # client keeps the same NQN/namespace across fail-over and migration.
     target_nqn: str = ""
     target_ns_id: int = 0
+    # Set to True by POST .../replication/cutover-proceed once the operator has
+    # connected the target NVMe paths. The task runner waits for this before
+    # calling run_cutover(); REPL_CUTOVER_PROCEED_TIMEOUT_SEC is the safety fallback.
+    cutover_proceed: bool = False
 
 class LVolMini(BaseModel):
     lvol_uuid: str = ""

simplyblock_core/services/tasks_runner_replication_final.py

@@ -166,6 +166,31 @@ def task_runner(task: JobSchedule):
                 return _finalize(task, False, err)
             params = task.function_params
 
+        # Wait for the operator to signal that target NVMe paths are connected
+        # (operator calls POST .../replication/cutover-proceed after its preconnect
+        # Job succeeds). REPL_CUTOVER_PROCEED_TIMEOUT_SEC is the safety fallback
+        # so cutover proceeds even if the operator is unavailable.
+        replication_id = params.get("replication_id")
+        if replication_id:
+            try:
+                rep = db.get_lvol_replication_by_id(replication_id)
+                if not rep.cutover_proceed:
+                    if "cutover_proceed_timeout" not in params:
+                        params["cutover_proceed_timeout"] = (
+                            int(time.time()) + constants.REPL_CUTOVER_PROCEED_TIMEOUT_SEC)
+                        task.write_to_db(db.kv_store)
+                    if int(time.time()) < params["cutover_proceed_timeout"]:
+                        task.function_result = "cutover_pending: waiting for preconnect signal"
+                        task.status = JobSchedule.STATUS_SUSPENDED
+                        task.write_to_db(db.kv_store)
+                        return False
+                    logger.warning(
+                        "cutover proceed timeout for replication %s; proceeding without signal",
+                        replication_id)
+            except KeyError:
+                logger.warning(
+                    "replication record %s not found; proceeding with cutover", replication_id)
+
         try:
             ok, err = replication_final_step.run_cutover(
                 src_node, tgt_node, lvol,

simplyblock_web/api/v2/cluster/storage_pool/volume/replication.py

@@ -159,6 +159,21 @@ def failback(cluster: Cluster, pool: StoragePool, volume: Volume, body: Failback
     return Response(status_code=204)
 
 
+@api.post('/cutover-proceed', name='clusters:storage-pools:volumes:replication:cutover-proceed',
+          status_code=204, responses={204: {"content": None}})
+def cutover_proceed(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response:
+    """Signal that target NVMe paths are connected and cutover may proceed.
+
+    Called by the operator after its preconnect Job succeeds. The task runner
+    is suspended waiting for this signal; once set, it advances to the ANA flip.
+    """
+    try:
+        replication_policy_controller.set_cutover_proceed(volume.get_id())
+    except KeyError as exc:
+        raise HTTPException(404, str(exc))
+    return Response(status_code=204)
+
+
 @api.get('/tasks', name='clusters:storage-pools:volumes:replication:tasks')
 def list_tasks(cluster: Cluster, pool: StoragePool, volume: Volume) -> List[TaskDTO]:
     return [TaskDTO.from_model(task) for task in lvol_controller.list_replication_tasks(volume.get_id())]