From f6cd7ad56e087a7b0949f7295ee890b27486e7b8 Mon Sep 17 00:00:00 2001 From: ebrahim savari <106956085+EbiRider@users.noreply.github.com> Date: Mon, 17 Aug 2026 06:08:03 +0330 Subject: [PATCH 1/7] inital implmentation: lvol migrate from secondary --- .../controllers/migration_controller.py | 111 ++++++++++++++++-- simplyblock_core/models/lvol_migration.py | 7 ++ .../models/lvol_migration_group.py | 7 ++ .../services/tasks_runner_batch_migration.py | 81 +++++++++---- .../services/tasks_runner_lvol_migration.py | 86 ++++++++++---- simplyblock_web/api/v2/_dtos.py | 4 + tests/unit/tasks/test_retry_ceiling.py | 6 + 7 files changed, 251 insertions(+), 51 deletions(-) diff --git a/simplyblock_core/controllers/migration_controller.py b/simplyblock_core/controllers/migration_controller.py index 49a173f874..d86e1d0a7b 100644 --- a/simplyblock_core/controllers/migration_controller.py +++ b/simplyblock_core/controllers/migration_controller.py @@ -103,10 +103,20 @@ def start_migration(migration_id, if lvol.status != LVol.STATUS_ONLINE: raise ValueError(f"Volume is not online (status={lvol.status})") - source_node_id = lvol.node_id + # source_node_id / active_source_node_id are read from the migration record + # (set once by create_migration()), never re-derived from lvol.node_id here — + # re-deriving could pick a different fallback than create_migration did if + # node health changed in between. + source_node_id = migration.source_node_id try: - source_node = db.get_storage_node_by_id(source_node_id) + db.get_storage_node_by_id(source_node_id) + except KeyError as e: + raise ValueError(str(e)) + + active_source_node_id = migration.active_source_node_id or source_node_id + try: + active_source_node = db.get_storage_node_by_id(active_source_node_id) except KeyError as e: raise ValueError(str(e)) @@ -118,19 +128,24 @@ def start_migration(migration_id, if source_node_id == target_node_id: raise ValueError("Source and target nodes must be different") - if source_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): - raise ValueError(f"Source node is not online (status={source_node.status})") + if active_source_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): + raise ValueError(f"Source node is not online (status={active_source_node.status})") if target_node.status != StorageNode.STATUS_ONLINE: raise ValueError(f"Target node is not online (status={target_node.status})") + if active_source_node_id != source_node_id: + logger.info( + f"start_migration {migration.uuid}: source primary {source_node_id} is offline; " + f"continuing with pre-selected fallback source {active_source_node_id}") + cluster = db.get_cluster_by_id(migration.cluster_id) if cluster.status != Cluster.STATUS_ACTIVE: raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})") if not _can_add_lvol_migration(cluster.get_id()): raise PreconditionError(f"Cluster {cluster.get_id()} is rebalancing; wait for it to finish before migrating") - for node_id in (source_node_id, target_node_id): + for node_id in {source_node_id, active_source_node_id, target_node_id}: if tasks_controller.get_active_node_mig_task(migration.cluster_id, node_id): raise PreconditionError(f"Node {node_id} has a data migration in progress; wait for it to finish") @@ -474,6 +489,50 @@ def _add(uid): return result +def _resolve_active_source_node(primary_node, target_node_id): + """ + Decide which node the migration will actually issue source-side RPCs + against: *primary_node* itself when reachable, otherwise its online + secondary, otherwise its online tertiary. + + This is called exactly once, at create time (create_migration / + create_batch_migration). The result is persisted as + migration.active_source_node_id / group.active_source_node_id and must + never be re-derived afterward — start_migration/start_batch_migration and + the task runners only ever read it. + + Raises ValueError if the primary is unreachable and no replica is + online either. Raises PreconditionError if the resolved node is the + same as target_node_id (can't migrate a replica onto itself). + """ + if primary_node.status in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): + active_node = primary_node + else: + active_node = None + for replica_id in (primary_node.secondary_node_id, primary_node.tertiary_node_id): + if not replica_id: + continue + try: + replica = db.get_storage_node_by_id(replica_id) + except KeyError: + continue + if replica.status == StorageNode.STATUS_ONLINE: + active_node = replica + break + if active_node is None: + raise ValueError( + f"Source node is not online (status={primary_node.status}) " + f"and no online secondary/tertiary replica is available") + + if active_node.get_id() == target_node_id: + raise PreconditionError( + f"Cannot migrate to node {target_node_id}: source primary " + f"{primary_node.get_id()} is offline and {target_node_id} is " + f"currently serving as the fallback source for this volume") + + return active_node + + def _is_snap_on_node(snap_id, node_id): """Return True if *snap_id* already has a copy on *node_id*. @@ -1003,6 +1062,12 @@ def create_migration(lvol_id, target_node_id, except KeyError: raise ValueError(f"Source node {src_node_id} not found") + active_src_node = _resolve_active_source_node(src_node, target_node_id) + if active_src_node.get_id() != src_node_id: + logger.warning( + f"create_migration: source primary {src_node_id} is offline; " + f"using {active_src_node.get_id()} as the effective source for lvol={lvol_id}") + cluster = db.get_cluster_by_id(tgt_node.cluster_id) if cluster.status != Cluster.STATUS_ACTIVE: raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})") @@ -1315,6 +1380,7 @@ def create_migration(lvol_id, target_node_id, migration.cluster_id = tgt_node.cluster_id migration.lvol_id = lvol_id migration.source_node_id = lvol.node_id + migration.active_source_node_id = active_src_node.get_id() migration.target_node_id = target_node_id migration.phase = LVolMigration.PHASE_PRE_CREATED migration.status = LVolMigration.STATUS_NEW @@ -1401,6 +1467,10 @@ def create_batch_migration(lvol_id, target_node_id, # Pre-create individual migration records for each member. # connect_strings come from the master (ns_id=1) since the NQN is shared. + # Each create_migration() call independently resolves the same active + # source node (all members share the same primary/lvstore), so the + # group's own active_source_node_id below is read from the first member's + # already-resolved record rather than re-resolved here. member_records = [] # list of (ns_id, migration_id) master_connect_strings = [] for member in members: @@ -1414,6 +1484,14 @@ def create_batch_migration(lvol_id, target_node_id, if member.ns_id == 1: master_connect_strings = connect_strings + active_source_node_id = source_node_id + if member_records: + try: + active_source_node_id = db.get_migration_by_id( + member_records[0]["migration_id"]).active_source_node_id or source_node_id + except KeyError: + pass + # Compute snap ownership: snap_uuid → lvol_uuid, then remap to migration_id. lvol_uuid_to_migration_id = { member.uuid: rec["migration_id"] @@ -1438,6 +1516,11 @@ def create_batch_migration(lvol_id, target_node_id, group.uuid = str(uuid.uuid4()) group.cluster_id = tgt_node.cluster_id group.source_node_id = source_node_id + group.active_source_node_id = active_source_node_id + if active_source_node_id != source_node_id: + logger.warning( + f"create_batch_migration: source primary {source_node_id} is offline; " + f"using {active_source_node_id} as the effective source for group NQN={lvol.nqn}") group.target_node_id = target_node_id group.target_nqn = lvol.nqn group.members = member_records @@ -1491,7 +1574,22 @@ def start_batch_migration(group_id, raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})") if not _can_add_lvol_migration(cluster.get_id()): raise PreconditionError(f"Cluster {cluster.get_id()} is rebalancing; wait for it to finish before migrating") - for node_id in (group.source_node_id, group.target_node_id): + + # active_source_node_id is read-only here — it was resolved once, at + # create_batch_migration() time, and must never be re-derived. + active_source_node_id = group.active_source_node_id or group.source_node_id + try: + active_source_node = db.get_storage_node_by_id(active_source_node_id) + except KeyError as e: + raise ValueError(str(e)) + if active_source_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): + raise ValueError(f"Source node is not online (status={active_source_node.status})") + if active_source_node_id != group.source_node_id: + logger.info( + f"start_batch_migration {group_id}: source primary {group.source_node_id} is offline; " + f"continuing with pre-selected fallback source {active_source_node_id}") + + for node_id in {group.source_node_id, active_source_node_id, group.target_node_id}: if tasks_controller.get_active_node_mig_task(group.cluster_id, node_id): raise PreconditionError(f"Node {node_id} has a data migration in progress; wait for it to finish") @@ -1524,7 +1622,6 @@ def start_batch_migration(group_id, if s not in snaps_on_target and group.snap_owners.get(s) != migration_id] - migration.source_node_id = lvol.node_id migration.phase = LVolMigration.PHASE_SNAP_COPY migration.snap_migration_plan = owned_snaps migration.snaps_migrated = [] diff --git a/simplyblock_core/models/lvol_migration.py b/simplyblock_core/models/lvol_migration.py index b5c53d5c28..d4e46680e7 100644 --- a/simplyblock_core/models/lvol_migration.py +++ b/simplyblock_core/models/lvol_migration.py @@ -56,6 +56,13 @@ class LVolMigration(BaseModel): source_node_id: str = "" target_node_id: str = "" + # Node to actually issue source-side RPCs against. Equals source_node_id + # (the primary) unless the primary was offline at create_migration() time, + # in which case this is the online secondary/tertiary replica that was + # chosen as the effective source. Resolved once at create time and never + # re-derived afterward — the runner only ever reads it. + active_source_node_id: str = "" + # --- Phase tracking --- phase: str = "" diff --git a/simplyblock_core/models/lvol_migration_group.py b/simplyblock_core/models/lvol_migration_group.py index f5edee7e47..39454ffce1 100644 --- a/simplyblock_core/models/lvol_migration_group.py +++ b/simplyblock_core/models/lvol_migration_group.py @@ -68,6 +68,13 @@ class LVolMigrationGroup(BaseModel): source_node_id: str = "" target_node_id: str = "" + # Node to actually issue source-side RPCs against. Equals source_node_id + # (the primary) unless the primary was offline at create_batch_migration() + # time, in which case this is the online secondary/tertiary replica chosen + # as the effective source. Resolved once at create time and never + # re-derived afterward — the runner only ever reads it. + active_source_node_id: str = "" + # Shared NVMe-oF NQN created on the target during PRECREATE. target_nqn: str = "" diff --git a/simplyblock_core/services/tasks_runner_batch_migration.py b/simplyblock_core/services/tasks_runner_batch_migration.py index 6890a41805..a27d4eadb7 100644 --- a/simplyblock_core/services/tasks_runner_batch_migration.py +++ b/simplyblock_core/services/tasks_runner_batch_migration.py @@ -464,7 +464,8 @@ def _commit_intermediate_snapshot_chain(group, member_migrations, tgt_node, tgt_ return None -def _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc): +def _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=None): """ After a successful bdev_lvol_batch_final_step, drive clients to the new target. @@ -487,7 +488,8 @@ def _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node 4. Remove old SRC-port listener from overlap TGT nodes if port changed """ nqn = group.target_nqn - src_paths, tgt_paths, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + src_paths, tgt_paths, overlap_ids = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) src_port_by_id = {p['node_id']: p['port'] for p in src_paths} # Detect and repair a target-side node restart that wiped the migration's @@ -674,7 +676,8 @@ def _flip_all_required(rpc, ips, port, trtype, state, label, attempts=3): f"{tgt['node_id'][:8]} (non-fatal): {e}") -def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, src_rpc, tgt_rpc): +def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=None): """ Wait for all workers to reach intermediates_done, then call bdev_lvol_batch_final_step. Returns (batch_ok, error). @@ -729,8 +732,11 @@ def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, s # synchronous final-step transfer below (see the diagnostic block further # down for the current, temporarily-widened version of this). nqn = group.target_nqn - src_paths, tgt_paths, _ = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) - src_replica_paths = src_paths[1:] # secondary/tertiary only; used for the failure-path revert below + src_paths, tgt_paths, _ = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) + # secondary/tertiary only -- the active source is frozen internally by the + # RPC below; used for the failure-path revert further down. + src_replica_paths = src_paths[1:] def _flip(rpc, ip, port, trtype, state, label): try: @@ -909,7 +915,8 @@ def _revert_src_replicas(reason): logger.warning( f"Group {group.uuid[:8]}: add_clone for member {m.uuid[:8]} (non-fatal): {e}") - _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc) + _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) # Hub controller left attached on both success and failure — hub_manager # owns its lifecycle entirely via its own idle timeout. Detaching it here @@ -953,14 +960,23 @@ def _handle_cleanup_source_barrier(group): return expected.issubset(done_set) -def _delete_source_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc): +def _delete_source_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc, primary_src_node=None): """ - Delete the source NVMe-oF subsystem on all SRC replicas (primary, secondary, - tertiary). Overlap nodes (which also host TGT replicas) are skipped because - the subsystem is still in use on those nodes. Best-effort. + Delete the source NVMe-oF subsystem on all SRC replicas (active source, + plus whichever of the primary's secondary/tertiary are still online and + aren't the active source itself). Overlap nodes (which also host TGT + replicas) are skipped because the subsystem is still in use on those + nodes. Best-effort. + + *src_node* is the active source (primary, or its fallback replica when + the primary was offline at create time). *primary_src_node* — the true + primary — is only consulted for its own secondary_node_id/tertiary_node_id + fields, since a replica's own such fields describe an unrelated pairing. """ nqn = group.target_nqn - _, _, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + primary_src_node = primary_src_node or src_node + _, _, overlap_ids = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) def _try_delete(rpc, node_id, label): if node_id in overlap_ids: @@ -972,24 +988,25 @@ def _try_delete(rpc, node_id, label): except Exception as e: logger.warning(f"Group {group.uuid[:8]}: {label} source subsystem delete (non-fatal): {e}") - _try_delete(src_rpc, src_node.get_id(), "primary") + _active_label = "active-source" if src_node.get_id() != primary_src_node.get_id() else "primary" + _try_delete(src_rpc, src_node.get_id(), _active_label) - if src_node.secondary_node_id: + if primary_src_node.secondary_node_id and primary_src_node.secondary_node_id != src_node.get_id(): try: - sec_node = db.get_storage_node_by_id(src_node.secondary_node_id) + sec_node = db.get_storage_node_by_id(primary_src_node.secondary_node_id) sec_rpc = _make_rpc(sec_node) _try_delete(sec_rpc, sec_node.get_id(), "secondary") except Exception as e: logger.warning( f"Group {group.uuid[:8]}: secondary src node lookup (non-fatal): {e}") - tert_node = _get_source_tertiary_node(src_node) - if tert_node: + tert_node = _get_source_tertiary_node(primary_src_node) + if tert_node and tert_node.get_id() != src_node.get_id(): tert_rpc = _make_rpc(tert_node) _try_delete(tert_rpc, tert_node.get_id(), "tertiary") -def _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc): +def _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc, primary_src_node=None): """Delete the target NVMe-oF subsystem on all TGT replicas. Best-effort. Nodes that appear in both SRC and TGT replica sets (overlap nodes) share the @@ -999,7 +1016,8 @@ def _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc): nqn = group.target_nqn try: - _, _, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + _, _, overlap_ids = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) except Exception as e: logger.warning( f"Group {group.uuid[:8]}: _build_paths in _delete_target_subsystem (non-fatal): {e}") @@ -1102,12 +1120,25 @@ def task_runner(task): task.write_to_db(db.kv_store) return True + # primary_src_node is the true primary (kept only for HA topology lookups + # via its own secondary_node_id/tertiary_node_id). src_node is the node + # this group actually issues source-side RPCs against — the primary when + # reachable, otherwise the replica pinned once at create_batch_migration() + # time as group.active_source_node_id. All data-plane calls below must use + # src_node/src_rpc, never primary_src_node. try: - src_node = db.get_storage_node_by_id(group.source_node_id) + primary_src_node = db.get_storage_node_by_id(group.source_node_id) except KeyError: return _batch_budget_suspend( task, group, group_id, f"source node {group.source_node_id} not found") + try: + src_node = db.get_storage_node_by_id( + group.active_source_node_id or group.source_node_id) + except KeyError: + return _batch_budget_suspend( + task, group, group_id, "active source node not found") + try: tgt_node = db.get_storage_node_by_id(group.target_node_id) except KeyError: @@ -1170,7 +1201,8 @@ def task_runner(task): task.write_to_db(db.kv_store) return False - fresh_src = db.get_storage_node_by_id(group.source_node_id) + fresh_src = db.get_storage_node_by_id( + group.active_source_node_id or group.source_node_id) if fresh_src.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): logger.warning( f"Group {group_id[:8]}: source node unavailable " @@ -1212,7 +1244,8 @@ def task_runner(task): # ── PHASE_INTERMEDIATE: wait for intermediates, then batch_final_step ──── if phase == LVolMigrationGroup.PHASE_INTERMEDIATE: batch_ok, err = _handle_intermediate_barrier( - group, member_migrations, src_node, tgt_node, src_rpc, tgt_rpc) + group, member_migrations, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) if err: logger.error(f"Group {group_id[:8]}: intermediate barrier error: {err}") @@ -1261,7 +1294,8 @@ def task_runner(task): task.write_to_db(db.kv_store) return False - _delete_source_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc) + _delete_source_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) group.phase = LVolMigrationGroup.PHASE_COMPLETED group.status = LVolMigrationGroup.STATUS_DONE @@ -1281,7 +1315,8 @@ def task_runner(task): task.write_to_db(db.kv_store) return False - _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc) + _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) group.status = LVolMigrationGroup.STATUS_FAILED group.write_to_db(db.kv_store) diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index f6ac117b2c..2075260843 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -581,7 +581,7 @@ def _get_source_tertiary_node(src_node): -def _build_paths(src_node, tgt_node, src_rpc, tgt_rpc): +def _build_paths(src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=None): """Build ordered path lists for source and target nodes and compute overlap. Returns (src_paths, tgt_paths, overlap_ids) where each path entry is: @@ -593,7 +593,20 @@ def _build_paths(src_node, tgt_node, src_rpc, tgt_rpc): Port is role-specific: SRC entries use src_node.lvstore; TGT entries use tgt_node.lvstore. Adding tertiary support = append one more entry to each list; all callers automatically handle it via loop/set operations. + + *src_node* is the node actually driving the transfer (position 0 in + src_paths) — normally the source primary, but the online secondary/ + tertiary when the primary was offline at migration create time + (migration.active_source_node_id). *primary_src_node* — defaulting to + src_node when the fallback isn't in play — is only consulted for its own + secondary_node_id/tertiary_node_id fields, which describe the primary's + true HA replicas; a replica node's own such fields describe an unrelated + pairing and must never be used for this lookup. Whichever replica was + chosen as src_node is excluded from the discovered peer set so it isn't + listed twice. """ + primary_src_node = primary_src_node or src_node + def _entry(node, rpc, lvstore): trtype, ip = _get_migration_nic(node) fabric = trtype.lower() @@ -611,16 +624,18 @@ def _entry(node, rpc, lvstore): } src_paths = [_entry(src_node, src_rpc, src_node.lvstore)] - if src_node.secondary_node_id: + _src_seen_ids = {src_node.get_id()} + if primary_src_node.secondary_node_id and primary_src_node.secondary_node_id not in _src_seen_ids: try: - ss = db.get_storage_node_by_id(src_node.secondary_node_id) + ss = db.get_storage_node_by_id(primary_src_node.secondary_node_id) if ss.status == StorageNode.STATUS_ONLINE: src_paths.append(_entry(ss, _make_rpc(ss), src_node.lvstore)) + _src_seen_ids.add(ss.get_id()) except KeyError: pass - if src_node.tertiary_node_id: + if primary_src_node.tertiary_node_id and primary_src_node.tertiary_node_id not in _src_seen_ids: try: - ts = db.get_storage_node_by_id(src_node.tertiary_node_id) + ts = db.get_storage_node_by_id(primary_src_node.tertiary_node_id) if ts.status == StorageNode.STATUS_ONLINE: src_paths.append(_entry(ts, _make_rpc(ts), src_node.lvstore)) except KeyError: @@ -1717,7 +1732,7 @@ def _take_intermediate_snapshot(migration): ) -def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): +def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=None): """ Drive the LVOL_MIGRATE phase. @@ -1754,7 +1769,8 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): # overlap_ids: nodes that appear in BOTH source and target paths — they # already have a subsystem (from SRC role); their namespace is swapped in # the Done handler's step 4. - src_paths, tgt_paths, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + src_paths, tgt_paths, overlap_ids = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) src_replica_paths = src_paths[1:] # secondary/tertiary only; primary stays live until cutover # Detect and repair a target-side node restart that wiped the migration's @@ -2490,7 +2506,7 @@ def _rename_with_fallback(current_short, label): lvol.write_to_db(db.kv_store) -def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): +def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc, primary_src_node=None): """ Best-effort source cleanup after a successful migration. The lvol is already live on the target — this phase only removes source-side artifacts @@ -2562,9 +2578,19 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): migration.transfer_context = ctx migration.write_to_db(db.kv_store) - src_sec = _get_source_secondary_node(src_node) + # Peer discovery must key off the true primary's own secondary_node_id/ + # tertiary_node_id fields (a replica node's own such fields describe an + # unrelated pairing — see _build_paths). Whichever replica is already + # src_node (the active fallback source) is excluded so it isn't cleaned + # up twice. + _primary_src_node = primary_src_node or src_node + src_sec = _get_source_secondary_node(_primary_src_node) + if src_sec is not None and src_sec.get_id() == src_node.get_id(): + src_sec = None src_sec_rpc = _make_rpc(src_sec) if src_sec else None - src_ter = _get_source_tertiary_node(src_node) + src_ter = _get_source_tertiary_node(_primary_src_node) + if src_ter is not None and src_ter.get_id() == src_node.get_id(): + src_ter = None src_ter_rpc = _make_rpc(src_ter) if src_ter else None # --- Delete source snapshots (best-effort, leader-routed) --- @@ -2603,7 +2629,7 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): else: logger.info(f"Step 8: removing source NVMe-oF subsystem {lvol.nqn}") _src_paths_cu, _, _overlap_ids_cu = _build_paths( - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=_primary_src_node) for _sp in _src_paths_cu: if _sp['node_id'] in _overlap_ids_cu: logger.info( @@ -2964,11 +2990,23 @@ def task_runner(task): migration_events.migration_phase_changed(migration) # --- Load nodes --- + # primary_src_node is the true primary (lvol.node_id) — kept only for HA + # topology lookups (its own secondary_node_id/tertiary_node_id fields). + # src_node is the node this migration actually issues source-side RPCs + # against: the primary when reachable, otherwise the online replica + # pinned once at create_migration() time as active_source_node_id. Every + # data-plane call below must use src_node/src_rpc, never primary_src_node. try: - src_node = db.get_storage_node_by_id(migration.source_node_id) + primary_src_node = db.get_storage_node_by_id(migration.source_node_id) except KeyError: return _budget_suspend(task, migration, migration_id, "source node not found") + try: + src_node = db.get_storage_node_by_id( + migration.active_source_node_id or migration.source_node_id) + except KeyError: + return _budget_suspend(task, migration, migration_id, "active source node not found") + try: tgt_node = db.get_storage_node_by_id(migration.target_node_id) except KeyError: @@ -3050,7 +3088,8 @@ def task_runner(task): try: if migration.migration_group_id: return _group_worker_phase_dispatch( - task, migration, phase, src_node, tgt_node, src_rpc, tgt_rpc) + task, migration, phase, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) if phase == LVolMigration.PHASE_SNAP_COPY: done, suspend, error = _handle_snap_copy( @@ -3059,11 +3098,14 @@ def task_runner(task): elif phase == LVolMigration.PHASE_LVOL_MIGRATE: done, suspend, error = _handle_lvol_migrate( - migration, src_node, tgt_node, src_rpc, tgt_rpc) + migration, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) next_phase = LVolMigration.PHASE_CLEANUP_SOURCE elif phase == LVolMigration.PHASE_CLEANUP_SOURCE: - done, suspend, error = _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc) + done, suspend, error = _handle_cleanup_source( + migration, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) next_phase = LVolMigration.PHASE_COMPLETED elif phase == LVolMigration.PHASE_CLEANUP_TARGET: @@ -3561,7 +3603,8 @@ def _group_worker_budget_suspend(task, migration, group_id, error_msg): return _suspend_task(task, migration, error_msg) -def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src_rpc, tgt_rpc): +def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=None): """ Complete phase dispatcher for FN_LVOL_MIG tasks that belong to a batch migration group (``migration.migration_group_id`` is set). @@ -3625,13 +3668,13 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src migration_events.migration_phase_changed(migration) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_LVOL_MIGRATE, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) if group.phase == LVolMigrationGroup.PHASE_CLEANUP_TARGET: migration.phase = LVolMigration.PHASE_CLEANUP_TARGET migration.write_to_db(db.kv_store) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_CLEANUP_TARGET, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) # Still waiting for other workers. task.write_to_db(db.kv_store) return False @@ -3710,7 +3753,7 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src migration_events.migration_phase_changed(migration) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_CLEANUP_SOURCE, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) if group.batch_result is False: migration.phase = LVolMigration.PHASE_CLEANUP_TARGET migration.transfer_context = {} @@ -3718,7 +3761,7 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src migration_events.migration_phase_changed(migration) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_CLEANUP_TARGET, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) task.write_to_db(db.kv_store) return False @@ -3726,7 +3769,8 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src if phase == LVolMigration.PHASE_CLEANUP_SOURCE: try: done, suspend, error = _handle_cleanup_source( - migration, src_node, src_rpc, tgt_node, tgt_rpc) + migration, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) except RPCException as exc: return _suspend_task(task, migration, str(exc)) diff --git a/simplyblock_web/api/v2/_dtos.py b/simplyblock_web/api/v2/_dtos.py index f72508c72c..99ca994c6d 100644 --- a/simplyblock_web/api/v2/_dtos.py +++ b/simplyblock_web/api/v2/_dtos.py @@ -648,6 +648,7 @@ class MigrationDTO(BaseModel): id: UUID lvol_id: str source_node_id: str + active_source_node_id: str target_node_id: str phase: str status: str @@ -668,6 +669,7 @@ def from_model(model: LVolMigration, connect_strings: Optional[List[NvmeConnectE id=UUID(model.uuid), lvol_id=model.lvol_id, source_node_id=model.source_node_id, + active_source_node_id=model.active_source_node_id or model.source_node_id, target_node_id=model.target_node_id, phase=model.phase, status=model.status, @@ -688,6 +690,7 @@ class BatchMigrationDTO(BaseModel): id: UUID cluster_id: str source_node_id: str + active_source_node_id: str target_node_id: str target_nqn: str phase: str @@ -702,6 +705,7 @@ def from_model(model: LVolMigrationGroup, connect_strings: Optional[List[NvmeCon id=UUID(model.uuid), cluster_id=model.cluster_id, source_node_id=model.source_node_id, + active_source_node_id=model.active_source_node_id or model.source_node_id, target_node_id=model.target_node_id, target_nqn=model.target_nqn, phase=model.phase, diff --git a/tests/unit/tasks/test_retry_ceiling.py b/tests/unit/tasks/test_retry_ceiling.py index c3769966dc..b5d514cd61 100644 --- a/tests/unit/tasks/test_retry_ceiling.py +++ b/tests/unit/tasks/test_retry_ceiling.py @@ -442,6 +442,12 @@ def _spec_batch_migration(runner, monkeypatch): group = MagicMock() group.phase = LVolMigrationGroup.PHASE_SNAP_COPY group.source_node_id = "src-1" + # MagicMock auto-creates a truthy attribute for anything unset; without + # this, `group.active_source_node_id or group.source_node_id` in the + # runner picks the mock object instead of "src-1", silently routing node + # lookups to the tgt_node branch below and masking the source-offline + # retry path this test exercises. + group.active_source_node_id = "src-1" group.target_node_id = "tgt-1" group.cluster_id = "cl-1" group.members = [{"migration_id": "mig-1"}] From 674b4626585eb53192d401dcaedbd7502d7d8708 Mon Sep 17 00:00:00 2001 From: ebrahim savari <106956085+EbiRider@users.noreply.github.com> Date: Mon, 17 Aug 2026 08:21:46 +0330 Subject: [PATCH 2/7] fixing an issue with wrong name for lvstore upon trying to migrate from secondary --- .../services/tasks_runner_batch_migration.py | 12 ++- .../services/tasks_runner_lvol_migration.py | 74 ++++++++++++------- 2 files changed, 56 insertions(+), 30 deletions(-) diff --git a/simplyblock_core/services/tasks_runner_batch_migration.py b/simplyblock_core/services/tasks_runner_batch_migration.py index a27d4eadb7..8dd3ba37e9 100644 --- a/simplyblock_core/services/tasks_runner_batch_migration.py +++ b/simplyblock_core/services/tasks_runner_batch_migration.py @@ -269,12 +269,17 @@ def _handle_snap_copy_barrier(group, member_migrations, tgt_node, tgt_rpc): return True, None -def _build_batch_final_args(group, member_migrations, src_node, tgt_node, tgt_rpc): +def _build_batch_final_args(group, member_migrations, src_node, tgt_node, tgt_rpc, + primary_src_node=None): """ Build the argument lists for bdev_lvol_batch_final_step, ordered by ns_id. Returns (lvol_names, lvol_ids, snapshot_names) or raises ValueError. """ + # The lvstore NAME is always the true primary's own lvstore, regardless + # of which node is actually driving the transfer as src_node/src_rpc — + # see tasks_runner_lvol_migration._build_paths' matching comment. + src_lvstore = (primary_src_node or src_node).lvstore mid_to_migration = {m.uuid: m for m in member_migrations} ordered_ids = group.ordered_migration_ids() @@ -295,7 +300,7 @@ def _build_batch_final_args(group, member_migrations, src_node, tgt_node, tgt_rp raise ValueError(f"migration {migration_id} not found in member_migrations") lvol = db.get_lvol_by_id(m.lvol_id) - src_composite = f"{src_node.lvstore}/{lvol.lvol_bdev}" + src_composite = f"{src_lvstore}/{lvol.lvol_bdev}" lvol_names.append(src_composite) tgt_bdev_short = _lvol_tgt_bdev_name(lvol.lvol_bdev) @@ -717,7 +722,8 @@ def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, s try: lvol_names, lvol_ids, snapshot_names = _build_batch_final_args( - group, member_migrations, src_node, tgt_node, tgt_rpc) + group, member_migrations, src_node, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) except (ValueError, KeyError) as e: # Hub controller left attached — hub_manager owns its lifecycle # entirely via its own idle timeout. diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index 2075260843..5e9b152fab 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -623,13 +623,18 @@ def _entry(node, rpc, lvstore): 'node_id': node.get_id(), } - src_paths = [_entry(src_node, src_rpc, src_node.lvstore)] + # The lvstore NAME is always the true primary's own lvstore — a replica + # node hosts this data under the primary's lvstore name, not its own (a + # node's own .lvstore is whatever IT owns as a primary elsewhere in the + # HA ring, which is unrelated). Only the node/rpc/IP differ per entry. + src_lvstore = primary_src_node.lvstore + src_paths = [_entry(src_node, src_rpc, src_lvstore)] _src_seen_ids = {src_node.get_id()} if primary_src_node.secondary_node_id and primary_src_node.secondary_node_id not in _src_seen_ids: try: ss = db.get_storage_node_by_id(primary_src_node.secondary_node_id) if ss.status == StorageNode.STATUS_ONLINE: - src_paths.append(_entry(ss, _make_rpc(ss), src_node.lvstore)) + src_paths.append(_entry(ss, _make_rpc(ss), src_lvstore)) _src_seen_ids.add(ss.get_id()) except KeyError: pass @@ -637,7 +642,7 @@ def _entry(node, rpc, lvstore): try: ts = db.get_storage_node_by_id(primary_src_node.tertiary_node_id) if ts.status == StorageNode.STATUS_ONLINE: - src_paths.append(_entry(ts, _make_rpc(ts), src_node.lvstore)) + src_paths.append(_entry(ts, _make_rpc(ts), src_lvstore)) except KeyError: pass @@ -884,7 +889,7 @@ def _setup_snap_transfer(snap, snap_index, src_node, tgt_node, src_rpc, tgt_rpc, trtype, tgt_sec=None, sec_rpc=None, tgt_ter=None, ter_rpc=None, lvol_size_mib=None, migration=None, - existing_bdev_info=_BDEV_INFO_UNSET): + existing_bdev_info=_BDEV_INFO_UNSET, primary_src_node=None): """ Prepare a single snapshot for async transfer: 1. Create writable lvol on target primary @@ -907,7 +912,7 @@ def _setup_snap_transfer(snap, snap_index, src_node, tgt_node, """ snap_uuid = snap.uuid snap_short = _snap_tgt_short_name(snap) - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite((primary_src_node or src_node).lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short}" # Step 1: create target lvol on primary. @@ -1182,7 +1187,7 @@ def _post_process_snap(snap: SnapShot, tgt_node: StorageNode, tgt_rpc: RPCClient return True, None -def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): +def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=None): """ Drive the SNAP_COPY phase. @@ -1223,6 +1228,10 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): plan = migration.snap_migration_plan trtype, _ = _get_migration_nic(tgt_node) ctx = migration.transfer_context or {} + # The lvstore NAME is always the true primary's own lvstore, regardless + # of which node (primary, secondary, or tertiary) is actually driving the + # transfer as src_node/src_rpc — see _build_paths' matching comment. + src_lvstore = (primary_src_node or src_node).lvstore # Snap bdevs on TGT must cover the full logical address range of the lvol, # not just each snap's own allocated clusters. @@ -1337,7 +1346,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): return False, True, f"Snapshot {snap_uuid} not found in DB" snap_short_tgt = _snap_tgt_short_name(snap) - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short_tgt}" # Idempotency: transfer already running from a previous crashed run @@ -1390,7 +1399,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): tgt_ter=tgt_ter, ter_rpc=ter_rpc, lvol_size_mib=_snap_lvol_size_mib, migration=migration, - existing_bdev_info=_existing_bdev) + existing_bdev_info=_existing_bdev, primary_src_node=primary_src_node) if t is None: return False, True, err @@ -1449,7 +1458,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): migration.write_to_db(db.kv_store) return False, True, f"Snapshot {snap_uuid} disappeared during transfer" - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) # Update transfer-done status for this entry if not t['transfer_done']: @@ -1521,7 +1530,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): # additional shrink pass is worth the overhead. while migration.intermediate_snap_rounds < migration.max_intermediate_snap_rounds: _lvol = db.get_lvol_by_id(migration.lvol_id) - _src_composite = f"{src_node.lvstore}/{_lvol.lvol_bdev}" + _src_composite = f"{src_lvstore}/{_lvol.lvol_bdev}" _delta = _get_lvol_delta_bytes(src_rpc, _src_composite) _threshold = constants.LVOL_MIG_INTERMEDIATE_SNAP_THRESHOLD_BYTES if migration.intermediate_snap_rounds > 0 and _delta is not None and _delta <= _threshold: @@ -1577,7 +1586,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): ter_rpc = _make_rpc(tgt_ter) snap_short_tgt = _snap_tgt_short_name(snap) - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short_tgt}" # Pre-cleanup: if a bdev exists on the target it is a writable leftover @@ -1753,7 +1762,7 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc, primar return False, True, str(e) trtype, _ = _get_migration_nic(tgt_node) - src_lvol_composite = f"{src_node.lvstore}/{lvol.lvol_bdev}" + src_lvol_composite = f"{(primary_src_node or src_node).lvstore}/{lvol.lvol_bdev}" tgt_lvol_bdev = _lvol_tgt_bdev_name(lvol.lvol_bdev) tgt_lvol_composite = f"{tgt_node.lvstore}/{tgt_lvol_bdev}" ctx = migration.transfer_context or {} @@ -2601,12 +2610,12 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc, prim try: snap = db.get_snapshot_by_id(snap_uuid) bdev_name = (source_snap_bdevs.get(snap_uuid) - or f"{src_node.lvstore}/{_snap_short_name(snap)}") + or f"{_primary_src_node.lvstore}/{_snap_short_name(snap)}") try: _delete_bdev_blocking(bdev_name, src_rpc, secondary_rpc=src_sec_rpc, tertiary_rpc=src_ter_rpc, all_nodes=[n for n in [src_node, src_sec, src_ter] if n], - lvs_name=src_node.lvstore) + lvs_name=_primary_src_node.lvstore) logger.info(f"Deleted source bdev {bdev_name}") except Exception as e: logger.warning(f"delete source bdev {bdev_name}: {e}") @@ -2645,13 +2654,13 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc, prim # lvol.lvol_bdev in the DB to the target name, so we must not use lvol.lvol_bdev. src_bdev_short = ctx.get('source_lvol_bdev') if lvol is not None and src_bdev_short: - src_lvol_composite = f"{src_node.lvstore}/{src_bdev_short}" + src_lvol_composite = f"{_primary_src_node.lvstore}/{src_bdev_short}" try: _delete_bdev_blocking( src_lvol_composite, src_rpc, secondary_rpc=src_sec_rpc, tertiary_rpc=src_ter_rpc, all_nodes=[n for n in [src_node, src_sec, src_ter] if n], - lvs_name=src_node.lvstore) + lvs_name=_primary_src_node.lvstore) logger.info(f"Deleted source lvol bdev {src_lvol_composite}") except Exception as e: logger.warning(f"Source lvol delete failed: {e}") @@ -3093,7 +3102,8 @@ def task_runner(task): if phase == LVolMigration.PHASE_SNAP_COPY: done, suspend, error = _handle_snap_copy( - migration, src_node, tgt_node, src_rpc, tgt_rpc) + migration, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) next_phase = LVolMigration.PHASE_LVOL_MIGRATE elif phase == LVolMigration.PHASE_LVOL_MIGRATE: @@ -3260,7 +3270,7 @@ def _post_process_snap_group(snap, migration): return True, None -def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): +def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=None): """ SNAP_COPY phase for a group worker. @@ -3274,6 +3284,10 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): plan = migration.snap_migration_plan trtype, _ = _get_migration_nic(tgt_node) ctx = migration.transfer_context or {} + # See _build_paths' matching comment: the lvstore NAME is always the true + # primary's own lvstore, regardless of which node is actually driving the + # transfer as src_node/src_rpc. + src_lvstore = (primary_src_node or src_node).lvstore try: _lvol_for_size = db.get_lvol_by_id(migration.lvol_id) @@ -3297,7 +3311,7 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): return False, True, f"Snapshot {snap_uuid} not found in DB" snap_short_tgt = _snap_tgt_short_name(snap) - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short_tgt}" existing_stat = src_rpc.bdev_lvol_transfer_stat(src_composite) @@ -3350,7 +3364,7 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): tgt_ter=_g_tgt_ter, ter_rpc=_g_ter_rpc, lvol_size_mib=_snap_lvol_size_mib, migration=migration, - existing_bdev_info=_existing_bdev) + existing_bdev_info=_existing_bdev, primary_src_node=primary_src_node) if t is None: return False, True, err @@ -3375,7 +3389,7 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): migration.write_to_db(db.kv_store) return False, True, f"Snapshot {snap_uuid} disappeared during transfer" - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) if not t['transfer_done']: result = src_rpc.bdev_lvol_transfer_stat(src_composite) if result is None: @@ -3413,7 +3427,8 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): return True, False, None -def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, target_round=0): +def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, + target_round=0, primary_src_node=None): """ INTERMEDIATE phase for a group worker. @@ -3432,6 +3447,10 @@ def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, """ trtype, _ = _get_migration_nic(tgt_node) ctx = migration.transfer_context or {} + # See _build_paths' matching comment: the lvstore NAME is always the true + # primary's own lvstore, regardless of which node is actually driving the + # transfer as src_node/src_rpc. + src_lvstore = (primary_src_node or src_node).lvstore # If we already took and transferred the intermediate snap for the round # the group is currently on, we're done. Otherwise the group has asked @@ -3502,7 +3521,7 @@ def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, tgt_ter=_g_tgt_ter, ter_rpc=_g_ter_rpc, lvol_size_mib=_snap_lvol_size_mib, migration=migration, - existing_bdev_info=_existing_bdev) + existing_bdev_info=_existing_bdev, primary_src_node=primary_src_node) if t is None: return False, True, err @@ -3523,7 +3542,7 @@ def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, migration.write_to_db(db.kv_store) return False, True, f"Intermediate snap {snap_uuid} disappeared" - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) if not t.get('transfer_done'): result = src_rpc.bdev_lvol_transfer_stat(src_composite) if result is None: @@ -3635,7 +3654,8 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src # Still transferring owned snaps. try: done, suspend, error = _handle_group_snap_copy( - migration, src_node, tgt_node, src_rpc, tgt_rpc) + migration, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) except RPCException as exc: # Charge this worker's own retry budget and report failure to # the group -- never decide/roll back unilaterally (see @@ -3691,12 +3711,12 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src migration.write_to_db(db.kv_store) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_CLEANUP_TARGET, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) try: done, suspend, error = _handle_group_intermediate( migration, src_node, tgt_node, src_rpc, tgt_rpc, - target_round=group.intermediate_round) + target_round=group.intermediate_round, primary_src_node=primary_src_node) except RPCException as exc: # Charge this worker's own retry budget and report failure to # the group -- never decide/roll back unilaterally (see From 67b54b9e2890cfe902d5ec7b75e317b7f424addd Mon Sep 17 00:00:00 2001 From: ebrahim savari <106956085+EbiRider@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:09:23 +0330 Subject: [PATCH 3/7] fix: transfer snapshots calls the secondary's main lvstore instead of the replica lvstore --- simplyblock_core/services/tasks_runner_lvol_migration.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index 5e9b152fab..b3f03d669c 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -1622,7 +1622,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc, primary_s tgt_ter=tgt_ter, ter_rpc=ter_rpc, lvol_size_mib=_snap_lvol_size_mib, migration=migration, - existing_bdev_info=_existing_bdev) + existing_bdev_info=_existing_bdev, primary_src_node=primary_src_node) if t is None: return False, True, err From 9581525f4905d7cdb91a1f3a7dee2bf3cbd62273 Mon Sep 17 00:00:00 2001 From: ebrahim savari <106956085+EbiRider@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:20:24 +0330 Subject: [PATCH 4/7] allowing the migration from secondary while the cluster status is degraded --- .../controllers/migration_controller.py | 45 ++++++-- .../migration/test_unit_controller.py | 107 ++++++++++++++++++ 2 files changed, 142 insertions(+), 10 deletions(-) diff --git a/simplyblock_core/controllers/migration_controller.py b/simplyblock_core/controllers/migration_controller.py index d86e1d0a7b..fd7f167797 100644 --- a/simplyblock_core/controllers/migration_controller.py +++ b/simplyblock_core/controllers/migration_controller.py @@ -134,13 +134,22 @@ def start_migration(migration_id, if target_node.status != StorageNode.STATUS_ONLINE: raise ValueError(f"Target node is not online (status={target_node.status})") - if active_source_node_id != source_node_id: + is_fallback_source = active_source_node_id != source_node_id + if is_fallback_source: logger.info( f"start_migration {migration.uuid}: source primary {source_node_id} is offline; " f"continuing with pre-selected fallback source {active_source_node_id}") cluster = db.get_cluster_by_id(migration.cluster_id) - if cluster.status != Cluster.STATUS_ACTIVE: + # A fallback migration exists precisely because its primary source node is + # down, which is what drives the cluster to DEGRADED in the first place + # (storage_node_monitor's one-node-down verdict) — requiring strict ACTIVE + # here would make the feature unusable in the scenario it exists for. + # Ordinary (non-fallback) migrations keep the stricter ACTIVE-only gate. + allowed_statuses = ( + (Cluster.STATUS_ACTIVE, Cluster.STATUS_DEGRADED) if is_fallback_source + else (Cluster.STATUS_ACTIVE,)) + if cluster.status not in allowed_statuses: raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})") if not _can_add_lvol_migration(cluster.get_id()): raise PreconditionError(f"Cluster {cluster.get_id()} is rebalancing; wait for it to finish before migrating") @@ -1063,13 +1072,21 @@ def create_migration(lvol_id, target_node_id, raise ValueError(f"Source node {src_node_id} not found") active_src_node = _resolve_active_source_node(src_node, target_node_id) - if active_src_node.get_id() != src_node_id: + is_fallback_source = active_src_node.get_id() != src_node_id + if is_fallback_source: logger.warning( f"create_migration: source primary {src_node_id} is offline; " f"using {active_src_node.get_id()} as the effective source for lvol={lvol_id}") cluster = db.get_cluster_by_id(tgt_node.cluster_id) - if cluster.status != Cluster.STATUS_ACTIVE: + # See the matching comment in start_migration(): a fallback migration's + # primary is down, which is what drives the cluster to DEGRADED, so the + # strict ACTIVE-only gate would make the feature unusable for the + # scenario it exists for. Non-fallback migrations keep the stricter gate. + allowed_statuses = ( + (Cluster.STATUS_ACTIVE, Cluster.STATUS_DEGRADED) if is_fallback_source + else (Cluster.STATUS_ACTIVE,)) + if cluster.status not in allowed_statuses: raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})") if not _can_add_lvol_migration(cluster.get_id()): raise PreconditionError(f"Cluster {cluster.get_id()} is rebalancing; wait for it to finish before migrating") @@ -1565,26 +1582,34 @@ def start_batch_migration(group_id, f"Group {group_id} is not in PHASE_PRE_CREATED (phase={group.phase})" ) + # active_source_node_id is read-only here — it was resolved once, at + # create_batch_migration() time, and must never be re-derived. + active_source_node_id = group.active_source_node_id or group.source_node_id + is_fallback_source = active_source_node_id != group.source_node_id + # Same preconditions as start_migration's single-lvol path — these are # only checked at create_batch_migration (precreate) time today, so a # cluster rebalance / conflicting node migration starting in the gap # before migrate-continue --batch would otherwise go unnoticed here. + # A fallback group's primary is down, which is what drives the cluster to + # DEGRADED in the first place, so the strict ACTIVE-only gate would make + # the feature unusable for the scenario it exists for (see the matching + # comment in start_migration()). Non-fallback groups keep the stricter gate. cluster = db.get_cluster_by_id(group.cluster_id) - if cluster.status != Cluster.STATUS_ACTIVE: + allowed_statuses = ( + (Cluster.STATUS_ACTIVE, Cluster.STATUS_DEGRADED) if is_fallback_source + else (Cluster.STATUS_ACTIVE,)) + if cluster.status not in allowed_statuses: raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})") if not _can_add_lvol_migration(cluster.get_id()): raise PreconditionError(f"Cluster {cluster.get_id()} is rebalancing; wait for it to finish before migrating") - - # active_source_node_id is read-only here — it was resolved once, at - # create_batch_migration() time, and must never be re-derived. - active_source_node_id = group.active_source_node_id or group.source_node_id try: active_source_node = db.get_storage_node_by_id(active_source_node_id) except KeyError as e: raise ValueError(str(e)) if active_source_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): raise ValueError(f"Source node is not online (status={active_source_node.status})") - if active_source_node_id != group.source_node_id: + if is_fallback_source: logger.info( f"start_batch_migration {group_id}: source primary {group.source_node_id} is offline; " f"continuing with pre-selected fallback source {active_source_node_id}") diff --git a/tests/integration/migration/test_unit_controller.py b/tests/integration/migration/test_unit_controller.py index 5865159c86..362d7ba247 100644 --- a/tests/integration/migration/test_unit_controller.py +++ b/tests/integration/migration/test_unit_controller.py @@ -12,10 +12,12 @@ import pytest from simplyblock_core.models.lvol_migration import LVolMigration +from simplyblock_core.models.lvol_migration_group import LVolMigrationGroup from simplyblock_core.models.lvol_model import LVol from simplyblock_core.models.snapshot import SnapShot from simplyblock_core.models.storage_node import StorageNode from simplyblock_core.models.cluster import Cluster +from simplyblock_core.exceptions import PreconditionError # Module under test (import after patching, but top-level import is fine since # we patch the db attribute before each individual call). @@ -692,6 +694,55 @@ def test_reject_target_node_offline(self): with pytest.raises(ValueError, match="Target node is not online"): ctl.start_migration("mig-uuid") + def test_reject_degraded_cluster_for_non_fallback_source(self): + # Primary is the active source (no fallback) -- a DEGRADED cluster + # must still be rejected here, matching the pre-existing strict + # ACTIVE-only gate for ordinary migrations. + mig = self._pre_created() + lvol = _lvol("lvol-1", "node-src") + src = _node("node-src") + tgt = _node("node-tgt") + mock_db = self._base_db(mig, lvol, src, tgt) + mock_db.get_cluster_by_id.return_value.status = Cluster.STATUS_DEGRADED + with patch.object(ctl, 'db', mock_db): + with pytest.raises(PreconditionError, match="not active"): + ctl.start_migration("mig-uuid") + + def test_allow_degraded_cluster_for_fallback_source(self): + # Primary is offline; active_source_node_id was pinned to the + # secondary at create_migration() time. The primary being down is + # exactly what drives the cluster to DEGRADED, so DEGRADED must be + # allowed here or the fallback feature could never actually run. + mig = self._pre_created() + mig.active_source_node_id = "node-sec" + lvol = _lvol("lvol-1", "node-src") + src = _node("node-src", status=StorageNode.STATUS_OFFLINE) + sec = _node("node-sec") + tgt = _node("node-tgt") + + mock_db = MagicMock() + mock_db.get_migration_by_id.return_value = mig + mock_db.get_lvol_by_id.return_value = lvol + nodes_by_id = {"node-src": src, "node-sec": sec, "node-tgt": tgt} + mock_db.get_storage_node_by_id.side_effect = lambda nid: nodes_by_id[nid] + mock_db.get_snapshots_by_node_id.return_value = [ + _snap("s1", "lvol-1", "node-src")] + cluster = Cluster() + cluster.uuid = "cluster-1" + cluster.status = Cluster.STATUS_DEGRADED + mock_db.get_cluster_by_id.return_value = cluster + mock_db.get_job_tasks.return_value = [] + mock_db.kv_store = MagicMock() + + with patch.object(ctl, 'db', mock_db), \ + patch('simplyblock_core.controllers.migration_controller.tasks_controller') as tc, \ + patch('simplyblock_core.controllers.migration_controller.migration_events'): + tc.add_lvol_mig_task.return_value = "task-uuid" + tc.get_active_node_mig_task.return_value = None + result = ctl.start_migration("mig-uuid") + + assert result == "mig-uuid" + def test_success_creates_task(self): mig = self._pre_created() lvol = _lvol("lvol-1", "node-src") @@ -714,3 +765,59 @@ def test_success_creates_task(self): assert result == "mig-uuid" tc.add_lvol_mig_task.assert_called_once() + + +# --------------------------------------------------------------------------- +# start_batch_migration – precondition validation (same DEGRADED/fallback +# gate as start_migration, applied to LVolMigrationGroup) +# --------------------------------------------------------------------------- + +class TestStartBatchMigrationPreconditions(unittest.TestCase): + + def _group(self, active_source_node_id=""): + g = LVolMigrationGroup() + g.uuid = "group-uuid" + g.cluster_id = "cluster-1" + g.source_node_id = "node-src" + g.target_node_id = "node-tgt" + g.active_source_node_id = active_source_node_id + g.members = [] + g.snap_owners = {} + g.phase = LVolMigrationGroup.PHASE_PRE_CREATED + return g + + def test_reject_degraded_cluster_for_non_fallback_source(self): + group = self._group() # active_source_node_id empty -> falls back to source_node_id + src = _node("node-src") + mock_db = MagicMock() + mock_db.get_migration_group_by_id.return_value = group + mock_db.get_storage_node_by_id.return_value = src + cluster = Cluster() + cluster.uuid = "cluster-1" + cluster.status = Cluster.STATUS_DEGRADED + mock_db.get_cluster_by_id.return_value = cluster + with patch.object(ctl, 'db', mock_db): + with pytest.raises(PreconditionError, match="not active"): + ctl.start_batch_migration("group-uuid") + + def test_allow_degraded_cluster_for_fallback_source(self): + group = self._group(active_source_node_id="node-sec") + sec = _node("node-sec") + mock_db = MagicMock() + mock_db.get_migration_group_by_id.return_value = group + mock_db.get_storage_node_by_id.return_value = sec + cluster = Cluster() + cluster.uuid = "cluster-1" + cluster.status = Cluster.STATUS_DEGRADED + mock_db.get_cluster_by_id.return_value = cluster + mock_db.get_job_tasks.return_value = [] + mock_db.kv_store = MagicMock() + + with patch.object(ctl, 'db', mock_db), \ + patch('simplyblock_core.controllers.migration_controller.tasks_controller') as tc: + tc.get_active_node_mig_task.return_value = None + tc.add_batch_mig_task.return_value = "task-uuid" + result = ctl.start_batch_migration("group-uuid") + + assert result == "group-uuid" + tc.add_batch_mig_task.assert_called_once() From 0025a57c6703b8950e0ec92a2ff079daf28bd09b Mon Sep 17 00:00:00 2001 From: ebrahim savari <106956085+EbiRider@users.noreply.github.com> Date: Wed, 2 Sep 2026 17:05:58 +0330 Subject: [PATCH 5/7] fix an issue where batch migration get stuck on cleanup --- .../services/tasks_runner_lvol_migration.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index b3f03d669c..a88dab07c7 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -3762,6 +3762,18 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src # intermediates_done signalled — wait for batch_result. group = db.get_migration_group_by_id(group_id) + if group.batch_result is None and group.phase == LVolMigrationGroup.PHASE_CLEANUP_TARGET: + # A sibling exhausted its retry budget and forced the group into + # cleanup without ever setting batch_result (only the normal + # INTERMEDIATE barrier path sets it) -- notice the forced phase + # directly instead of polling batch_result forever. + migration.phase = LVolMigration.PHASE_CLEANUP_TARGET + migration.transfer_context = {} + migration.write_to_db(db.kv_store) + migration_events.migration_phase_changed(migration) + return _group_worker_phase_dispatch( + task, migration, LVolMigration.PHASE_CLEANUP_TARGET, + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) if group.batch_result is True: lvol = db.get_lvol_by_id(migration.lvol_id) migration.phase = LVolMigration.PHASE_CLEANUP_SOURCE From 20e94493305490bf5e86e1bcb28188ea7c8876fa Mon Sep 17 00:00:00 2001 From: ebrahim savari <106956085+EbiRider@users.noreply.github.com> Date: Thu, 3 Sep 2026 22:47:10 +0330 Subject: [PATCH 6/7] return conflict 409 if chosing secondary when migrating from seconadry --- .../api/v2/cluster/subsystem/migration.py | 20 +++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/simplyblock_web/api/v2/cluster/subsystem/migration.py b/simplyblock_web/api/v2/cluster/subsystem/migration.py index a767d3b46c..85b5bcd75f 100644 --- a/simplyblock_web/api/v2/cluster/subsystem/migration.py +++ b/simplyblock_web/api/v2/cluster/subsystem/migration.py @@ -96,7 +96,15 @@ def create_migration( ctrl_loss_tmo=parameters.ctrl_loss_tmo, host_nqn=parameters.host_nqn, ) - except (ValueError, MigrationConflictError, PreconditionError, RuntimeError) as e: + except (MigrationConflictError, PreconditionError) as e: + # Conflicting/not-yet-satisfiable state (e.g. a migration already + # active for this subsystem, or -- for a fallback-source migration -- + # the chosen target is the node currently serving as the fallback + # source itself) -- matches the 409 convention used for the same + # shape of error elsewhere in v2 (storage_node shutdown, pool/volume + # already-exists, in-flight replication cutover). + raise HTTPException(409, str(e)) + except (ValueError, RuntimeError) as e: raise HTTPException(400, str(e)) def get_full(id): @@ -149,7 +157,15 @@ def continue_migration(migration: SubsystemMigration, parameters: _ContinueParam max_retries=parameters.max_retries, deadline_seconds=parameters.deadline_seconds, ) - except (ValueError, MigrationConflictError, PreconditionError, RuntimeError) as e: + except (MigrationConflictError, PreconditionError) as e: + # Conflicting/not-yet-satisfiable state (e.g. a migration already + # active for this subsystem, or -- for a fallback-source migration -- + # the chosen target is the node currently serving as the fallback + # source itself) -- matches the 409 convention used for the same + # shape of error elsewhere in v2 (storage_node shutdown, pool/volume + # already-exists, in-flight replication cutover). + raise HTTPException(409, str(e)) + except (ValueError, RuntimeError) as e: raise HTTPException(400, str(e)) return {"migration_id": result_id} From 31a10f0a50b2acc4906fd1f55eda7743110146fe Mon Sep 17 00:00:00 2001 From: ebrahim savari <106956085+EbiRider@users.noreply.github.com> Date: Fri, 4 Sep 2026 23:46:15 +0330 Subject: [PATCH 7/7] fixing an issue where the batch group doesn't enter cleanup properly --- .../services/tasks_runner_lvol_migration.py | 80 +++++++++++++------ 1 file changed, 55 insertions(+), 25 deletions(-) diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index 75cbd4da8b..2c7699c462 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -3008,21 +3008,37 @@ def task_runner(task): # against: the primary when reachable, otherwise the online replica # pinned once at create_migration() time as active_source_node_id. Every # data-plane call below must use src_node/src_rpc, never primary_src_node. + # Group workers must never give up through the plain (non-group-aware) + # _budget_suspend below: it only marks this one migration record + # CLEANUP_TARGET, never the shared group.phase, so the orchestrator's + # barrier (waiting for every member's snap_copy_done/intermediates_done) + # never learns this worker is gone and waits for it forever. Route through + # _group_worker_budget_suspend instead whenever this migration belongs to + # a batch group, so one member failing here fails the whole group instead + # of hanging it (discovered 2026-09-04: a fallback migration whose active + # source node went non-online mid-INTERMEDIATE left all 3 workers silently + # CLEANUP_TARGET'd while the group orchestrator polled "waiting for 3 + # workers" forever, past the test's own 10-minute timeout). + def _node_lookup_suspend(error_msg): + if migration.migration_group_id: + return _group_worker_budget_suspend(task, migration, migration.migration_group_id, error_msg) + return _budget_suspend(task, migration, migration_id, error_msg) + try: primary_src_node = db.get_storage_node_by_id(migration.source_node_id) except KeyError: - return _budget_suspend(task, migration, migration_id, "source node not found") + return _node_lookup_suspend("source node not found") try: src_node = db.get_storage_node_by_id( migration.active_source_node_id or migration.source_node_id) except KeyError: - return _budget_suspend(task, migration, migration_id, "active source node not found") + return _node_lookup_suspend("active source node not found") try: tgt_node = db.get_storage_node_by_id(migration.target_node_id) except KeyError: - return _budget_suspend(task, migration, migration_id, "target node not found") + return _node_lookup_suspend("target node not found") # Cleanup phases proceed regardless of node status: deletes go through LVS # leadership, so a downed node doesn't block the cleanup path. @@ -3030,9 +3046,7 @@ def task_runner(task): LVolMigration.PHASE_CLEANUP_TARGET, LVolMigration.PHASE_CLEANUP_SOURCE) if not _is_cleanup_phase: if src_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): - return _budget_suspend( - task, migration, migration_id, - f"source node not online (status={src_node.status})") + return _node_lookup_suspend(f"source node not online (status={src_node.status})") if tgt_node.status != StorageNode.STATUS_ONLINE: if (migration.phase in (LVolMigration.PHASE_SNAP_COPY, @@ -3053,6 +3067,8 @@ def task_runner(task): migration.write_to_db(db.kv_store) task.write_to_db(db.kv_store) migration_events.migration_phase_changed(migration) + if migration.migration_group_id: + _fail_group_from_worker(migration, migration.migration_group_id, migration.error_message) return False if not _is_cleanup_phase: # cleanup phases are exempt: deletes go through LVS leadership; @@ -3572,6 +3588,38 @@ def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, return True, False, None +def _fail_group_from_worker(migration, group_id, error_msg): + """Force the whole group into CLEANUP_TARGET because one member just gave + up (retry budget exhausted, or an unconditional hard-fail like the target + going offline mid-transfer). + + A worker that stops here without this will never signal its barrier + (snap_copy_done / intermediates_done / cleanup_source_done) again -- the + orchestrator's barrier check has no way to distinguish "still working" from + "silently gone", so it waits forever instead of failing the group. Every + place a group worker can reach a terminal state outside its own normal + phase progression must call this. + """ + try: + group = db.get_migration_group_by_id(group_id) + if group.phase not in (LVolMigrationGroup.PHASE_CLEANUP_TARGET, + LVolMigrationGroup.PHASE_CLEANUP_SOURCE, + LVolMigrationGroup.PHASE_COMPLETED): + group.phase = LVolMigrationGroup.PHASE_CLEANUP_TARGET + group.error_message = ( + f"worker {migration.uuid[:8]} (lvol={migration.lvol_id}) failed: {error_msg}") + group.write_to_db(db.kv_store) + logger.error( + f"Group {group_id[:8]}: failing whole group — worker " + f"{migration.uuid[:8]} failed: {error_msg}") + except KeyError: + # Group may already be removed/cleaned up by another workflow. + # We keep worker cleanup flow idempotent by not re-raising. + logger.warning( + f"Group {group_id[:8]} not found while propagating worker " + f"{migration.uuid[:8]} failure; continuing.") + + def _group_worker_budget_suspend(task, migration, group_id, error_msg): """Charge retry budget for a group worker; fail the WHOLE GROUP when this worker's own budget is exhausted. @@ -3602,25 +3650,7 @@ def _group_worker_budget_suspend(task, migration, group_id, error_msg): # This worker will never signal done to its barrier now -- fail the # whole group rather than let siblings (and the orchestrator) wait # on it forever. - try: - group = db.get_migration_group_by_id(group_id) - if group.phase not in (LVolMigrationGroup.PHASE_CLEANUP_TARGET, - LVolMigrationGroup.PHASE_CLEANUP_SOURCE, - LVolMigrationGroup.PHASE_COMPLETED): - group.phase = LVolMigrationGroup.PHASE_CLEANUP_TARGET - group.error_message = ( - f"worker {migration.uuid[:8]} (lvol={migration.lvol_id}) " - f"exceeded max retries: {error_msg}") - group.write_to_db(db.kv_store) - logger.error( - f"Group {group_id[:8]}: failing whole group — worker " - f"{migration.uuid[:8]} exhausted its retry budget") - except KeyError: - # Group may already be removed/cleaned up by another workflow. - # We keep worker cleanup flow idempotent by not re-raising. - logger.warning( - f"Group {group_id[:8]} not found while propagating worker " - f"{migration.uuid[:8]} retry-budget exhaustion; continuing.") + _fail_group_from_worker(migration, group_id, error_msg) return False return _suspend_task(task, migration, error_msg)