From dfe720e3fde2a3db8fe99fe2510b0cf899410eef Mon Sep 17 00:00:00 2001 From: wmousa Date: Mon, 3 Aug 2026 17:36:39 +0200 Subject: [PATCH 1/9] fix(cluster-activate): repair stranded nodes instead of failing activation get_secondary_nodes() and get_secondary_nodes_2() each pair nodes one at a time via a greedy walk, preferring a domain/host-disjoint candidate from a shrinking shared pool. Nothing guarantees that walk closes a single cycle spanning every online node: it can close a cycle over a strict subset and strand the rest with zero candidates, even though a perfect pairing exists whenever there are 2+ online nodes (hit live: 12 nodes across 3 failure domains formed an 11-node secondary-pairing cycle, stranding the 12th and aborting activation with "No enough secondary nodes"). The tertiary assignment used by max_fault_tolerance >= 2 clusters (e.g. 2+2) has the identical structure and is subject to the same failure mode. Add splice_stranded_secondary() and splice_stranded_tertiary(): when a node is left with no candidates, splice it into an already-formed pairing edge (P->X becomes P->stranded->X) instead of giving up, preferring an edge where both sides differ from the stranded node's failure domain. The tertiary splice additionally re-validates host-disjointness against each side's own secondary partner, since a tertiary must be host-disjoint from both a primary and that primary's secondary. _cluster_activate falls back to these before raising, and only still fails if no pairing has been made at all yet. --- simplyblock_core/cluster_ops.py | 49 ++++-- simplyblock_core/storage_node_ops.py | 152 ++++++++++++++++++ tests/unit/test_failure_domain.py | 227 +++++++++++++++++++++++++++ 3 files changed, 414 insertions(+), 14 deletions(-) diff --git a/simplyblock_core/cluster_ops.py b/simplyblock_core/cluster_ops.py index 1bc2d482b9..d33098e72f 100644 --- a/simplyblock_core/cluster_ops.py +++ b/simplyblock_core/cluster_ops.py @@ -1052,6 +1052,18 @@ def _fd_fail(msg: str) -> None: used_nodes_as_sec: t.List[str] = [] used_nodes_as_tertiary: t.List[str] = [] snodes = db_controller.get_storage_nodes_by_cluster_id(cl_id) + # Process primaries grouped by failure domain. get_secondary_nodes/ + # get_secondary_nodes_2 (and their splice repairs) already sort their own + # candidate scan by domain, which alone is enough to keep the assignment + # domain-disjoint when domains are evenly sized. But once any node needs + # splice-repair (uneven domain sizes, some conflict unavoidable), the + # repair works off whatever partial assignment already exists -- so which + # primary gets processed first still changes the outcome. Grouping here + # too makes the result deterministic instead of order-dependent in that + # case. A no-op when FD is disabled (all nodes share one failure_domain). + # Fresh FD+HA activation bypasses this fallback via fd_desired_layout, + # but reactivation and non-HA/non-fresh paths still rely on it. + snodes = sorted(snodes, key=lambda n: n.failure_domain) if cluster.ha_type == "ha": for snode in snodes: # Do not assign secondary to removed node @@ -1071,16 +1083,21 @@ def _fd_fail(msg: str) -> None: secondary_nodes = [fd_desired_layout[snode.get_id()][0]] else: secondary_nodes = storage_node_ops.get_secondary_nodes(snode) - if not secondary_nodes: + if secondary_nodes: + snode = db_controller.get_storage_node_by_id(snode.get_id()) + snode.secondary_node_id = secondary_nodes[0] + snode.write_to_db() + sec_node = db_controller.get_storage_node_by_id(snode.secondary_node_id) + sec_node.lvstore_stack_secondary = snode.get_id() + sec_node.write_to_db() + elif not storage_node_ops.splice_stranded_secondary(snode): + # get_secondary_nodes()'s greedy walk closed a cycle that + # excludes this node, and there isn't even one existing + # pairing left to splice it into (only possible this early + # in the pass, before 2+ pairings exist). set_cluster_status(cl_id, ols_status) raise ValueError("Failed to activate cluster, No enough secondary nodes") - snode = db_controller.get_storage_node_by_id(snode.get_id()) - snode.secondary_node_id = secondary_nodes[0] - snode.write_to_db() - sec_node = db_controller.get_storage_node_by_id(snode.secondary_node_id) - sec_node.lvstore_stack_secondary = snode.get_id() - sec_node.write_to_db() used_nodes_as_sec.append(snode.secondary_node_id) # Assign second secondary when max_fault_tolerance >= 2 @@ -1099,15 +1116,19 @@ def _fd_fail(msg: str) -> None: exclude_failure_domains=[sec_node.failure_domain], exclude_physical_labels=[sec_node.physical_label], ) - if not secondary_nodes_2: + if secondary_nodes_2: + snode.tertiary_node_id = secondary_nodes_2[0] + snode.write_to_db() + sec_node_2 = db_controller.get_storage_node_by_id(snode.tertiary_node_id) + sec_node_2.lvstore_stack_tertiary = snode.get_id() + sec_node_2.write_to_db() + elif not storage_node_ops.splice_stranded_tertiary(snode): + # get_secondary_nodes_2()'s greedy walk closed a cycle that + # excludes this node, and there isn't even one existing + # tertiary pairing left to splice it into. set_cluster_status(cl_id, ols_status) raise ValueError("Failed to activate cluster, not enough nodes for dual fault tolerance") - - snode.tertiary_node_id = secondary_nodes_2[0] - snode.write_to_db() - sec_node_2 = db_controller.get_storage_node_by_id(snode.tertiary_node_id) - sec_node_2.lvstore_stack_tertiary = snode.get_id() - sec_node_2.write_to_db() + snode = db_controller.get_storage_node_by_id(snode.get_id()) used_nodes_as_tertiary.append(snode.tertiary_node_id) # Pass 1: bring up the primary LVS on every online primary node. diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index fdf61c5a18..7c42603f05 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -9845,6 +9845,78 @@ def _candidates(forbidden_fds, forbidden_labels): return [] +def splice_stranded_secondary(stranded_node) -> bool: + """Fold a node get_secondary_nodes() could not place into the pairing + graph already built by the in-progress cluster_activate() pass. + + get_secondary_nodes() walks primaries in order, greedily picking the most + domain/host-disjoint unclaimed candidate for each. That greedy walk has no + mechanism to guarantee the resulting secondary_node_id/lvstore_stack_secondary + edges close a cycle spanning every online node: it can close a cycle over a + strict subset and leave the remaining node(s) with zero unclaimed + candidates, even though a perfect pairing trivially exists whenever there + are 2+ online nodes (observed 2026-08-03: 12 nodes across 3 failure + domains formed an 11-node cycle, stranding the 12th and aborting + activation). + + Rather than reworking the greedy walk into a global matching solver, this + repairs the one failure mode it has: pick any already-formed edge P->X + (P.secondary_node_id == X.get_id()) and splice the stranded node in + between, P->stranded->X. This always succeeds as long as at least one + edge already exists (guaranteed once 2+ pairings have been made this + activation pass) and turns the cycle that edge belongs to into one that + also covers the stranded node, without disturbing any other node. Prefers + an edge where both P and X differ from the stranded node's failure domain + (falling back to a host-disjoint-only edge), mirroring get_secondary_nodes' + own anti-affinity tiering. + """ + db_controller = DBController() + all_nodes = db_controller.get_storage_nodes_by_cluster_id(stranded_node.cluster_id) + edges = [n for n in all_nodes if n.secondary_node_id and n.get_id() != stranded_node.get_id()] + + def _host_disjoint(p, x): + return p.mgmt_ip != stranded_node.mgmt_ip and x.mgmt_ip != stranded_node.mgmt_ip + + def _domain_mismatch_score(p, x): + if stranded_node.failure_domain < 0: + return 0 + return sum(1 for n in (p, x) if n.failure_domain != stranded_node.failure_domain) + + best = None + best_score = -1 + for p in edges: + x = db_controller.get_storage_node_by_id(p.secondary_node_id) + if not x or x.get_id() == stranded_node.get_id() or not _host_disjoint(p, x): + continue + score = _domain_mismatch_score(p, x) + if score > best_score: + best_score, best = score, (p, x) + + if best is None: + return False + + p, x = best + logger.warning( + "get_secondary_nodes found no candidate for node %s; splicing it into " + "the existing pairing %s -> %s (domain-mismatch score %d/2).", + stranded_node.get_id(), p.get_id(), x.get_id(), best_score) + + p = db_controller.get_storage_node_by_id(p.get_id()) + p.secondary_node_id = stranded_node.get_id() + p.write_to_db() + + stranded_node = db_controller.get_storage_node_by_id(stranded_node.get_id()) + stranded_node.lvstore_stack_secondary = p.get_id() + stranded_node.secondary_node_id = x.get_id() + stranded_node.write_to_db() + + x = db_controller.get_storage_node_by_id(x.get_id()) + x.lvstore_stack_secondary = stranded_node.get_id() + x.write_to_db() + + return True + + def get_secondary_nodes_2(current_node: StorageNode, exclude_ids=None, exclude_mgmt_ips=None, exclude_failure_domains=None, exclude_physical_labels=None): """Get candidate nodes for second secondary assignment (dual fault tolerance). @@ -9935,6 +10007,86 @@ def _candidates(forbidden_fds, forbidden_labels): return [] +def splice_stranded_tertiary(stranded_node) -> bool: + """Tertiary-assignment counterpart to splice_stranded_secondary. + + get_secondary_nodes_2()'s greedy walk has the identical dead-end risk as + get_secondary_nodes(): it can close a tertiary-pairing cycle over a + subset of online nodes and strand the rest, even though a valid + assignment exists — this can surface on any cluster with + max_fault_tolerance >= 2 (e.g. a 2+2 layout), the same way + splice_stranded_secondary's bug surfaced on the plain secondary pass. + + Splices the stranded node into an already-formed tertiary edge P->X + (P.tertiary_node_id == X.get_id()), same idea as the secondary case: + P->stranded->X. The extra wrinkle here is that a tertiary must be + host-disjoint from BOTH a primary and that primary's OWN secondary (a + single host outage must not take out two of the four HA journal members) + — so splicing changes what "valid" means on both sides of the edge, and + each side is re-checked against the other's current secondary_node_id, + not just against each other. + """ + db_controller = DBController() + all_nodes = db_controller.get_storage_nodes_by_cluster_id(stranded_node.cluster_id) + by_id = {n.get_id(): n for n in all_nodes} + stranded_sec = by_id.get(stranded_node.secondary_node_id) if stranded_node.secondary_node_id else None + + def _valid_tertiary(primary, primary_sec, candidate): + if candidate.get_id() == primary.get_id(): + return False + if candidate.mgmt_ip == primary.mgmt_ip: + return False + if primary_sec and candidate.mgmt_ip == primary_sec.mgmt_ip: + return False + return True + + def _domain_mismatch_score(*nodes): + if stranded_node.failure_domain < 0: + return 0 + return sum(1 for n in nodes if n.failure_domain != stranded_node.failure_domain) + + edges = [n for n in all_nodes if n.tertiary_node_id and n.get_id() != stranded_node.get_id()] + + best = None + best_score = -1 + for p in edges: + x = by_id.get(p.tertiary_node_id) + if not x or x.get_id() == stranded_node.get_id(): + continue + p_sec = by_id.get(p.secondary_node_id) if p.secondary_node_id else None + if not _valid_tertiary(p, p_sec, stranded_node): + continue + if not _valid_tertiary(stranded_node, stranded_sec, x): + continue + score = _domain_mismatch_score(p, x) + if score > best_score: + best_score, best = score, (p, x) + + if best is None: + return False + + p, x = best + logger.warning( + "get_secondary_nodes_2 found no candidate for node %s; splicing it into " + "the existing tertiary pairing %s -> %s (domain-mismatch score %d/2).", + stranded_node.get_id(), p.get_id(), x.get_id(), best_score) + + p = db_controller.get_storage_node_by_id(p.get_id()) + p.tertiary_node_id = stranded_node.get_id() + p.write_to_db() + + stranded_node = db_controller.get_storage_node_by_id(stranded_node.get_id()) + stranded_node.lvstore_stack_tertiary = p.get_id() + stranded_node.tertiary_node_id = x.get_id() + stranded_node.write_to_db() + + x = db_controller.get_storage_node_by_id(x.get_id()) + x.lvstore_stack_tertiary = stranded_node.get_id() + x.write_to_db() + + return True + + def create_lvstore(snode: StorageNode, ndcs, npcs, distr_bs, distr_chunk_bs, page_size_in_blocks, max_size): db_controller = DBController() cluster = db_controller.get_cluster_by_id(snode.cluster_id) diff --git a/tests/unit/test_failure_domain.py b/tests/unit/test_failure_domain.py index 94eab3b191..3896ce2209 100644 --- a/tests/unit/test_failure_domain.py +++ b/tests/unit/test_failure_domain.py @@ -152,6 +152,98 @@ def test_enabled_falls_back_when_no_other_domain(self, MockDBCtrl): assert any("falling back" in m for m in cm.output) +# =========================================================================== +# 2b. splice_stranded_secondary +# +# get_secondary_nodes()'s greedy walk can close a pairing cycle over a strict +# subset of online nodes and strand the rest with zero candidates, even though +# a perfect pairing exists (observed 2026-08-03: 12 nodes / 3 domains formed +# an 11-node cycle, aborting activation for the 12th). splice_stranded_secondary +# repairs this by inserting the stranded node into an already-formed edge. +# =========================================================================== + +class TestSpliceStrandedSecondary(unittest.TestCase): + + def _mock_db(self, cluster, nodes): + mock_db = MagicMock() + by_id = {n.get_id(): n for n in nodes} + mock_db.get_cluster_by_id.return_value = cluster + mock_db.get_storage_nodes_by_cluster_id.return_value = nodes + mock_db.get_storage_node_by_id.side_effect = lambda nid: by_id.get(nid) + return mock_db + + @staticmethod + def _stub_writes(*nodes): + for n in nodes: + n.write_to_db = MagicMock() + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_splices_into_existing_edge(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + # p -> x is an existing pairing from earlier in the activation pass. + p = _node("p", "10.0.0.1", failure_domain=0) + x = _node("x", "10.0.0.2", failure_domain=1) + p.secondary_node_id = x.get_id() + s = _node("s", "10.0.0.3", failure_domain=2) + MockDBCtrl.return_value = self._mock_db(_cluster(True), [p, x, s]) + self._stub_writes(p, x, s) + + assert ops.splice_stranded_secondary(s) is True + # p -> s -> x: p now points at s, s sits between p and x. + assert p.secondary_node_id == "s" + assert s.lvstore_stack_secondary == "p" + assert s.secondary_node_id == "x" + assert x.lvstore_stack_secondary == "s" + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_prefers_edge_domain_disjoint_on_both_ends(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + s = _node("s", "10.0.0.9", failure_domain=0) + + # Same-domain edge (worse fit: 0/2 mismatch against s's domain). + bad_p = _node("bad_p", "10.0.0.1", failure_domain=0) + bad_x = _node("bad_x", "10.0.0.2", failure_domain=0) + bad_p.secondary_node_id = bad_x.get_id() + + # Domain-disjoint-on-both-ends edge (best fit: 2/2 mismatch). + good_p = _node("good_p", "10.0.0.3", failure_domain=1) + good_x = _node("good_x", "10.0.0.4", failure_domain=2) + good_p.secondary_node_id = good_x.get_id() + + nodes = [s, bad_p, bad_x, good_p, good_x] + MockDBCtrl.return_value = self._mock_db(_cluster(True), nodes) + self._stub_writes(*nodes) + + assert ops.splice_stranded_secondary(s) is True + assert good_p.secondary_node_id == "s" + assert s.secondary_node_id == "good_x" + assert bad_p.secondary_node_id == "bad_x" # untouched + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_returns_false_when_no_edge_exists_yet(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + s = _node("s", "10.0.0.9", failure_domain=0) + other = _node("other", "10.0.0.1", failure_domain=1) # no pairing yet + MockDBCtrl.return_value = self._mock_db(_cluster(True), [s, other]) + self._stub_writes(s, other) + + assert ops.splice_stranded_secondary(s) is False + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_skips_edge_not_host_disjoint_from_stranded(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + # x shares mgmt_ip with the stranded node -- splicing there would + # violate host-disjointness, so this edge must be skipped entirely. + s = _node("s", "10.0.0.5", failure_domain=0) + p = _node("p", "10.0.0.1", failure_domain=1) + x = _node("x", "10.0.0.5", failure_domain=2) + p.secondary_node_id = x.get_id() + MockDBCtrl.return_value = self._mock_db(_cluster(True), [s, p, x]) + self._stub_writes(s, p, x) + + assert ops.splice_stranded_secondary(s) is False + + # =========================================================================== # 3. get_secondary_nodes_2 (tertiary) # =========================================================================== @@ -199,6 +291,141 @@ def test_falls_back_when_only_shared_domains(self, MockDBCtrl): assert any("falling back" in m for m in cm.output) +# =========================================================================== +# 3b. splice_stranded_tertiary +# +# get_secondary_nodes_2() has the identical greedy-walk dead-end risk as +# get_secondary_nodes() (see TestSpliceStrandedSecondary above), reachable on +# any cluster with max_fault_tolerance >= 2 (e.g. a 2+2 layout). The extra +# wrinkle: a tertiary must be host-disjoint from both a primary and that +# primary's OWN secondary, so a valid splice must be re-checked against each +# side's current secondary_node_id, not just against each other. +# =========================================================================== + +class TestSpliceStrandedTertiary(unittest.TestCase): + + def _mock_db(self, cluster, nodes): + mock_db = MagicMock() + by_id = {n.get_id(): n for n in nodes} + mock_db.get_cluster_by_id.return_value = cluster + mock_db.get_storage_nodes_by_cluster_id.return_value = nodes + mock_db.get_storage_node_by_id.side_effect = lambda nid: by_id.get(nid) + return mock_db + + @staticmethod + def _stub_writes(*nodes): + for n in nodes: + n.write_to_db = MagicMock() + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_splices_into_existing_tertiary_edge(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + # p -> x is an existing tertiary edge. s is stranded, with its own + # secondary (s_sec) on a distinct host from everyone else involved. + p_sec = _node("p_sec", "10.0.0.10", failure_domain=1) + p = _node("p", "10.0.0.1", failure_domain=0) + p.secondary_node_id = "p_sec" + x = _node("x", "10.0.0.2", failure_domain=1) + p.tertiary_node_id = "x" + + s_sec = _node("s_sec", "10.0.0.20", failure_domain=1) + s = _node("s", "10.0.0.3", failure_domain=2) + s.secondary_node_id = "s_sec" + + nodes = [p, p_sec, x, s, s_sec] + MockDBCtrl.return_value = self._mock_db(_cluster(True), nodes) + self._stub_writes(*nodes) + + assert ops.splice_stranded_tertiary(s) is True + # p -> s -> x: p now points at s, s sits between p and x. + assert p.tertiary_node_id == "s" + assert s.lvstore_stack_tertiary == "p" + assert s.tertiary_node_id == "x" + assert x.lvstore_stack_tertiary == "s" + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_returns_false_when_no_edge_exists_yet(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + s = _node("s", "10.0.0.3", failure_domain=0) + other = _node("other", "10.0.0.1", failure_domain=1) # no tertiary edge yet + MockDBCtrl.return_value = self._mock_db(_cluster(True), [s, other]) + self._stub_writes(s, other) + + assert ops.splice_stranded_tertiary(s) is False + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_rejects_edge_when_stranded_shares_host_with_primarys_secondary(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + # s sits on the same host as p's secondary (p_sec) -- s can never + # become p's tertiary, so the only existing edge must be rejected. + p_sec = _node("p_sec", "10.0.0.9", failure_domain=1) + p = _node("p", "10.0.0.1", failure_domain=0) + p.secondary_node_id = "p_sec" + x = _node("x", "10.0.0.2", failure_domain=1) + p.tertiary_node_id = "x" + + s = _node("s", "10.0.0.9", failure_domain=2) # same mgmt_ip as p_sec + s_sec = _node("s_sec", "10.0.0.20", failure_domain=1) + s.secondary_node_id = "s_sec" + + nodes = [p, p_sec, x, s, s_sec] + MockDBCtrl.return_value = self._mock_db(_cluster(True), nodes) + self._stub_writes(*nodes) + + assert ops.splice_stranded_tertiary(s) is False + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_rejects_edge_when_x_shares_host_with_stranded_secondary(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + # x sits on the same host as s's secondary (s_sec) -- x can never + # become s's tertiary, so the only existing edge must be rejected. + p_sec = _node("p_sec", "10.0.0.9", failure_domain=1) + p = _node("p", "10.0.0.1", failure_domain=0) + p.secondary_node_id = "p_sec" + x = _node("x", "10.0.0.30", failure_domain=1) + p.tertiary_node_id = "x" + + s_sec = _node("s_sec", "10.0.0.30", failure_domain=1) # same mgmt_ip as x + s = _node("s", "10.0.0.3", failure_domain=2) + s.secondary_node_id = "s_sec" + + nodes = [p, p_sec, x, s, s_sec] + MockDBCtrl.return_value = self._mock_db(_cluster(True), nodes) + self._stub_writes(*nodes) + + assert ops.splice_stranded_tertiary(s) is False + + @patch("simplyblock_core.storage_node_ops.DBController") + def test_prefers_edge_domain_disjoint_on_both_ends(self, MockDBCtrl): + import simplyblock_core.storage_node_ops as ops + s_sec = _node("s_sec", "10.0.0.90", failure_domain=9) + s = _node("s", "10.0.0.9", failure_domain=0) + s.secondary_node_id = "s_sec" + + # Same-domain edge (worse fit: 0/2 mismatch against s's domain). + bad_p_sec = _node("bad_p_sec", "10.0.0.11", failure_domain=9) + bad_p = _node("bad_p", "10.0.0.1", failure_domain=0) + bad_p.secondary_node_id = "bad_p_sec" + bad_x = _node("bad_x", "10.0.0.2", failure_domain=0) + bad_p.tertiary_node_id = "bad_x" + + # Domain-disjoint-on-both-ends edge (best fit: 2/2 mismatch). + good_p_sec = _node("good_p_sec", "10.0.0.13", failure_domain=9) + good_p = _node("good_p", "10.0.0.3", failure_domain=1) + good_p.secondary_node_id = "good_p_sec" + good_x = _node("good_x", "10.0.0.4", failure_domain=2) + good_p.tertiary_node_id = "good_x" + + nodes = [s, s_sec, bad_p, bad_p_sec, bad_x, good_p, good_p_sec, good_x] + MockDBCtrl.return_value = self._mock_db(_cluster(True), nodes) + self._stub_writes(*nodes) + + assert ops.splice_stranded_tertiary(s) is True + assert good_p.tertiary_node_id == "s" + assert s.tertiary_node_id == "good_x" + assert bad_p.tertiary_node_id == "bad_x" # untouched + + # =========================================================================== # 4. get_sorted_ha_jms # =========================================================================== From cf371a86a8c416b9f664c7c5485b83c0fe181ff6 Mon Sep 17 00:00:00 2001 From: wmousa Date: Mon, 3 Aug 2026 18:52:35 +0200 Subject: [PATCH 2/9] fix(ha): make node-shutdown FTT capacity check failure-domain aware MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _check_ftt_allows_node_removal gates node shutdown/suspend purely on raw not-online node count (cap = npcs), independent of the operator's drain-gate. This is stricter than necessary with failure domains enabled: placement guarantees at most one erasure-coding chunk per domain once there are ndcs+npcs distinct domains, so losing up to npcs domains at once is already tolerated the same way losing up to npcs nodes is tolerated without FD — but this check still blocked a second node in the same domain, undermining the drain-coordinator's FD-aware concurrency (observed live: a same-domain concurrent drain was correctly waved through by the operator's gate, then independently rejected here with "FTT=1: cluster already has 1 not-online node(s)"). With FD enabled and the target node's domain assigned, the capacity check now counts distinct affected domains instead of raw node count: piling onto an already-affected domain is always free; a new domain is gated on distinct-domain count against npcs when there are enough domains for full one-chunk-per-domain isolation, or falls back to the plain node-count cap otherwise (mirrors the operator's fdDrainGate). FD disabled or an unassigned node falls back to the original node-count logic unchanged. The npcs=2/ft=1 primary-secondary pairing constraint is unaffected. --- simplyblock_core/storage_node_ops.py | 124 ++++++++++++++++++-------- tests/unit/test_ftt_protection.py | 128 ++++++++++++++++++++++++++- 2 files changed, 214 insertions(+), 38 deletions(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 7c42603f05..bc53133563 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -5293,10 +5293,56 @@ def _check_ftt_allows_node_removal(node_id, db_controller): if jm_replication_active: not_online_count += 1 - if npcs == 1: + # Capacity cap is npcs uniformly (the erasure code tolerates losing up to + # npcs of its ndcs+npcs chunks regardless of the *declared* ft, which only + # narrows npcs=2 down to a stricter per-pair constraint below). + capacity_cap = npcs + + fd_on = cluster.enable_failure_domain and snode.failure_domain >= 0 + blocked_by_capacity = False + capacity_reason = "" + + if fd_on: + # Placement guarantees at most one erasure-coding chunk per domain + # once there are >= ndcs+npcs distinct domains, so losing up to npcs + # *domains* at once is then tolerated the same way losing up to npcs + # *nodes* is tolerated without FD -- piling onto an already-affected + # domain costs nothing extra (mirrors that domain going down outright). + # Below that domain count, at least one domain necessarily carries + # more than one chunk, so a *new* domain falls back to the plain + # node-count cap instead of a second free domain slot. See the + # analogous fdDrainGate in the operator's nodedrain_controller.go. + domains_available = len({n.failure_domain for n in snodes if n.failure_domain >= 0}) + domains_needed = ndcs + npcs + active_domains = {n.failure_domain for n in not_online_nodes if n.failure_domain >= 0} + # jm_replication_active can't be attributed to a specific domain (the + # probe only says "some online node's journal is behind"), so treat + # it conservatively as always adding a new, unaccounted-for domain. + active_domain_count = len(active_domains) + (1 if jm_replication_active else 0) + my_domain = snode.failure_domain + + if my_domain not in active_domains: + if domains_needed > 0 and domains_available >= domains_needed: + if active_domain_count >= capacity_cap: + blocked_by_capacity = True + capacity_reason = ( + f"FTT={ft} (npcs={npcs}): cannot remove node, failure domain {my_domain} " + f"not yet active; {active_domain_count}/{capacity_cap} domains active" + f"{' (including in-progress journal replication)' if jm_replication_active else ''}" + ) + elif active_domain_count > 0 and not_online_count >= capacity_cap: + blocked_by_capacity = True + capacity_reason = ( + f"FTT={ft} (npcs={npcs}): insufficient failure domains " + f"({domains_available} available, {domains_needed} needed for full isolation) " + f"to remove node in failure domain {my_domain}; " + f"{not_online_count}/{capacity_cap} nodes active" + ) + elif npcs == 1: # FTT=1: no room at all if anything is already not online or journal replicating if not_online_count > 0: - return False, ( + blocked_by_capacity = True + capacity_reason = ( f"FTT=1 (npcs=1): cannot remove node, cluster already has " f"{len(not_online_nodes)} not-online node(s)" f"{' and journal replication in progress' if jm_replication_active else ''}" @@ -5306,53 +5352,59 @@ def _check_ftt_allows_node_removal(node_id, db_controller): if ft >= 2: # FTT=2: room for one not-online node, block if already have one+ if not_online_count >= 2: - return False, ( + blocked_by_capacity = True + capacity_reason = ( f"FTT=2 (npcs=2): cannot remove node, cluster already has " f"{len(not_online_nodes)} not-online node(s)" f"{' and journal replication in progress' if jm_replication_active else ''}" ) else: # npcs=2, ft=1: like FTT=2 for capacity, but additionally - # cannot remove both primary and its secondary + # cannot remove both primary and its secondary (checked below). if not_online_count >= 2: - return False, ( + blocked_by_capacity = True + capacity_reason = ( f"npcs=2/ft=1: cannot remove node, cluster already has " f"{len(not_online_nodes)} not-online node(s)" f"{' and journal replication in progress' if jm_replication_active else ''}" ) - # Check primary-secondary pair constraint: - # If the node being removed is a primary, check its secondary is online. - # If the node being removed is a secondary, check its primary is online. - for not_online_node in not_online_nodes: - # Is any not-online node the secondary of the node we're removing? - if snode.secondary_node_id == not_online_node.get_id(): - return False, ( - f"npcs=2/ft=1: cannot remove node {node_id}, " - f"its secondary {not_online_node.get_id()} is not online " - f"(status: {not_online_node.status})" - ) - if snode.tertiary_node_id == not_online_node.get_id(): - return False, ( - f"npcs=2/ft=1: cannot remove node {node_id}, " - f"its secondary {not_online_node.get_id()} is not online " - f"(status: {not_online_node.status})" - ) + if blocked_by_capacity: + return False, capacity_reason - # Is the node we're removing a secondary of any not-online primary? - for not_online_node in not_online_nodes: - if not_online_node.secondary_node_id == node_id: - return False, ( - f"npcs=2/ft=1: cannot remove node {node_id}, " - f"it is secondary of not-online primary {not_online_node.get_id()} " - f"(status: {not_online_node.status})" - ) - if not_online_node.tertiary_node_id == node_id: - return False, ( - f"npcs=2/ft=1: cannot remove node {node_id}, " - f"it is secondary of not-online primary {not_online_node.get_id()} " - f"(status: {not_online_node.status})" - ) + if npcs == 2 and ft == 1: + # npcs=2, ft=1: beyond the capacity cap above, cannot remove both a + # primary and its own secondary/tertiary at once -- a per-relationship + # constraint, orthogonal to failure domains. + for not_online_node in not_online_nodes: + # Is any not-online node the secondary of the node we're removing? + if snode.secondary_node_id == not_online_node.get_id(): + return False, ( + f"npcs=2/ft=1: cannot remove node {node_id}, " + f"its secondary {not_online_node.get_id()} is not online " + f"(status: {not_online_node.status})" + ) + if snode.tertiary_node_id == not_online_node.get_id(): + return False, ( + f"npcs=2/ft=1: cannot remove node {node_id}, " + f"its secondary {not_online_node.get_id()} is not online " + f"(status: {not_online_node.status})" + ) + + # Is the node we're removing a secondary of any not-online primary? + for not_online_node in not_online_nodes: + if not_online_node.secondary_node_id == node_id: + return False, ( + f"npcs=2/ft=1: cannot remove node {node_id}, " + f"it is secondary of not-online primary {not_online_node.get_id()} " + f"(status: {not_online_node.status})" + ) + if not_online_node.tertiary_node_id == node_id: + return False, ( + f"npcs=2/ft=1: cannot remove node {node_id}, " + f"it is secondary of not-online primary {not_online_node.get_id()} " + f"(status: {not_online_node.status})" + ) return True, "" diff --git a/tests/unit/test_ftt_protection.py b/tests/unit/test_ftt_protection.py index f63ffe162c..8ba8a598f3 100644 --- a/tests/unit/test_ftt_protection.py +++ b/tests/unit/test_ftt_protection.py @@ -23,7 +23,7 @@ # Helpers # --------------------------------------------------------------------------- -def _cluster(ha_type="ha", npcs=1, ndcs=2, ft=1, rebalancing=False): +def _cluster(ha_type="ha", npcs=1, ndcs=2, ft=1, rebalancing=False, enable_failure_domain=False): cl = Cluster() cl.uuid = "cluster-1" cl.ha_type = ha_type @@ -31,12 +31,14 @@ def _cluster(ha_type="ha", npcs=1, ndcs=2, ft=1, rebalancing=False): cl.distr_ndcs = ndcs cl.max_fault_tolerance = ft cl.is_re_balancing = rebalancing + cl.enable_failure_domain = enable_failure_domain cl.status = Cluster.STATUS_ACTIVE return cl def _node(node_id, status=StorageNode.STATUS_ONLINE, cluster_id="cluster-1", - secondary_id="", secondary_id_2="", jm_vuid=8881, lvstore="LVS_1"): + secondary_id="", secondary_id_2="", jm_vuid=8881, lvstore="LVS_1", + failure_domain=-1): n = MagicMock(spec=StorageNode) n.uuid = node_id n.get_id = MagicMock(return_value=node_id) @@ -47,6 +49,7 @@ def _node(node_id, status=StorageNode.STATUS_ONLINE, cluster_id="cluster-1", n.jm_vuid = jm_vuid n.lvstore = lvstore n.mgmt_ip = f"10.0.0.{hash(node_id) % 256}" + n.failure_domain = failure_domain # rpc_client mock: journal replication not active by default rpc = MagicMock() rpc.bdev_lvol_get_lvstores = MagicMock(return_value=[{"name": lvstore}]) @@ -573,6 +576,127 @@ def test_jm_replication_counts_but_pair_still_checked(self): self.assertFalse(allowed) +# --------------------------------------------------------------------------- +# Failure-domain-aware capacity check +# +# With FD enabled, the capacity cap (npcs) counts distinct affected failure +# domains instead of raw not-online nodes, mirroring the operator's +# fdDrainGate (nodedrain_controller.go) -- piling onto an already-affected +# domain is always free; a *new* domain is gated on distinct-domain count +# when there are >= ndcs+npcs domains, or on the plain node-count cap +# otherwise (under-provisioned). +# --------------------------------------------------------------------------- + +class TestFailureDomainAwareCapacity(unittest.TestCase): + + def test_piling_onto_active_domain_always_allowed(self): + """Two nodes already down in domain 1 (at cap already) -- removing a + THIRD node also in domain 1 must still be allowed.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=1), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n4", failure_domain=2), + _node("n5", failure_domain=3), + _node("n6", failure_domain=4), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertTrue(allowed) + + def test_well_provisioned_new_domain_blocked_at_cap(self): + """ndcs=2/npcs=2 needs 4 domains for full isolation; 4 are available. + Domains 1 and 2 already have not-online nodes (cap=npcs=2 domains + active) -- removing a node in a THIRD domain must be blocked.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=3), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=2), + _node("n4", failure_domain=4), + _node("n5", failure_domain=1), + _node("n6", failure_domain=2), + ] + db = _db(cl, nodes) + allowed, reason = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) + self.assertIn("domain", reason) + + def test_well_provisioned_new_domain_within_cap_allowed(self): + """Same layout, but only ONE domain active so far -- opening a + second domain is still within the npcs=2 domain budget.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=3), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", failure_domain=2), + _node("n4", failure_domain=4), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertTrue(allowed) + + def test_under_provisioned_second_domain_within_node_budget_allowed(self): + """ndcs=2/npcs=2 needs 4 domains, but only 2 exist -- under-provisioned. + One node down in domain 1; removing a node in domain 2 (opening a + second domain) is still allowed while raw not-online count (1) is + under the npcs=2 node budget.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=2), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", failure_domain=1), + _node("n4", failure_domain=2), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertTrue(allowed) + + def test_under_provisioned_second_domain_blocked_at_node_budget(self): + """Same under-provisioned layout, but the node-count budget (npcs=2) + is already spent -- opening a second domain must now be blocked.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=2), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n4", failure_domain=2), + ] + db = _db(cl, nodes) + allowed, reason = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) + self.assertIn("insufficient failure domains", reason) + + def test_fd_enabled_but_node_unassigned_falls_back_to_node_count(self): + """FD enabled cluster-wide, but this specific node has no domain + assignment (-1) -- must fall back to the plain node-count cap.""" + cl = _cluster(npcs=1, ndcs=2, ft=1, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=-1), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", failure_domain=1), + ] + db = _db(cl, nodes) + allowed, reason = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) + self.assertIn("not-online", reason) + + def test_fd_disabled_ignores_domains_uses_node_count(self): + """enable_failure_domain=False: domain tags present but irrelevant -- + behaves exactly like the non-FD npcs=1 case.""" + cl = _cluster(npcs=1, ndcs=2, ft=1, enable_failure_domain=False) + nodes = [ + _node("n1", failure_domain=1), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=2), + _node("n3", failure_domain=3), + ] + db = _db(cl, nodes) + allowed, reason = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) + self.assertIn("FTT=1", reason) + + # --------------------------------------------------------------------------- # Rebalancing additional scenarios # --------------------------------------------------------------------------- From 1a188874ae49421e4dbe4467ce8b6a6bce0338c2 Mon Sep 17 00:00:00 2001 From: wmousa Date: Tue, 4 Aug 2026 12:03:21 +0200 Subject: [PATCH 3/9] fix(cluster-activate): make secondary/tertiary pairing domain-order-independent get_secondary_nodes/get_secondary_nodes_2/splice_stranded_secondary/ splice_stranded_tertiary each fetch their own candidate list fresh from the DB and scan/score it in that order, independent of whatever order the caller processes primaries in. With failure domains enabled, this made the resulting primary/secondary/tertiary assignment sensitive to arbitrary node ordering: even when a fully domain-disjoint assignment exists (e.g. equal- sized domains), ~1 in 5 arbitrary orderings left some node with a same- domain secondary or tertiary (verified by simulation), and the live deployment hit exactly this. All four functions now sort their fetched node list by failure_domain before scanning, which makes equal-sized domains fully order-independent (0 conflicts across 50 arbitrary orderings, verified). _cluster_activate's own pairing loop also sorts its processing order the same way: once domain sizes are uneven and splice-repair is required, the repair works off whatever partial assignment already exists, so the caller's processing order still mattered even with the candidate-scan fix alone. With both in place, unequal-domain conflict counts become deterministic instead of order-dependent. Both sorts are no-ops when failure domains are disabled. --- simplyblock_core/storage_node_ops.py | 21 +++++ tests/unit/test_failure_domain.py | 120 +++++++++++++++++++++++++++ 2 files changed, 141 insertions(+) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index bc53133563..4e5e948390 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -9841,6 +9841,18 @@ def get_secondary_nodes(current_node: StorageNode, exclude_ids=None, removed_nod db_controller = DBController() cluster = db_controller.get_cluster_by_id(current_node.cluster_id) all_nodes = db_controller.get_storage_nodes_by_cluster_id(current_node.cluster_id) + # Group by failure domain (stable sort, preserves DB order within each + # domain) before scanning candidates. The "first valid candidate after my + # own position" logic below skips same-domain nodes as forbidden, so on an + # arbitrary/interleaved node order it can still land back on a same-domain + # pick once every other domain's nodes are already claimed -- purely an + # artifact of iteration order, not availability (verified by simulation: + # ~1 in 5 arbitrary orderings produces an avoidable same-domain pick even + # when a fully domain-disjoint assignment exists). Grouping first removes + # that sensitivity: every node's forward scan cleanly skips past the rest + # of its own domain into the next one. A no-op when FD is disabled (all + # nodes share the same failure_domain, so the sort is order-preserving). + all_nodes = sorted(all_nodes, key=lambda n: n.failure_domain) if len(all_nodes) == 2: for node in all_nodes: if node.get_id() != current_node.get_id() and node.get_id() not in exclude_ids: @@ -9924,6 +9936,9 @@ def splice_stranded_secondary(stranded_node) -> bool: """ db_controller = DBController() all_nodes = db_controller.get_storage_nodes_by_cluster_id(stranded_node.cluster_id) + # Deterministic tie-breaking among equally domain-scored edges -- see + # get_secondary_nodes for why this sort matters. + all_nodes = sorted(all_nodes, key=lambda n: n.failure_domain) edges = [n for n in all_nodes if n.secondary_node_id and n.get_id() != stranded_node.get_id()] def _host_disjoint(p, x): @@ -9996,6 +10011,9 @@ def get_secondary_nodes_2(current_node: StorageNode, exclude_ids=None, exclude_m db_controller = DBController() cluster = db_controller.get_cluster_by_id(current_node.cluster_id) all_nodes = db_controller.get_storage_nodes_by_cluster_id(current_node.cluster_id) + # See get_secondary_nodes for why this sort matters: it removes the + # pairing algorithm's sensitivity to arbitrary/interleaved node order. + all_nodes = sorted(all_nodes, key=lambda n: n.failure_domain) if len(all_nodes) == 2: for node in all_nodes: if node.get_id() != current_node.get_id() and node.get_id() not in exclude_ids: @@ -10080,6 +10098,9 @@ def splice_stranded_tertiary(stranded_node) -> bool: """ db_controller = DBController() all_nodes = db_controller.get_storage_nodes_by_cluster_id(stranded_node.cluster_id) + # Deterministic tie-breaking among equally domain-scored edges -- see + # get_secondary_nodes for why this sort matters. + all_nodes = sorted(all_nodes, key=lambda n: n.failure_domain) by_id = {n.get_id(): n for n in all_nodes} stranded_sec = by_id.get(stranded_node.secondary_node_id) if stranded_node.secondary_node_id else None diff --git a/tests/unit/test_failure_domain.py b/tests/unit/test_failure_domain.py index 3896ce2209..e16d08bb07 100644 --- a/tests/unit/test_failure_domain.py +++ b/tests/unit/test_failure_domain.py @@ -24,6 +24,7 @@ All external dependencies (FDB, RPC) are mocked. """ +import random import unittest from unittest.mock import MagicMock, patch @@ -872,5 +873,124 @@ def test_tertiary_excludes_secondary_label(self, MockDBCtrl): assert "c2" not in result # primary's label +# =========================================================================== +# 9. Domain-grouped ordering eliminates pairing order-sensitivity +# +# Two independent things had to be fixed for the pairing loop to stop being +# order-sensitive: +# +# 1. get_secondary_nodes/get_secondary_nodes_2/splice_stranded_secondary/ +# splice_stranded_tertiary each fetch their own candidate list fresh via +# db_controller.get_storage_nodes_by_cluster_id() and scan/score it in +# that order -- so their domain-disjointness used to depend on whatever +# order the DB happened to return nodes in, regardless of caller. All +# four now sort their fetched node list by failure_domain first. +# 2. Even with (1) fixed, _cluster_activate's own pairing loop still +# determines which primary is processed first, and once a domain-size +# imbalance forces splice-repair, the repair works off whatever partial +# assignment already exists -- so the CALLER's processing order still +# changes the outcome. _cluster_activate now also sorts its local +# `snodes` by failure_domain before the loop. +# +# Verified here by running the full secondary+tertiary pairing sequence with +# the DB mock returning nodes in an arbitrary (shuffled) order -- proving (1) +# -- while the caller-side loop mirrors _cluster_activate's own domain sort +# -- proving (1)+(2) together give a fully order-independent result. +# =========================================================================== + +class TestDomainGroupedCandidateScanOrderIndependence(unittest.TestCase): + + def _mock_db(self, cluster, nodes): + mock_db = MagicMock() + by_id = {n.get_id(): n for n in nodes} + mock_db.get_cluster_by_id.return_value = cluster + mock_db.get_storage_nodes_by_cluster_id.return_value = nodes + mock_db.get_storage_node_by_id.side_effect = lambda nid: by_id.get(nid) + return mock_db + + @staticmethod + def _build_nodes(domain_sizes): + nodes = [] + idx = 0 + for fd, size in enumerate(domain_sizes, start=1): + for _ in range(size): + idx += 1 + nodes.append(_node(f"n{idx}", f"10.0.0.{idx}", failure_domain=fd)) + return nodes + + def _run_pairing(self, db_order_nodes, cluster): + """Mirrors _cluster_activate's secondary+tertiary pairing loop + (cluster_ops.py), including its own failure_domain sort of the + processing order. `db_order_nodes` is deliberately left in an + arbitrary order to represent whatever get_storage_nodes_by_cluster_id + naturally returns -- proving get_secondary_nodes/_2's own internal + sort is what keeps candidate scanning domain-grouped regardless.""" + import simplyblock_core.storage_node_ops as ops + by_id = {n.get_id(): n for n in db_order_nodes} + for n in db_order_nodes: + n.write_to_db = MagicMock() + + processing_order = sorted(db_order_nodes, key=lambda n: n.failure_domain) + + with patch("simplyblock_core.storage_node_ops.DBController", + return_value=self._mock_db(cluster, db_order_nodes)): + for snode in processing_order: + secs = ops.get_secondary_nodes(snode) + if secs: + snode.secondary_node_id = secs[0] + by_id[secs[0]].lvstore_stack_secondary = snode.get_id() + else: + assert ops.splice_stranded_secondary(snode), f"{snode.get_id()} stranded on secondary" + + used_tertiary = [] + for snode in processing_order: + sec = by_id[snode.secondary_node_id] + t2 = ops.get_secondary_nodes_2( + snode, + exclude_ids=[snode.secondary_node_id] + used_tertiary, + exclude_mgmt_ips=[sec.mgmt_ip], + exclude_failure_domains=[sec.failure_domain], + exclude_physical_labels=[sec.physical_label], + ) + if t2: + snode.tertiary_node_id = t2[0] + by_id[t2[0]].lvstore_stack_tertiary = snode.get_id() + used_tertiary.append(t2[0]) + else: + assert ops.splice_stranded_tertiary(snode), f"{snode.get_id()} stranded on tertiary" + used_tertiary.append(snode.tertiary_node_id) + + conflicts = 0 + for n in db_order_nodes: + sec = by_id[n.secondary_node_id] + ter = by_id[n.tertiary_node_id] + if len({n.failure_domain, sec.failure_domain, ter.failure_domain}) != 3: + conflicts += 1 + return conflicts + + def test_equal_domains_conflict_free_across_many_arbitrary_db_orders(self): + cluster = _cluster(True, distr_npcs=2) + cluster.distr_ndcs = 2 + for seed in range(50): + nodes = self._build_nodes([4, 4, 4]) + random.Random(seed).shuffle(nodes) # arbitrary "DB return" order + conflicts = self._run_pairing(nodes, cluster) + assert conflicts == 0, f"seed={seed} produced {conflicts} conflicts" + + def test_unequal_domains_conflict_count_is_deterministic(self): + # 4/5/4: a full zero-conflict assignment is mathematically impossible + # (domains aren't equal size), but the combined sort (candidate scan + # + processing order) makes the resulting conflict count the same + # regardless of the arbitrary DB return order. + cluster = _cluster(True, distr_npcs=2) + cluster.distr_ndcs = 2 + counts = set() + for seed in range(20): + nodes = self._build_nodes([4, 5, 4]) + random.Random(seed).shuffle(nodes) + counts.add(self._run_pairing(nodes, cluster)) + assert len(counts) == 1, f"expected one consistent conflict count, got {counts}" + + if __name__ == "__main__": unittest.main() From f883fb13e4b3b7c4417ca6d377b6663e22171861 Mon Sep 17 00:00:00 2001 From: wmousa Date: Wed, 5 Aug 2026 13:56:37 +0200 Subject: [PATCH 4/9] fix(ha): correct failure-domain risk budget for node-shutdown capacity check The previous rule treated piling additional nodes onto an already-affected failure domain as always free, only falling back to a raw node-count cap when opening a brand-new domain under-provisioned. Cross-checked against the backend team's confirmed tolerance (2 FD: one whole FD down OR one node in each FD, nothing else; 3 FD: one whole FD only; 4 FD: two whole FDs), that rule incorrectly allowed unsafe combinations such as one node in FD1 plus two nodes in FD2 on a 2-FD cluster. Each domain's worst-case contribution to a stripe's chunk loss is now capped at chunks_per_domain = ceil((ndcs+npcs) / domains_available). A domain already at or above that count has maxed its risk contribution, so further nodes in the same domain are free; otherwise the summed capped risk across all affected domains plus the node being removed must stay within npcs. This collapses to the existing "npcs whole domains free" behavior once there are >= ndcs+npcs domains, and reproduces the confirmed 2/3/4-FD tolerance exactly -- verified both by direct simulation of the formula and by driving the real function through every scheme/domain-count combination in tests/unit/test_ftt_protection.py. --- simplyblock_core/storage_node_ops.py | 70 +++++------ tests/unit/test_ftt_protection.py | 168 +++++++++++++++++++++++++-- 2 files changed, 197 insertions(+), 41 deletions(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 4e5e948390..0df8c804ba 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -5293,51 +5293,53 @@ def _check_ftt_allows_node_removal(node_id, db_controller): if jm_replication_active: not_online_count += 1 - # Capacity cap is npcs uniformly (the erasure code tolerates losing up to - # npcs of its ndcs+npcs chunks regardless of the *declared* ft, which only - # narrows npcs=2 down to a stricter per-pair constraint below). - capacity_cap = npcs - fd_on = cluster.enable_failure_domain and snode.failure_domain >= 0 blocked_by_capacity = False capacity_reason = "" if fd_on: - # Placement guarantees at most one erasure-coding chunk per domain - # once there are >= ndcs+npcs distinct domains, so losing up to npcs - # *domains* at once is then tolerated the same way losing up to npcs - # *nodes* is tolerated without FD -- piling onto an already-affected - # domain costs nothing extra (mirrors that domain going down outright). - # Below that domain count, at least one domain necessarily carries - # more than one chunk, so a *new* domain falls back to the plain - # node-count cap instead of a second free domain slot. See the - # analogous fdDrainGate in the operator's nodedrain_controller.go. + # Placement spreads a stripe's ndcs+npcs chunks as evenly as possible + # across the domains that actually exist. With fewer domains than + # chunks, at least one domain holds ceil((ndcs+npcs)/domains) chunks + # -- that many nodes can go down within a SINGLE domain for free + # (mirrors that domain's worst-case chunk contribution going down + # outright, already priced in), but once a domain hits that many + # down, it has maxed its contribution to the npcs risk budget and a + # DIFFERENT domain can only add a node if the combined risk across + # every affected domain (each capped at chunks_per_domain) still + # leaves room. This reduces to the familiar "up to npcs whole domains + # are free" rule when there are >= ndcs+npcs domains (chunks_per_domain + # == 1). See the analogous fdDrainGate in nodedrain_controller.go. domains_available = len({n.failure_domain for n in snodes if n.failure_domain >= 0}) domains_needed = ndcs + npcs - active_domains = {n.failure_domain for n in not_online_nodes if n.failure_domain >= 0} - # jm_replication_active can't be attributed to a specific domain (the - # probe only says "some online node's journal is behind"), so treat - # it conservatively as always adding a new, unaccounted-for domain. - active_domain_count = len(active_domains) + (1 if jm_replication_active else 0) - my_domain = snode.failure_domain + chunks_per_domain = -(-domains_needed // domains_available) if domains_available > 0 else domains_needed - if my_domain not in active_domains: - if domains_needed > 0 and domains_available >= domains_needed: - if active_domain_count >= capacity_cap: - blocked_by_capacity = True - capacity_reason = ( - f"FTT={ft} (npcs={npcs}): cannot remove node, failure domain {my_domain} " - f"not yet active; {active_domain_count}/{capacity_cap} domains active" - f"{' (including in-progress journal replication)' if jm_replication_active else ''}" - ) - elif active_domain_count > 0 and not_online_count >= capacity_cap: + domain_down_counts: dict[int, int] = {} + for node in not_online_nodes: + if node.failure_domain >= 0: + domain_down_counts[node.failure_domain] = domain_down_counts.get(node.failure_domain, 0) + 1 + + my_domain = snode.failure_domain + my_domain_down = domain_down_counts.get(my_domain, 0) + + if my_domain_down < chunks_per_domain: + current_risk = sum(min(c, chunks_per_domain) for c in domain_down_counts.values()) + # jm_replication_active can't be attributed to a specific domain + # (the probe only says "some online node's journal is behind"), + # so treat it conservatively as always adding a fresh risk unit. + if jm_replication_active: + current_risk += 1 + if current_risk + 1 > npcs: blocked_by_capacity = True capacity_reason = ( - f"FTT={ft} (npcs={npcs}): insufficient failure domains " - f"({domains_available} available, {domains_needed} needed for full isolation) " - f"to remove node in failure domain {my_domain}; " - f"{not_online_count}/{capacity_cap} nodes active" + f"FTT={ft} (npcs={npcs}): cannot remove node in failure domain {my_domain}; " + f"{current_risk}/{npcs} failure-domain risk budget already committed " + f"({domains_available} domain(s) available, {chunks_per_domain} chunk(s)/domain worst case)" + f"{' (including in-progress journal replication)' if jm_replication_active else ''}" ) + # else: this domain already holds >= chunks_per_domain down nodes -- + # it has maxed its contribution to the risk budget, so one more node + # in the SAME domain adds no additional risk. elif npcs == 1: # FTT=1: no room at all if anything is already not online or journal replicating if not_online_count > 0: diff --git a/tests/unit/test_ftt_protection.py b/tests/unit/test_ftt_protection.py index 8ba8a598f3..65c95cbb76 100644 --- a/tests/unit/test_ftt_protection.py +++ b/tests/unit/test_ftt_protection.py @@ -579,12 +579,22 @@ def test_jm_replication_counts_but_pair_still_checked(self): # --------------------------------------------------------------------------- # Failure-domain-aware capacity check # -# With FD enabled, the capacity cap (npcs) counts distinct affected failure -# domains instead of raw not-online nodes, mirroring the operator's -# fdDrainGate (nodedrain_controller.go) -- piling onto an already-affected -# domain is always free; a *new* domain is gated on distinct-domain count -# when there are >= ndcs+npcs domains, or on the plain node-count cap -# otherwise (under-provisioned). +# With FD enabled, the risk budget is npcs, spent per domain at a rate of +# min(nodes_down_in_that_domain, chunks_per_domain), where chunks_per_domain +# = ceil((ndcs+npcs)/domains_available). A domain that already has +# chunks_per_domain nodes down has maxed its contribution -- further nodes in +# THAT SAME domain are free -- but a node in a domain that hasn't maxed out +# yet is only allowed if the combined risk across every affected domain still +# leaves room in the npcs budget. This mirrors the operator's fdDrainGate +# (nodedrain_controller.go) and reduces to "up to npcs whole domains are +# free" when there are >= ndcs+npcs domains (chunks_per_domain == 1). +# +# Confirmed against the backend team's stated requirements (2026-08, Dmitrii +# Iakovlev): for a 2+2 layout, +# - 2 failure domains: safe combos are "1 whole FD down" (any node count +# within it) OR "1 node in each of the 2 FDs" -- nothing beyond that. +# - 3 failure domains: only 1 FD may be fully down, not 2. +# - 4 failure domains: 2 FDs may be fully down (the well-provisioned case). # --------------------------------------------------------------------------- class TestFailureDomainAwareCapacity(unittest.TestCase): @@ -666,7 +676,151 @@ def test_under_provisioned_second_domain_blocked_at_node_budget(self): db = _db(cl, nodes) allowed, reason = _check_ftt_allows_node_removal("n1", db) self.assertFalse(allowed) - self.assertIn("insufficient failure domains", reason) + self.assertIn("risk budget", reason) + + # --- Dmitrii's confirmed scenarios: 2 FD / 2+2 --------------------------- + # chunks_per_domain = ceil(4/2) = 2. Safe: "1 whole FD" or "1 node per FD", + # nothing more -- verified against the specific combinations he ruled out. + + def test_2fd_one_whole_fd_down_is_safe(self): + """Domain 1 already has 3-of-4 nodes down; the 4th is still allowed + (piling within a single domain, which may go fully down).""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=1), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n4", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n5", failure_domain=2), _node("n6", failure_domain=2), + _node("n7", failure_domain=2), _node("n8", failure_domain=2), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertTrue(allowed) + + def test_2fd_one_node_per_fd_is_safe(self): + """1 node down in domain 1; removing 1 node in domain 2 is safe.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=2), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", failure_domain=1), + _node("n4", failure_domain=2), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertTrue(allowed) + + def test_2fd_whole_fd_plus_other_fd_node_is_unsafe(self): + """Domain 1 fully down (4 nodes) -- a node in domain 2 must now be + blocked (not one of the two safe combinations).""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=2), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n4", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n5", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n6", failure_domain=2), _node("n7", failure_domain=2), _node("n8", failure_domain=2), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) + + def test_2fd_one_per_fd_plus_second_node_in_same_fd_is_unsafe(self): + """1 node down in EACH of domains 1 and 2 already (the safe combo) -- + piling a SECOND node onto domain 2 must now be blocked, even though + domain 2 is already 'active'. This is the exact gap the old + unconditional-piling logic missed.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=2), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=2), + _node("n4", failure_domain=1), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) + + def test_2fd_one_per_fd_plus_second_node_in_other_fd_is_unsafe(self): + """Same setup, but piling the extra node onto domain 1 (the first, + 'already active' domain) instead of domain 2 -- also unsafe.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=1), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=2), + _node("n4", failure_domain=2), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) + + # --- Dmitrii's confirmed scenarios: 3 FD / 2+2 --------------------------- + # chunks_per_domain = ceil(4/3) = 2 -- same per-domain cap as 2 FD, but + # spread across 3 domains. Confirmed: 1 FD fully down is safe, 2 FDs is not. + + def test_3fd_one_whole_fd_down_is_safe(self): + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=1), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n4", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n5", failure_domain=2), _node("n6", failure_domain=3), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertTrue(allowed) + + def test_3fd_two_whole_fds_down_is_unsafe(self): + """Domain 1 fully down already (2+ nodes, maxing its chunks_per_domain + budget) -- a node in domain 2 must be blocked.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=2), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n4", failure_domain=2), _node("n5", failure_domain=3), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) + + # --- Dmitrii's confirmed scenarios: 4 FD / 2+2 (well-provisioned) -------- + # chunks_per_domain = ceil(4/4) = 1. Confirmed: 2 whole FDs down is safe. + + def test_4fd_two_whole_fds_down_is_safe(self): + """Domain 1 already has 3-of-4 nodes down (maxed, chunks_per_domain=1 + means it only ever contributes 1 to the budget regardless of count); + removing a node in domain 2 -- opening the second domain -- is still + within the npcs=2 budget.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=2), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n4", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n5", failure_domain=2), _node("n6", failure_domain=3), _node("n7", failure_domain=4), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertTrue(allowed) + + def test_4fd_three_whole_fds_down_is_unsafe(self): + """Domains 1 and 2 already have a node down each (budget=2/2 spent); + removing a node in a THIRD domain (3) must be blocked.""" + cl = _cluster(npcs=2, ndcs=2, ft=2, enable_failure_domain=True) + nodes = [ + _node("n1", failure_domain=3), + _node("n2", status=StorageNode.STATUS_OFFLINE, failure_domain=1), + _node("n3", status=StorageNode.STATUS_OFFLINE, failure_domain=2), + _node("n4", failure_domain=4), + ] + db = _db(cl, nodes) + allowed, _ = _check_ftt_allows_node_removal("n1", db) + self.assertFalse(allowed) def test_fd_enabled_but_node_unassigned_falls_back_to_node_count(self): """FD enabled cluster-wide, but this specific node has no domain From 6b8cc018dd3ee5d918bf1e04399fc338c96db83c Mon Sep 17 00:00:00 2001 From: wmousa Date: Wed, 5 Aug 2026 18:10:25 +0200 Subject: [PATCH 5/9] fix(cluster-activate): require npcs+2 domains, not npcs+1, at fresh activation The bare correctness minimum for the interleaved rotation layout is npcs+1 distinct domains (2 for npcs=1, 3 for npcs=2) -- below that even the initial static placement is wrong: at exactly 2 domains the tertiary role mathematically always lands back in the primary's own domain, since "2 steps ahead" in a period-2 round-robin wraps to where it started (verified directly against rotation_layout: 8/8 tertiary placements landed same-domain at 2 domains, 0/12 at 3+). But a minimum-correct STATIC layout has zero spare hosts per domain, and the moment a single node is added or removed, the relocation logic (_pick_replica_relocation_node) has no spare candidate left to reassign the stranded role to. Verified directly: removing one node from a bare-minimum npcs=1/2-domain or npcs=2/3-domain layout strands another node's secondary/tertiary with _pick_replica_relocation_node returning None -- blocking the removal outright, not just degrading placement quality. This also matches the backend team's confirmed stance that a 2-FD layout can never absorb a second independent failure once one domain is down, so it's excluded at any npcs level. Fresh activation now hard-requires npcs+2 distinct domains (3 for npcs=1, 4 for npcs=2) -- one domain of spare capacity beyond the bare correctness floor, so a later single add/remove has somewhere to place the relocated role. Extracted as fd_activation_domain_count_violation() in planner.py (alongside fd_balance_violation, same pattern) since _cluster_activate itself has no unit-test mocking infrastructure and was otherwise untestable; 7 new tests cover the boundary directly. --- simplyblock_core/cluster_ops.py | 24 +++++++---- .../controllers/cluster_expansion/planner.py | 41 +++++++++++++++++++ tests/unit/test_fd_topology_policy.py | 35 ++++++++++++++++ 3 files changed, 92 insertions(+), 8 deletions(-) diff --git a/simplyblock_core/cluster_ops.py b/simplyblock_core/cluster_ops.py index d33098e72f..052d6ba4c6 100644 --- a/simplyblock_core/cluster_ops.py +++ b/simplyblock_core/cluster_ops.py @@ -966,14 +966,16 @@ def _cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None: set_cluster_status(cl_id, ols_status) raise - # Failure-domain coverage check (best-effort: warn, don't block). To - # survive losing a whole failure domain we need at least npcs+1 distinct - # domains; with fewer, placement falls back to host-disjoint and a domain - # outage may exceed the cluster's fault tolerance. + # Failure-domain coverage check (best-effort: warn, don't block). A 2-FD + # layout can never absorb a second independent failure once one domain + # is down, so the hard minimum below (enforced at fresh activation) is + # npcs+2, not npcs+1 -- this warning uses the same number so a + # reactivation that's short of it gets the same signal without being + # blocked (recovering a drifted layout must not turn into an outage). fd_desired_layout: t.Dict[str, t.Tuple[str, str]] = {} if cluster.enable_failure_domain: distinct_domains = {node.failure_domain for node in online_nodes if node.failure_domain >= 0} - min_domains = cluster.distr_npcs + 1 + min_domains = cluster.distr_npcs + 2 if len(distinct_domains) < min_domains: logger.warning( "Failure-domain feature is enabled but only %d distinct failure " @@ -1013,9 +1015,15 @@ def _fd_fail(msg: str) -> None: f"a host must sit entirely in one domain") fd_host_counts = Counter(host_fd.values()) - if len(fd_host_counts) < 2: - _fd_fail("failure domains are enabled but all hosts are in a " - "single domain; at least two domains are required") + # See fd_activation_domain_count_violation's docstring: npcs+2 + # domains, not just the bare rotation-correctness minimum, so a + # later single add/remove has a spare candidate instead of + # stranding another node's secondary/tertiary with none at all. + # This also subsumes the plain "at least two domains" floor. + domain_count_violation = fd_planner.fd_activation_domain_count_violation( + cluster.distr_npcs, len(fd_host_counts)) + if domain_count_violation: + _fd_fail(domain_count_violation) if len(set(fd_host_counts.values())) != 1: _fd_fail( f"failure domains must hold an EQUAL number of hosts at " diff --git a/simplyblock_core/controllers/cluster_expansion/planner.py b/simplyblock_core/controllers/cluster_expansion/planner.py index 324a2b79de..4f4adf9ac1 100644 --- a/simplyblock_core/controllers/cluster_expansion/planner.py +++ b/simplyblock_core/controllers/cluster_expansion/planner.py @@ -503,6 +503,47 @@ def fd_balance_violation( return None +def fd_activation_domain_count_violation( + npcs: int, distinct_domain_count: int, +) -> Optional[str]: + """Validate the number of distinct failure domains for fresh activation. + + A 2-FD layout can never absorb a second independent failure once one + domain is fully down (confirmed with the backend team), so it is not + supported at any npcs level. + + The bare *correctness* minimum for the rotation layout itself is + npcs+1 (e.g. 2 domains for npcs=1, 3 for npcs=2 -- below that even the + initial static placement is wrong: at exactly 2 domains the tertiary + role mathematically always lands back in the primary's own domain, + since "2 steps ahead" in a period-2 round-robin wraps to where it + started; verified directly against rotation_layout()). But a + minimum-correct STATIC layout has zero spare hosts per domain, and the + moment a single node is added or removed, the relocation logic + (_pick_replica_relocation_node) has no spare candidate left to + reassign the stranded role to -- verified directly: removing one node + from a bare-minimum npcs=1/2-domain or npcs=2/3-domain layout strands + another node's secondary/tertiary with no replacement at all, blocking + the removal outright rather than just degrading placement quality. + + Requiring npcs+2 domains (3 for npcs=1, 4 for npcs=2, which also rules + out exactly 2 for both) keeps one domain of spare capacity beyond the + bare correctness floor, so a single add/remove has somewhere to place + the relocated role instead of failing immediately. Returns a + human-readable reason on violation, ``None`` when the count is + acceptable. + """ + min_domains = npcs + 2 + if distinct_domain_count < min_domains: + return ( + f"failure domains are enabled with npcs={npcs}, which requires at " + f"least {min_domains} distinct failure domains (2 domains is not " + f"supported at any npcs level); currently have " + f"{distinct_domain_count}. Add hosts in additional domains, or " + f"disable failure domains, then activate.") + return None + + # --------------------------------------------------------------------------- # Persistence helpers for ``Cluster.expand_state``. # diff --git a/tests/unit/test_fd_topology_policy.py b/tests/unit/test_fd_topology_policy.py index f8540d977a..46478d0c3f 100644 --- a/tests/unit/test_fd_topology_policy.py +++ b/tests/unit/test_fd_topology_policy.py @@ -178,6 +178,41 @@ def test_unset_domain_ignored(self): self.assertIsNone(planner.fd_balance_violation({-1: 7, 0: 2, 1: 2})) +# --------------------------------------------------------------------------- +# planner.fd_activation_domain_count_violation +# --------------------------------------------------------------------------- + +class TestFdActivationDomainCount(unittest.TestCase): + + def test_npcs1_two_domains_violates(self): + self.assertIsNotNone( + planner.fd_activation_domain_count_violation(1, 2)) + + def test_npcs1_three_domains_ok(self): + self.assertIsNone( + planner.fd_activation_domain_count_violation(1, 3)) + + def test_npcs1_one_domain_violates(self): + self.assertIsNotNone( + planner.fd_activation_domain_count_violation(1, 1)) + + def test_npcs2_two_domains_violates(self): + self.assertIsNotNone( + planner.fd_activation_domain_count_violation(2, 2)) + + def test_npcs2_three_domains_violates(self): + self.assertIsNotNone( + planner.fd_activation_domain_count_violation(2, 3)) + + def test_npcs2_four_domains_ok(self): + self.assertIsNone( + planner.fd_activation_domain_count_violation(2, 4)) + + def test_npcs2_more_than_four_domains_ok(self): + self.assertIsNone( + planner.fd_activation_domain_count_violation(2, 6)) + + # --------------------------------------------------------------------------- # preconditions: add / remove / current admission # --------------------------------------------------------------------------- From bfe4e69abedd6a308237c8e9424f5011ff3bffa3 Mon Sep 17 00:00:00 2001 From: wmousa Date: Fri, 7 Aug 2026 17:38:04 +0200 Subject: [PATCH 6/9] fix(node-removal): splice a stranded replica into an existing pairing when no free cross-domain candidate exists get_secondary_nodes/get_secondary_nodes_2 only ever offer UNCLAIMED nodes (each node hosts at most one secondary/tertiary at a time). A node removal frees exactly one node cluster-wide -- whoever hosted the removed node's own role -- so _pick_replica_relocation_node has exactly one candidate to work with. If that one candidate lands in the wrong failure domain (or nothing is free at all), the direct search had nothing else to offer even though a valid rearrangement exists elsewhere in the cluster. Verified directly: two removals in a row (each individually fine) can chain into exactly this dead end -- the second removal's repair needs a new cross-domain home, the only free node is same-domain, and the search gave up. Confirmed the same 9-node/3-domain topology that hit this now resolves via the new fallback. Adds _find_splice_target_for_relocation, generalizing splice_stranded_secondary/splice_stranded_tertiary's fix for the identical dead end at cluster-activation time (splice into an already-formed pairing P->X instead of requiring an idle node) to the removal-repair path, with an exclude list for the node being removed. _pick_replica_relocation_node now falls back to it whenever the direct search comes up empty. Unlike the activation-time splice helpers -- which only ever run before any physical LVS exists -- this can be asked to splice into a pairing that already has real data on both ends, so _relocate_one_replica gained _relocate_replica_between to execute it: evict the existing occupant onto the node being relocated (tear down + rebuild), then claim the freed slot. Both legs follow the same idempotent, commit-pointers-then-build pattern _relocate_one_replica already uses, so a crash mid-splice resumes cleanly on retry. --- simplyblock_core/storage_node_ops.py | 154 +++++++++++++- tests/unit/test_node_removal.py | 294 ++++++++++++++++++++++++++- 2 files changed, 438 insertions(+), 10 deletions(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 0df8c804ba..0634f45b0d 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -3560,7 +3560,23 @@ def _pick_replica_relocation_node(primary, removed_node: StorageNode, role, db_c enforced HARD here (not best-effort): if the primary's OTHER non-leader role does not already live in a different domain than the primary, the replacement must — otherwise a full-domain outage would leave the LVS - with zero surviving paths. Returning None makes + with zero surviving paths. + + ``get_secondary_nodes``/``get_secondary_nodes_2`` only ever offer + UNCLAIMED nodes (each node hosts at most one secondary/tertiary at a + time — ``lvstore_stack_secondary``/``_tertiary`` is a single field, not + a list). A removal frees exactly one node system-wide (whoever hosted + ``removed_node``'s own role); if that one lands in the wrong domain — + or nothing is free at all — the direct search has nothing else to + offer even though a valid rearrangement exists elsewhere in the + cluster (2026-08-07, chained-removal incident: two removals in a row + stranded a third node's secondary with zero free cross-domain + candidates, while an existing pairing two hops away could have + absorbed it). Falls back to splicing ``primary`` into an already-formed + pairing (see ``_find_splice_target_for_relocation``) — exactly the fix + ``splice_stranded_secondary``/``splice_stranded_tertiary`` already apply + to the identical dead end at cluster-activation time. Only returning + None here (both searches exhausted) makes ``_check_replica_relocation_feasible`` refuse the removal up front. """ exclude_ids = [removed_node.get_id()] @@ -3581,12 +3597,9 @@ def _pick_replica_relocation_node(primary, removed_node: StorageNode, role, db_c pass cands = get_secondary_nodes_2( primary, exclude_ids=exclude_ids, exclude_mgmt_ips=exclude_mgmt_ips) - if not cands: - return None cluster = db_controller.get_cluster_by_id(primary.cluster_id) - if (getattr(cluster, "enable_failure_domain", False) - and primary.failure_domain >= 0): + if cands and getattr(cluster, "enable_failure_domain", False) and primary.failure_domain >= 0: other_cross = False if other_id and other_id != removed_node.get_id(): try: @@ -3606,8 +3619,84 @@ def _pick_replica_relocation_node(primary, removed_node: StorageNode, role, db_c if (cand.failure_domain >= 0 and cand.failure_domain != primary.failure_domain): return cand_id - return None - return cands[0] + else: + return cands[0] + elif cands: + return cands[0] + + splice = _find_splice_target_for_relocation( + primary, role, db_controller, exclude_ids=exclude_ids + [primary.get_id()]) + return splice[1] if splice else None + + +def _find_splice_target_for_relocation(stranded_primary, role, db_controller, exclude_ids=()): + """Find an already-formed pairing ``P -> X`` (``P. == X``) + elsewhere in the cluster to splice ``stranded_primary`` into: + ``P -> stranded_primary -> X``. Read-only — callers decide whether and + how to execute the resulting move (see ``_relocate_one_replica``). + + Generalizes ``splice_stranded_secondary``/``splice_stranded_tertiary``'s + edge search (same scoring: prefer both ends domain-disjoint from the + stranded node, then relax) with an ``exclude_ids`` list, so the + node-removal repair path can rule out the node being removed and any + other already-claimed id. Unlike the activation-time splice helpers — + which only ever run before any physical LVS exists — this can be asked + to splice into a pairing that already has real data on both ends; + executing that move (not just picking the edge) is the caller's job. + + Returns ``(p_id, x_id)`` or ``None`` if no valid edge exists. + """ + field = "secondary_node_id" if role == "secondary" else "tertiary_node_id" + all_nodes = db_controller.get_storage_nodes_by_cluster_id(stranded_primary.cluster_id) + all_nodes = sorted(all_nodes, key=lambda n: n.failure_domain) + by_id = {n.get_id(): n for n in all_nodes} + exclude = set(exclude_ids) | {stranded_primary.get_id()} + + stranded_sec = None + if role == "tertiary" and stranded_primary.secondary_node_id: + stranded_sec = by_id.get(stranded_primary.secondary_node_id) + + def _online(*nodes): + return all(n.status == StorageNode.STATUS_ONLINE for n in nodes) + + def _valid_tertiary(node, node_sec, candidate): + if candidate.get_id() == node.get_id(): + return False + if candidate.mgmt_ip == node.mgmt_ip: + return False + if node_sec and candidate.mgmt_ip == node_sec.mgmt_ip: + return False + return True + + def _domain_mismatch_score(*nodes): + if stranded_primary.failure_domain < 0: + return 0 + return sum(1 for n in nodes if n.failure_domain != stranded_primary.failure_domain) + + edges = [n for n in all_nodes if getattr(n, field) and n.get_id() not in exclude] + + best, best_score = None, -1 + for p in edges: + x_id = getattr(p, field) + if x_id in exclude: + continue + x = by_id.get(x_id) + if not x or not _online(p, x): + continue + if role == "secondary": + if p.mgmt_ip == stranded_primary.mgmt_ip or x.mgmt_ip == stranded_primary.mgmt_ip: + continue + else: + p_sec = by_id.get(p.secondary_node_id) if p.secondary_node_id else None + if not _valid_tertiary(p, p_sec, stranded_primary): + continue + if not _valid_tertiary(stranded_primary, stranded_sec, x): + continue + score = _domain_mismatch_score(p, x) + if score > best_score: + best_score, best = score, (p.get_id(), x.get_id()) + + return best def node_removal_orchestrate(node_id, force_remove=False): @@ -3796,6 +3885,21 @@ def _relocate_one_replica(removed_node: StorageNode, primary_id, role): logger.error( f"[REMOVAL] no relocation target for {role} replica of {primary_id}") return False + + new_node = db_controller.get_storage_node_by_id(new_id) + occupant_id = getattr(new_node, backref) + if occupant_id and occupant_id not in (primary_id, removed_node.get_id()): + # _pick_replica_relocation_node fell back to a splice candidate: + # new_id is currently busy hosting occupant_id's replica. Evict + # that occupant onto `primary` (the node whose replica we're + # relocating) before claiming new_id for primary's own role — + # see _find_splice_target_for_relocation's docstring. + if not _relocate_replica_between(occupant_id, new_id, primary_id, role, db_controller): + logger.error( + f"[REMOVAL] failed to splice {primary_id} into the pairing " + f"occupying {new_id} (occupant {occupant_id})") + return False + primary = db_controller.get_storage_node_by_id(primary_id) setattr(primary, field, new_id) primary.write_to_db() @@ -3819,6 +3923,42 @@ def _relocate_one_replica(removed_node: StorageNode, primary_id, role): return True +def _relocate_replica_between(occupant_primary_id, old_host_id, new_host_id, role, db_controller): + """Physically move ``occupant_primary_id``'s ``role`` replica off + ``old_host_id`` onto ``new_host_id``, updating its forward pointer. + + Used by the splice fallback in ``_relocate_one_replica``: before an + already-busy node can be claimed for the primary being relocated, its + current occupant must move onto that primary's node instead (see + ``_find_splice_target_for_relocation``'s docstring for why an + already-formed pairing, not an idle node, is what's available). + + Idempotent: skips the pointer flip (and the teardown it implies) if a + prior attempt already committed it; ``recreate_lvstore_on_non_leader`` + is retried unconditionally either way, matching ``_relocate_one_replica``'s + own idempotency pattern. Returns True if ``occupant_primary_id`` no + longer exists — nothing left to relocate. + """ + field = "secondary_node_id" if role == "secondary" else "tertiary_node_id" + try: + occupant_primary = db_controller.get_storage_node_by_id(occupant_primary_id) + except KeyError: + return True + + if getattr(occupant_primary, field) != new_host_id: + old_host = db_controller.get_storage_node_by_id(old_host_id) + cluster = db_controller.get_cluster_by_id(occupant_primary.cluster_id) + if old_host.status == StorageNode.STATUS_ONLINE: + _delete_replica_on_peer(old_host, occupant_primary, cluster) + occupant_primary = db_controller.get_storage_node_by_id(occupant_primary_id) + setattr(occupant_primary, field, new_host_id) + occupant_primary.write_to_db() + + new_host = db_controller.get_storage_node_by_id(new_host_id) + occupant_primary = db_controller.get_storage_node_by_id(occupant_primary_id) + return bool(recreate_lvstore_on_non_leader(new_host, occupant_primary, occupant_primary)) + + def _clear_replica_backref(removed_node: StorageNode, backref): db_controller = DBController() removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) diff --git a/tests/unit/test_node_removal.py b/tests/unit/test_node_removal.py index 3a0b78840a..f7461edeca 100644 --- a/tests/unit/test_node_removal.py +++ b/tests/unit/test_node_removal.py @@ -30,7 +30,8 @@ # Fixtures # --------------------------------------------------------------------------- -def _cluster(ha_type="ha", npcs=1, ndcs=2, ft=1, mode="docker"): +def _cluster(ha_type="ha", npcs=1, ndcs=2, ft=1, mode="docker", + enable_failure_domain=False): cl = Cluster() cl.uuid = "cluster-1" cl.ha_type = ha_type @@ -40,12 +41,14 @@ def _cluster(ha_type="ha", npcs=1, ndcs=2, ft=1, mode="docker"): cl.mode = mode cl.status = Cluster.STATUS_ACTIVE cl.nqn = "nqn.2023-01.io.simplyblock:cluster-1" + cl.enable_failure_domain = enable_failure_domain return cl def _node(node_id, status=StorageNode.STATUS_ONLINE, lvstore="", secondary_id="", tertiary_id="", - stack_secondary="", stack_tertiary="", n_devices=0, with_jm=False): + stack_secondary="", stack_tertiary="", n_devices=0, with_jm=False, + failure_domain=-1, mgmt_ip=None): n = MagicMock(spec=StorageNode) n.uuid = node_id n.get_id = MagicMock(return_value=node_id) @@ -59,7 +62,8 @@ def _node(node_id, status=StorageNode.STATUS_ONLINE, lvstore="", n.tertiary_node_id = tertiary_id n.lvstore_stack_secondary = stack_secondary n.lvstore_stack_tertiary = stack_tertiary - n.mgmt_ip = f"10.0.0.{abs(hash(node_id)) % 250 + 1}" + n.failure_domain = failure_domain + n.mgmt_ip = mgmt_ip or f"10.0.0.{abs(hash(node_id)) % 250 + 1}" n.write_to_db = MagicMock() n.rpc_client = MagicMock(return_value=MagicMock()) n.hublvol_nqn_for_lvstore = MagicMock(return_value=f"nqn:hub:{lvstore}") @@ -260,6 +264,160 @@ def test_pick_secondary_uses_get_secondary_nodes(self): self.assertIn("n9", kwargs["exclude_ids"]) +# --------------------------------------------------------------------------- +# _find_splice_target_for_relocation — the removal-repair fallback used when +# get_secondary_nodes/_2 offer no free (unclaimed) cross-domain candidate. +# +# Regression coverage for the 2026-08-07 chained-removal incident: two +# removals in a row can strand a third node's secondary with zero free +# cross-domain candidates, even though an existing pairing two hops away +# could absorb it. Mirrors splice_stranded_secondary/_tertiary's edge search +# (used for the identical dead end at activation time), generalized with an +# exclude list for the removal path. +# --------------------------------------------------------------------------- + +class TestFindSpliceTargetForRelocation(unittest.TestCase): + + def test_splices_into_existing_secondary_edge(self): + cl = _cluster(enable_failure_domain=True) + p = _node("p", secondary_id="x", failure_domain=0, mgmt_ip="10.0.0.1") + x = _node("x", failure_domain=1, mgmt_ip="10.0.0.2") + stranded = _node("s", failure_domain=2, mgmt_ip="10.0.0.3") + db = FakeDB(cl, [p, x, stranded]) + got = storage_node_ops._find_splice_target_for_relocation(stranded, "secondary", db) + self.assertEqual(got, ("p", "x")) + + def test_prefers_edge_domain_disjoint_on_both_ends(self): + cl = _cluster(enable_failure_domain=True) + stranded = _node("s", failure_domain=0, mgmt_ip="10.0.0.9") + bad_p = _node("bad_p", secondary_id="bad_x", failure_domain=0, mgmt_ip="10.0.0.1") + bad_x = _node("bad_x", failure_domain=0, mgmt_ip="10.0.0.2") + good_p = _node("good_p", secondary_id="good_x", failure_domain=1, mgmt_ip="10.0.0.3") + good_x = _node("good_x", failure_domain=2, mgmt_ip="10.0.0.4") + db = FakeDB(cl, [stranded, bad_p, bad_x, good_p, good_x]) + got = storage_node_ops._find_splice_target_for_relocation(stranded, "secondary", db) + self.assertEqual(got, ("good_p", "good_x")) + + def test_excludes_ids_passed_by_caller(self): + cl = _cluster(enable_failure_domain=True) + p = _node("p", secondary_id="x", failure_domain=0, mgmt_ip="10.0.0.1") + x = _node("x", failure_domain=1, mgmt_ip="10.0.0.2") + stranded = _node("s", failure_domain=2, mgmt_ip="10.0.0.3") + db = FakeDB(cl, [p, x, stranded]) + got = storage_node_ops._find_splice_target_for_relocation( + stranded, "secondary", db, exclude_ids=["x"]) + self.assertIsNone(got) + + def test_no_edge_exists_returns_none(self): + cl = _cluster(enable_failure_domain=True) + stranded = _node("s", failure_domain=0, mgmt_ip="10.0.0.1") + other = _node("other", failure_domain=1, mgmt_ip="10.0.0.2") + db = FakeDB(cl, [stranded, other]) + got = storage_node_ops._find_splice_target_for_relocation(stranded, "secondary", db) + self.assertIsNone(got) + + def test_skips_edge_with_offline_endpoint(self): + cl = _cluster(enable_failure_domain=True) + p = _node("p", secondary_id="x", failure_domain=0, mgmt_ip="10.0.0.1") + x = _node("x", failure_domain=1, mgmt_ip="10.0.0.2", + status=StorageNode.STATUS_OFFLINE) + stranded = _node("s", failure_domain=2, mgmt_ip="10.0.0.3") + db = FakeDB(cl, [p, x, stranded]) + got = storage_node_ops._find_splice_target_for_relocation(stranded, "secondary", db) + self.assertIsNone(got) + + def test_skips_edge_not_host_disjoint_from_stranded(self): + stranded = _node("s", failure_domain=0, mgmt_ip="10.0.0.5") + p = _node("p", secondary_id="x", failure_domain=1, mgmt_ip="10.0.0.1") + x = _node("x", failure_domain=2, mgmt_ip="10.0.0.5") # shares stranded's host + cl = _cluster(enable_failure_domain=True) + db = FakeDB(cl, [stranded, p, x]) + got = storage_node_ops._find_splice_target_for_relocation(stranded, "secondary", db) + self.assertIsNone(got) + + def test_tertiary_edge_respects_secondary_host_disjointness(self): + cl = _cluster(enable_failure_domain=True) + stranded = _node("s", failure_domain=0, secondary_id="s_sec", mgmt_ip="10.0.0.9") + s_sec = _node("s_sec", failure_domain=1, mgmt_ip="10.0.0.50") + p = _node("p", tertiary_id="x", failure_domain=0, mgmt_ip="10.0.0.60") + x = _node("x", failure_domain=1, mgmt_ip="10.0.0.61") + db = FakeDB(cl, [stranded, s_sec, p, x]) + got = storage_node_ops._find_splice_target_for_relocation(stranded, "tertiary", db) + self.assertEqual(got, ("p", "x")) + + +# --------------------------------------------------------------------------- +# _pick_replica_relocation_node — falls back to the splice finder above when +# the direct free-candidate search comes up empty (no candidates at all, or +# none that satisfy the hard cross-domain requirement). +# --------------------------------------------------------------------------- + +class TestPickReplicaRelocationSpliceFallback(unittest.TestCase): + + def test_falls_back_to_splice_when_no_free_candidate_at_all(self): + cl = _cluster(enable_failure_domain=True) + primary = _node("p1", secondary_id="n1", failure_domain=2, mgmt_ip="10.0.0.9") + removed = _node("n1", failure_domain=1, mgmt_ip="10.0.0.99") + edge_p = _node("edge_p", secondary_id="edge_x", failure_domain=0, mgmt_ip="10.0.0.1") + edge_x = _node("edge_x", failure_domain=1, mgmt_ip="10.0.0.2") + db = FakeDB(cl, [primary, removed, edge_p, edge_x]) + with patch.object(storage_node_ops, "get_secondary_nodes", return_value=[]): + got = storage_node_ops._pick_replica_relocation_node( + primary, removed, "secondary", db) + self.assertEqual(got, "edge_x") + + def test_falls_back_to_splice_when_only_candidate_is_same_domain(self): + cl = _cluster(enable_failure_domain=True) + primary = _node("p1", secondary_id="n1", failure_domain=2, mgmt_ip="10.0.0.9") + removed = _node("n1", failure_domain=1, mgmt_ip="10.0.0.99") + same_domain_cand = _node("same_domain_cand", failure_domain=2, mgmt_ip="10.0.0.8") + edge_p = _node("edge_p", secondary_id="edge_x", failure_domain=0, mgmt_ip="10.0.0.1") + edge_x = _node("edge_x", failure_domain=1, mgmt_ip="10.0.0.2") + db = FakeDB(cl, [primary, removed, same_domain_cand, edge_p, edge_x]) + with patch.object(storage_node_ops, "get_secondary_nodes", + return_value=["same_domain_cand"]): + got = storage_node_ops._pick_replica_relocation_node( + primary, removed, "secondary", db) + self.assertEqual(got, "edge_x") + + def test_returns_none_when_no_free_and_no_splice_candidate(self): + cl = _cluster(enable_failure_domain=True) + primary = _node("p1", secondary_id="n1", failure_domain=0, mgmt_ip="10.0.0.9") + removed = _node("n1", failure_domain=1, mgmt_ip="10.0.0.99") + db = FakeDB(cl, [primary, removed]) + with patch.object(storage_node_ops, "get_secondary_nodes", return_value=[]): + got = storage_node_ops._pick_replica_relocation_node( + primary, removed, "secondary", db) + self.assertIsNone(got) + + def test_does_not_use_splice_when_free_cross_domain_candidate_exists(self): + cl = _cluster(enable_failure_domain=True) + primary = _node("p1", secondary_id="n1", failure_domain=0, mgmt_ip="10.0.0.9") + removed = _node("n1", failure_domain=1, mgmt_ip="10.0.0.99") + free1 = _node("free1", failure_domain=1, mgmt_ip="10.0.0.1") + db = FakeDB(cl, [primary, removed, free1]) + with patch.object(storage_node_ops, "get_secondary_nodes", return_value=["free1"]), \ + patch.object(storage_node_ops, "_find_splice_target_for_relocation") as finder: + got = storage_node_ops._pick_replica_relocation_node( + primary, removed, "secondary", db) + self.assertEqual(got, "free1") + finder.assert_not_called() + + def test_fd_disabled_never_needs_splice_when_candidate_exists(self): + # Non-FD clusters take the unconditional cands[0] path -- splice is + # only relevant once cands is genuinely empty. + cl = _cluster(enable_failure_domain=False) + primary = _node("p1", secondary_id="n1", mgmt_ip="10.0.0.9") + removed = _node("n1", mgmt_ip="10.0.0.99") + db = FakeDB(cl, [primary, removed]) + with patch.object(storage_node_ops, "get_secondary_nodes", return_value=["free1"]), \ + patch.object(storage_node_ops, "_find_splice_target_for_relocation") as finder: + got = storage_node_ops._pick_replica_relocation_node( + primary, removed, "secondary", db) + self.assertEqual(got, "free1") + finder.assert_not_called() + + # --------------------------------------------------------------------------- # Case A — teardown of own primary's replicas # --------------------------------------------------------------------------- @@ -411,6 +569,136 @@ def test_relocate_missing_primary_just_clears(self): self.assertEqual(removed.lvstore_stack_secondary, "") +# --------------------------------------------------------------------------- +# Case B, splice fallback — _pick_replica_relocation_node returned a BUSY +# node (a splice candidate, per _find_splice_target_for_relocation) instead +# of a free one. _relocate_one_replica must evict that node's current +# occupant onto the stranded primary before claiming the slot for itself. +# --------------------------------------------------------------------------- + +class TestRelocateOneReplicaSpliceExecution(unittest.TestCase): + + def test_relocate_via_splice_evicts_occupant_first(self): + cl = _cluster() + removed = _node("n1", stack_secondary="stranded") + stranded = _node("stranded", secondary_id="n1", lvstore="LVS_stranded") + occupant = _node("occupant", secondary_id="x", lvstore="LVS_occupant") + x = _node("x", stack_secondary="occupant") # x is busy, not free + db = FakeDB(cl, [removed, stranded, occupant, x]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="x"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + return_value=True) as rec, \ + patch.object(storage_node_ops, "_delete_replica_on_peer") as drp: + ret = storage_node_ops._relocate_one_replica(removed, "stranded", "secondary") + + self.assertTrue(ret) + drp.assert_called_once() # occupant's old replica torn down off x + self.assertEqual(occupant.secondary_node_id, "stranded") # occupant re-homed onto stranded + self.assertEqual(stranded.secondary_node_id, "x") # stranded takes over x's freed slot + self.assertEqual(x.lvstore_stack_secondary, "stranded") + self.assertEqual(rec.call_count, 2) # occupant's rebuild + stranded's own rebuild + self.assertEqual(removed.lvstore_stack_secondary, "") + + def test_relocate_via_splice_occupant_rebuild_failure_keeps_forward_pointers(self): + cl = _cluster() + removed = _node("n1", stack_secondary="stranded") + stranded = _node("stranded", secondary_id="n1", lvstore="LVS_stranded") + occupant = _node("occupant", secondary_id="x", lvstore="LVS_occupant") + x = _node("x", stack_secondary="occupant") + db = FakeDB(cl, [removed, stranded, occupant, x]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="x"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + return_value=False), \ + patch.object(storage_node_ops, "_delete_replica_on_peer"): + ret = storage_node_ops._relocate_one_replica(removed, "stranded", "secondary") + + self.assertFalse(ret) + # Forward pointer for occupant's move is committed even though the + # physical rebuild failed -- matches _relocate_one_replica's own + # idempotent-retry pattern: a retry resumes from the committed intent + # rather than re-picking (see test_relocate_resume_reuses_committed_target). + self.assertEqual(occupant.secondary_node_id, "stranded") + # The outer splice claim (stranded -> x) never got committed, since + # _relocate_replica_between reported failure first. + self.assertEqual(stranded.secondary_node_id, "n1") + self.assertEqual(removed.lvstore_stack_secondary, "stranded") + + def test_relocate_via_splice_own_rebuild_failure_after_occupant_moved(self): + # occupant's move succeeds (evicted + rebuilt on stranded), but + # stranded's own rebuild on the freed slot x fails. + cl = _cluster() + removed = _node("n1", stack_secondary="stranded") + stranded = _node("stranded", secondary_id="n1", lvstore="LVS_stranded") + occupant = _node("occupant", secondary_id="x", lvstore="LVS_occupant") + x = _node("x", stack_secondary="occupant") + db = FakeDB(cl, [removed, stranded, occupant, x]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="x"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + side_effect=[True, False]), \ + patch.object(storage_node_ops, "_delete_replica_on_peer") as drp: + ret = storage_node_ops._relocate_one_replica(removed, "stranded", "secondary") + + self.assertFalse(ret) + drp.assert_called_once() + self.assertEqual(occupant.secondary_node_id, "stranded") + # Forward pointers for stranded's own claim ARE committed (pre-build, + # same idempotent pattern) even though the rebuild on x failed. + self.assertEqual(stranded.secondary_node_id, "x") + self.assertEqual(x.lvstore_stack_secondary, "stranded") + self.assertEqual(removed.lvstore_stack_secondary, "stranded") + + def test_relocate_via_splice_tertiary_role(self): + cl = _cluster() + removed = _node("n1", stack_tertiary="stranded") + stranded = _node("stranded", tertiary_id="n1", lvstore="LVS_stranded") + occupant = _node("occupant", tertiary_id="x", lvstore="LVS_occupant") + x = _node("x", stack_tertiary="occupant") + db = FakeDB(cl, [removed, stranded, occupant, x]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="x"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + return_value=True), \ + patch.object(storage_node_ops, "_delete_replica_on_peer") as drp: + ret = storage_node_ops._relocate_one_replica(removed, "stranded", "tertiary") + + self.assertTrue(ret) + drp.assert_called_once() + self.assertEqual(occupant.tertiary_node_id, "stranded") + self.assertEqual(stranded.tertiary_node_id, "x") + self.assertEqual(x.lvstore_stack_tertiary, "stranded") + self.assertEqual(removed.lvstore_stack_tertiary, "") + + def test_relocate_free_target_never_triggers_splice_eviction(self): + # Regression guard: when the picked target is genuinely free (no + # backref set), _relocate_one_replica must behave exactly as before + # -- no eviction, no extra recreate_lvstore_on_non_leader call. + cl = _cluster() + removed = _node("n1", stack_secondary="p1") + primary = _node("p1", secondary_id="n1", lvstore="LVS_p1") + free_node = _node("n3") # stack_secondary="" -- genuinely free + db = FakeDB(cl, [removed, primary, free_node]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="n3"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + return_value=True) as rec, \ + patch.object(storage_node_ops, "_delete_replica_on_peer") as drp: + ret = storage_node_ops._relocate_one_replica(removed, "p1", "secondary") + + self.assertTrue(ret) + drp.assert_not_called() + rec.assert_called_once() + self.assertEqual(primary.secondary_node_id, "n3") + self.assertEqual(free_node.lvstore_stack_secondary, "p1") + + # --------------------------------------------------------------------------- # Device decommission completion gate # --------------------------------------------------------------------------- From b7275c922b70ca24866339ef4b3687af1d0a1475 Mon Sep 17 00:00:00 2001 From: wmousa Date: Fri, 7 Aug 2026 23:18:40 +0200 Subject: [PATCH 7/9] fix(node-removal): create-before-destroy in the splice fallback _relocate_replica_between tore down the occupant's existing, healthy replica BEFORE building its replacement -- between those two steps the occupant had zero surviving copies. Under FTT1 (no tertiary) that's a real gap: a cluster only tolerates one node down at a time, and that budget belongs to the node actually being removed, not to whatever unrelated, healthy node the splice happens to touch. Confirmed live (2026-08-07): removing a node correctly triggered the splice fallback, but the rebuild step hit a hublvol attach failure and RAISED an exception instead of returning False. That propagated uncaught, the task retried repeatedly, and the occupant's only copy sat torn down the whole time -- the cluster's health monitor eventually suspended it. Reorders to create-before-destroy: build the replacement on the new host first (old copy stays live and serving throughout); only tear down the old copy once the new one is confirmed. A raised exception from the rebuild is now caught and treated the same as a returned False -- both leave the old copy untouched and safe to retry. The teardown step is guarded by the old host's own back-reference (not the occupant's forward pointer), so a crash between the two commits still resumes the teardown on the next pass instead of leaking a stale replica. --- simplyblock_core/storage_node_ops.py | 55 ++++++++++++++++----- tests/unit/test_node_removal.py | 72 ++++++++++++++++++++++++---- 2 files changed, 106 insertions(+), 21 deletions(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 0634f45b0d..f3649c6abf 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -3933,30 +3933,61 @@ def _relocate_replica_between(occupant_primary_id, old_host_id, new_host_id, rol ``_find_splice_target_for_relocation``'s docstring for why an already-formed pairing, not an idle node, is what's available). - Idempotent: skips the pointer flip (and the teardown it implies) if a - prior attempt already committed it; ``recreate_lvstore_on_non_leader`` - is retried unconditionally either way, matching ``_relocate_one_replica``'s - own idempotency pattern. Returns True if ``occupant_primary_id`` no - longer exists — nothing left to relocate. + Create-before-destroy: the new replica is built on ``new_host_id`` + BEFORE the old one on ``old_host_id`` is torn down, so + ``occupant_primary`` never has zero surviving copies -- critical on + FTT1 (no tertiary): a cluster only tolerates one node down at a time, + and that budget belongs to the node actually being removed, not to + whatever healthy node this splice happens to touch. (2026-08-07 + incident: the old destroy-then-build order tore down the occupant's + only copy up front; a hublvol attach failure on the rebuild then + retried for minutes with that copy already gone.) A raised exception + from the rebuild is treated the same as a returned False -- both leave + the old copy untouched and safe to retry. + + Idempotent and retry-safe: "already built" is read from the occupant's + forward pointer, so a retry after a confirmed build skips straight to + the teardown check without re-running the rebuild. The teardown itself + is guarded separately by ``old_host``'s own back-reference (not the + occupant's forward pointer), so a crash between the two commits still + resumes the teardown on the next pass instead of leaking a stale + replica on ``old_host`` forever. + + Returns True if ``occupant_primary_id`` no longer exists — nothing left + to relocate. """ field = "secondary_node_id" if role == "secondary" else "tertiary_node_id" + backref = "lvstore_stack_secondary" if role == "secondary" else "lvstore_stack_tertiary" try: occupant_primary = db_controller.get_storage_node_by_id(occupant_primary_id) except KeyError: return True if getattr(occupant_primary, field) != new_host_id: - old_host = db_controller.get_storage_node_by_id(old_host_id) - cluster = db_controller.get_cluster_by_id(occupant_primary.cluster_id) - if old_host.status == StorageNode.STATUS_ONLINE: - _delete_replica_on_peer(old_host, occupant_primary, cluster) + new_host = db_controller.get_storage_node_by_id(new_host_id) + try: + built = recreate_lvstore_on_non_leader(new_host, occupant_primary, occupant_primary) + except Exception as e: + logger.error( + f"[REMOVAL] splice: failed to build {role} replica of " + f"{occupant_primary_id} on {new_host_id}, old copy on " + f"{old_host_id} left untouched: {e}") + return False + if not built: + return False occupant_primary = db_controller.get_storage_node_by_id(occupant_primary_id) setattr(occupant_primary, field, new_host_id) occupant_primary.write_to_db() - new_host = db_controller.get_storage_node_by_id(new_host_id) - occupant_primary = db_controller.get_storage_node_by_id(occupant_primary_id) - return bool(recreate_lvstore_on_non_leader(new_host, occupant_primary, occupant_primary)) + old_host = db_controller.get_storage_node_by_id(old_host_id) + if getattr(old_host, backref) == occupant_primary_id: + cluster = db_controller.get_cluster_by_id(occupant_primary.cluster_id) + if old_host.status == StorageNode.STATUS_ONLINE: + _delete_replica_on_peer(old_host, occupant_primary, cluster) + old_host = db_controller.get_storage_node_by_id(old_host_id) + setattr(old_host, backref, "") + old_host.write_to_db() + return True def _clear_replica_backref(removed_node: StorageNode, backref): diff --git a/tests/unit/test_node_removal.py b/tests/unit/test_node_removal.py index f7461edeca..4b02b09111 100644 --- a/tests/unit/test_node_removal.py +++ b/tests/unit/test_node_removal.py @@ -601,7 +601,11 @@ def test_relocate_via_splice_evicts_occupant_first(self): self.assertEqual(rec.call_count, 2) # occupant's rebuild + stranded's own rebuild self.assertEqual(removed.lvstore_stack_secondary, "") - def test_relocate_via_splice_occupant_rebuild_failure_keeps_forward_pointers(self): + def test_relocate_via_splice_occupant_rebuild_failure_leaves_old_copy_untouched(self): + # Create-before-destroy: a failed rebuild on the stranded node must + # NOT tear down or repoint the occupant's still-intact old copy on x + # -- that copy is occupant's ONLY surviving replica under FTT1, so a + # failed build must change nothing about it (2026-08-07 incident). cl = _cluster() removed = _node("n1", stack_secondary="stranded") stranded = _node("stranded", secondary_id="n1", lvstore="LVS_stranded") @@ -613,20 +617,70 @@ def test_relocate_via_splice_occupant_rebuild_failure_keeps_forward_pointers(sel return_value="x"), \ patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", return_value=False), \ - patch.object(storage_node_ops, "_delete_replica_on_peer"): + patch.object(storage_node_ops, "_delete_replica_on_peer") as drp: ret = storage_node_ops._relocate_one_replica(removed, "stranded", "secondary") self.assertFalse(ret) - # Forward pointer for occupant's move is committed even though the - # physical rebuild failed -- matches _relocate_one_replica's own - # idempotent-retry pattern: a retry resumes from the committed intent - # rather than re-picking (see test_relocate_resume_reuses_committed_target). - self.assertEqual(occupant.secondary_node_id, "stranded") - # The outer splice claim (stranded -> x) never got committed, since - # _relocate_replica_between reported failure first. + drp.assert_not_called() # old copy on x never torn down + self.assertEqual(occupant.secondary_node_id, "x") # unchanged -- still protected + self.assertEqual(x.lvstore_stack_secondary, "occupant") # unchanged + # The outer splice claim (stranded -> x) never got committed either. self.assertEqual(stranded.secondary_node_id, "n1") self.assertEqual(removed.lvstore_stack_secondary, "stranded") + def test_relocate_via_splice_occupant_rebuild_raises_treated_as_failure(self): + # The 2026-08-07 incident's actual failure mode: recreate_lvstore_on_non_leader + # RAISED (a hublvol attach error) instead of returning False. Must be + # caught and handled identically to a returned False -- old copy on x + # stays untouched either way. + cl = _cluster() + removed = _node("n1", stack_secondary="stranded") + stranded = _node("stranded", secondary_id="n1", lvstore="LVS_stranded") + occupant = _node("occupant", secondary_id="x", lvstore="LVS_occupant") + x = _node("x", stack_secondary="occupant") + db = FakeDB(cl, [removed, stranded, occupant, x]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="x"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + side_effect=Exception("connect_to_hublvol failed for LVS_18")), \ + patch.object(storage_node_ops, "_delete_replica_on_peer") as drp: + ret = storage_node_ops._relocate_one_replica(removed, "stranded", "secondary") + + self.assertFalse(ret) + drp.assert_not_called() + self.assertEqual(occupant.secondary_node_id, "x") + self.assertEqual(x.lvstore_stack_secondary, "occupant") + self.assertEqual(stranded.secondary_node_id, "n1") + + def test_relocate_via_splice_resumes_teardown_after_crash_between_writes(self): + # occupant's move was already built + committed by a PRIOR attempt + # (forward pointer already points at stranded), but the process + # crashed before the old copy on x was torn down -- x's backref is + # still stale. A retry must skip re-building (already done) and go + # straight to finishing the teardown, without erroring. + cl = _cluster() + removed = _node("n1", stack_secondary="stranded") + stranded = _node("stranded", secondary_id="n1", lvstore="LVS_stranded") + occupant = _node("occupant", secondary_id="stranded", lvstore="LVS_occupant") + x = _node("x", stack_secondary="occupant") # stale -- not yet cleared + db = FakeDB(cl, [removed, stranded, occupant, x]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="x"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + return_value=True) as rec, \ + patch.object(storage_node_ops, "_delete_replica_on_peer") as drp: + ret = storage_node_ops._relocate_one_replica(removed, "stranded", "secondary") + + self.assertTrue(ret) + drp.assert_called_once() # the deferred teardown finally runs + self.assertEqual(stranded.secondary_node_id, "x") + self.assertEqual(x.lvstore_stack_secondary, "stranded") + # rebuild still runs once for stranded's own claim on x -- the + # occupant's rebuild was already done, so it must NOT run again. + self.assertEqual(rec.call_count, 1) + def test_relocate_via_splice_own_rebuild_failure_after_occupant_moved(self): # occupant's move succeeds (evicted + rebuilt on stranded), but # stranded's own rebuild on the freed slot x fails. From 1d505e1aebfd0610681afcd9ca2d16cd4912be90 Mon Sep 17 00:00:00 2001 From: wmousa Date: Mon, 10 Aug 2026 23:45:10 +0200 Subject: [PATCH 8/9] fix(node-removal): stop a transient RPC/DNS blip from killing phase 5, and stop masking an incomplete phase 5 on retry Two related bugs found live-testing FD-aware node removal's second (splice-fallback) path on a real cluster: 1. _connect_to_remote_jm_devs' fallback bdev-existence poll called rpc_client.get_bdevs() unguarded, right after the primary connect_device() failure had already been correctly degraded (logged, not raised). A transient DNS/RPC blip against the connecting peer's own SPDK-proxy hostname hit that second call too, but this one propagated -- raising RPCException out of _decommission_node_devices and killing the whole node-removal task. Now wrapped in a bounded retry (3 attempts, 1s apart, tenacity Retrying/RetryError, matching the existing pattern in tasks_runner_lvol_migration.py) so a blip that clears within a few seconds is caught transparently; only once that's exhausted does it degrade to "this JM not connected" (self-heals later via the periodic health-check service's topology-diff sweep) instead of raising. 2. node_removal_orchestrate's top-of-function guard treated `status == REMOVED` as "fully done" and returned True immediately. But phase 4 flips that status *before* phase 5 (device/JM decommission) runs -- so if phase 5 raised (as in bug 1, before the retry existed) after phase 4 had already committed, every resumed attempt hit this guard and reported "Node removed" without phase 5 ever actually completing. Now only phases 1/3a/3b/4 are skipped on resume; phase 5 always (re)runs -- it's already idempotent, so this is a no-op once it has genuinely finished. Observed live: an RPC connection error mid phase-5 JM reassignment left a peer's LVS un-rebuilt on its new host (bdev_lvol_get_lvstores "No such device") while the task still reported done, and the cluster cycled IN_ACTIVATION <-> SUSPENDED. Adds: - TestNodeRemovalOrchestrateResumesPhase5 (3 tests) - TestConnectToRemoteJmDevsDegradesOnRpcException (4 tests, incl. one verifying the bounded retry actually recovers a blip that clears within budget, and one verifying it still degrades gracefully once exhausted) Co-Authored-By: Claude Sonnet 5 --- simplyblock_core/storage_node_ops.py | 124 ++++++++++++------ tests/unit/test_node_removal.py | 185 ++++++++++++++++++++++++++- 2 files changed, 265 insertions(+), 44 deletions(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index f3649c6abf..0342c8c8b0 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -2,6 +2,7 @@ import copy import datetime import json +import logging import math import platform import socket @@ -19,6 +20,7 @@ import docker from docker.types import LogConfig from pydantic import SecretStr +from tenacity import RetryError, Retrying, before_sleep_log, retry_if_exception_type, stop_after_attempt, wait_fixed from simplyblock_core import constants, scripts, distr_controller, cluster_ops from simplyblock_core import utils @@ -2144,17 +2146,42 @@ def _connect_to_remote_jm_devs(this_node: StorageNode, jm_ids=None, only_node_id # 5s for one that can never appear. During whole-cluster recovery the # 10x0.5s wait ran per dead peer JM (~30 of them), adding minutes to # every restart attempt (2026-07-13). - for _ in range(1 if connect_failed else 10): - if remote_device.remote_bdev and rpc_client.get_bdevs(remote_device.remote_bdev): - break - if rpc_client.get_bdevs(expected_bdev): - remote_device.remote_bdev = expected_bdev - break - time.sleep(0.5) - if not remote_device.remote_bdev and org_dev.get_id() in existing_remote_jm_devices: - existing_remote_device = existing_remote_jm_devices[org_dev.get_id()] - if existing_remote_device.remote_bdev and rpc_client.get_bdevs(existing_remote_device.remote_bdev): - remote_device.remote_bdev = existing_remote_device.remote_bdev + def _poll_for_remote_jm_bdev(): + for _ in range(1 if connect_failed else 10): + if remote_device.remote_bdev and rpc_client.get_bdevs(remote_device.remote_bdev): + return + if rpc_client.get_bdevs(expected_bdev): + remote_device.remote_bdev = expected_bdev + return + time.sleep(0.5) + if not remote_device.remote_bdev and org_dev.get_id() in existing_remote_jm_devices: + existing_remote_device = existing_remote_jm_devices[org_dev.get_id()] + if existing_remote_device.remote_bdev and rpc_client.get_bdevs(existing_remote_device.remote_bdev): + remote_device.remote_bdev = existing_remote_device.remote_bdev + + try: + # Bounded retry: a transient RPC/DNS blip against this_node's + # own proxy (the same one connect_device just hit above) is + # given a few seconds to clear before giving up. Same + # degrade-not-crash rationale as the connect_device catch + # above -- only RPCException (the transport-level failure) is + # retried; anything else propagates immediately. + Retrying( + stop=stop_after_attempt(3), + wait=wait_fixed(1), + retry=retry_if_exception_type(RPCException), + before_sleep=before_sleep_log(logger, logging.WARNING), + )(_poll_for_remote_jm_bdev) + except RetryError as e: + # Still failing after 3 attempts -- degrade to "this JM not + # connected" instead of aborting the whole node-removal / + # restart operation (2026-08-10 incident: this exact call + # raised uncaught and killed a node-removal task mid phase 5, + # leaving a peer's lvstore un-rebuilt while the task still + # reported "done"). + logger.warning( + f'get_bdevs kept failing while polling for {expected_bdev} ' + f'on {this_node.get_id()} after 3 attempts: {e}') if not remote_device.remote_bdev: logger.error(f"Failed to connect to remote JM device {org_dev.alceml_name}") continue @@ -3716,8 +3743,18 @@ def node_removal_orchestrate(node_id, force_remove=False): logger.error(f"node_removal_orchestrate: node {node_id} not found") return False - if snode.status == StorageNode.STATUS_REMOVED: - return True + # Phase 4 (below) flips status to REMOVED *before* phase 5 (device/JM + # decommission) runs -- so "status == REMOVED" means phases 1/3a/3b/4 + # committed, NOT that removal is fully done. A bare `return True` here + # would let a transient failure inside phase 5 (e.g. an RPC error + # against a peer) get permanently masked: the retry re-enters, hits this + # guard, and reports "done" forever without phase 5 ever completing + # (2026-08-10 incident: a mid-phase-5 RPC error left a peer's lvstore + # un-rebuilt while the task reported "Node removed"). Only phases + # 1/3a/3b/4 are skipped below when already_removed; phase 5 always + # runs and is itself idempotent (skips devices/JM already migrated), so + # resuming it here is a no-op once it has genuinely finished. + already_removed = snode.status == StorageNode.STATUS_REMOVED # Node removal is a recognised restart-phase owner: phase 3b relocates # replicas onto an ONLINE target and sets a restart phase there, which @@ -3730,40 +3767,43 @@ def node_removal_orchestrate(node_id, force_remove=False): prev_cluster_status = cluster.status cluster_ops.set_cluster_status(cluster.get_id(), Cluster.STATUS_IN_SHRINK) try: - # Phase 1 — shut the node down (graceful). Skipped on re-entry. - if snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: - logger.info(f"[REMOVAL] {node_id}: phase 1 — shutdown") - ret = shutdown_storage_node(node_id, force=force_remove) - if isinstance(ret, tuple): - ret, reason = ret - if not ret: - logger.error(f"[REMOVAL] {node_id}: shutdown failed: {reason}") + if not already_removed: + # Phase 1 — shut the node down (graceful). Skipped on re-entry. + if snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: + logger.info(f"[REMOVAL] {node_id}: phase 1 — shutdown") + ret = shutdown_storage_node(node_id, force=force_remove) + if isinstance(ret, tuple): + ret, reason = ret + if not ret: + logger.error(f"[REMOVAL] {node_id}: shutdown failed: {reason}") + return False + elif not ret: + logger.error(f"[REMOVAL] {node_id}: shutdown failed") return False - elif not ret: - logger.error(f"[REMOVAL] {node_id}: shutdown failed") - return False - snode = db_controller.get_storage_node_by_id(node_id) + snode = db_controller.get_storage_node_by_id(node_id) - # Phase 3a — tear down the (empty) secondary/tertiary replicas of THIS - # node's own primary LVS, on the peers that host them (Case A). - logger.info(f"[REMOVAL] {node_id}: phase 3a — tear down own replicas") - if not _teardown_replicas_of_primary(snode): - return False + # Phase 3a — tear down the (empty) secondary/tertiary replicas of THIS + # node's own primary LVS, on the peers that host them (Case A). + logger.info(f"[REMOVAL] {node_id}: phase 3a — tear down own replicas") + if not _teardown_replicas_of_primary(snode): + return False - # Phase 3b — relocate replicas this node hosts for OTHER primaries (Case B). - logger.info(f"[REMOVAL] {node_id}: phase 3b — relocate hosted replicas") - if not _relocate_replicas_hosted_on(snode): - return False + # Phase 3b — relocate replicas this node hosts for OTHER primaries (Case B). + logger.info(f"[REMOVAL] {node_id}: phase 3b — relocate hosted replicas") + if not _relocate_replicas_hosted_on(snode): + return False - # Phase 5 — finalize (swarm leave, gpt cleanup) and flip to removed. - logger.info(f"[REMOVAL] {node_id}: phase 4 — finalize") - _finalize_node_removal(snode) - set_node_status(node_id, StorageNode.STATUS_REMOVED, caused_by="remove") - snode = db_controller.get_storage_node_by_id(node_id) - # storage_events.snode_status_change( - # snode, StorageNode.STATUS_REMOVED, StorageNode.STATUS_IN_REMOVAL, caused_by="remove") + # Phase 4 — finalize (swarm leave, gpt cleanup) and flip to removed. + logger.info(f"[REMOVAL] {node_id}: phase 4 — finalize") + _finalize_node_removal(snode) + set_node_status(node_id, StorageNode.STATUS_REMOVED, caused_by="remove") + snode = db_controller.get_storage_node_by_id(node_id) + # storage_events.snode_status_change( + # snode, StorageNode.STATUS_REMOVED, StorageNode.STATUS_IN_REMOVAL, caused_by="remove") - # Phase 4 — remove + fail devices, then wait for failure-migration to finish. + # Phase 5 — remove + fail devices, then wait for failure-migration to + # finish. Always attempted, even on resume after status already + # flipped to REMOVED -- see the already_removed comment above. logger.info(f"[REMOVAL] {node_id}: phase 5 — devices remove/fail/migrate") if not _decommission_node_devices(snode): return False diff --git a/tests/unit/test_node_removal.py b/tests/unit/test_node_removal.py index 4b02b09111..0f202452f0 100644 --- a/tests/unit/test_node_removal.py +++ b/tests/unit/test_node_removal.py @@ -17,13 +17,13 @@ """ import unittest -from unittest.mock import MagicMock, patch +from unittest.mock import DEFAULT, MagicMock, patch from simplyblock_core import storage_node_ops from simplyblock_core.models.storage_node import StorageNode from simplyblock_core.models.nvme_device import NVMeDevice, JMDevice from simplyblock_core.models.cluster import Cluster -from simplyblock_core.rpc_client import RPCConnectionError +from simplyblock_core.rpc_client import RPCConnectionError, RPCException # --------------------------------------------------------------------------- @@ -804,6 +804,187 @@ def test_complete_when_all_migrated(self): dc.device_remove.assert_not_called() +# --------------------------------------------------------------------------- +# node_removal_orchestrate — phase-5 resume gap +# +# Phase 4 flips node status to REMOVED *before* phase 5 (device/JM +# decommission) runs. "status == REMOVED" therefore means phases 1/3a/3b/4 +# committed, NOT that removal is fully done -- a resumed attempt must still +# (re)run phase 5 rather than short-circuiting to "done" (2026-08-10 +# incident: an RPC error mid phase 5 left a peer's lvstore un-rebuilt while +# the task still reported "Node removed"). +# --------------------------------------------------------------------------- + +class TestNodeRemovalOrchestrateResumesPhase5(unittest.TestCase): + + def _patch_all(self): + return patch.multiple( + storage_node_ops, + DBController=DEFAULT, + cluster_ops=DEFAULT, + shutdown_storage_node=DEFAULT, + _teardown_replicas_of_primary=DEFAULT, + _relocate_replicas_hosted_on=DEFAULT, + _finalize_node_removal=DEFAULT, + set_node_status=DEFAULT, + _decommission_node_devices=DEFAULT, + ) + + def test_already_removed_skips_phases_1_to_4_but_reruns_phase5(self): + cl = _cluster() + node = _node("n1", status=StorageNode.STATUS_REMOVED) + db = FakeDB(cl, [node]) + with self._patch_all() as mocks: + mocks["DBController"].return_value = db + mocks["_decommission_node_devices"].return_value = True + ret = storage_node_ops.node_removal_orchestrate("n1") + + self.assertTrue(ret) + mocks["shutdown_storage_node"].assert_not_called() + mocks["_teardown_replicas_of_primary"].assert_not_called() + mocks["_relocate_replicas_hosted_on"].assert_not_called() + mocks["_finalize_node_removal"].assert_not_called() + mocks["set_node_status"].assert_not_called() + mocks["_decommission_node_devices"].assert_called_once_with(node) + + def test_already_removed_reports_incomplete_if_phase5_fails_again(self): + # The regression this guards: a prior attempt raised mid phase 5 + # after the status flip had already committed. The retry must + # actually retry phase 5, not silently report done because status + # already reads REMOVED. + cl = _cluster() + node = _node("n1", status=StorageNode.STATUS_REMOVED) + db = FakeDB(cl, [node]) + with self._patch_all() as mocks: + mocks["DBController"].return_value = db + mocks["_decommission_node_devices"].return_value = False + ret = storage_node_ops.node_removal_orchestrate("n1") + + self.assertFalse(ret) + mocks["_decommission_node_devices"].assert_called_once_with(node) + + def test_fresh_removal_still_runs_all_phases_then_phase5(self): + # Regression guard the other way: a from-scratch removal (status + # still ONLINE) must not skip phases 1/3a/3b/4. + cl = _cluster() + node = _node("n1", status=StorageNode.STATUS_ONLINE) + db = FakeDB(cl, [node]) + with self._patch_all() as mocks: + mocks["DBController"].return_value = db + mocks["shutdown_storage_node"].return_value = True + mocks["_teardown_replicas_of_primary"].return_value = True + mocks["_relocate_replicas_hosted_on"].return_value = True + mocks["_decommission_node_devices"].return_value = True + ret = storage_node_ops.node_removal_orchestrate("n1") + + self.assertTrue(ret) + mocks["shutdown_storage_node"].assert_called_once() + mocks["_teardown_replicas_of_primary"].assert_called_once() + mocks["_relocate_replicas_hosted_on"].assert_called_once() + mocks["_finalize_node_removal"].assert_called_once() + mocks["set_node_status"].assert_called_once_with( + "n1", StorageNode.STATUS_REMOVED, caused_by="remove") + mocks["_decommission_node_devices"].assert_called_once() + + +# --------------------------------------------------------------------------- +# _connect_to_remote_jm_devs — bounded retry + degrade-not-crash on a +# transient RPC/DNS failure during the fallback bdev-existence poll +# +# The primary connect_device() failure already degrades gracefully (logs +# "Failed to connect to ...", sets connect_failed=True). The get_bdevs() +# poll called right after it, against the same rpc_client, hits the +# identical transport and gets a bounded retry (3 attempts, 1s apart) to +# ride out a DNS blip; only once that's exhausted does it degrade to "this +# JM not connected" instead of raising -- 2026-08-10 incident: this exact +# call raised RPCException uncaught and killed a node-removal task mid +# phase 5. +# --------------------------------------------------------------------------- + +class TestConnectToRemoteJmDevsDegradesOnRpcException(unittest.TestCase): + + def _owner_setup(self, this_node_id="this-node"): + jm_dev = JMDevice() + jm_dev.uuid = "jm-owner" + jm_dev.jm_bdev = "jm_owner_bdev" + jm_dev.status = NVMeDevice.STATUS_ONLINE + + owner_node = MagicMock(spec=StorageNode) + owner_node.get_id = MagicMock(return_value="owner-node") + owner_node.status = StorageNode.STATUS_ONLINE + owner_node.jm_device = jm_dev + + this_node = MagicMock(spec=StorageNode) + this_node.get_id = MagicMock(return_value=this_node_id) + this_node.jm_ids = [] + this_node.lvstore_stack_secondary = "" + this_node.lvstore_stack_tertiary = "" + this_node.remote_jm_devices = [] + rpc_client = MagicMock() + this_node.rpc_client = MagicMock(return_value=rpc_client) + + db = MagicMock() + db.get_jm_device_by_id.return_value = jm_dev + db.get_storage_nodes.return_value = [owner_node] + + return this_node, rpc_client, db + + def test_get_bdevs_rpc_exception_exhausts_retries_and_does_not_raise(self): + # Persistent failure (all 3 bounded-retry attempts fail): must + # still degrade, not raise -- this is the exact call chain that + # took down a live node-removal task before the retry was added. + this_node, rpc_client, db = self._owner_setup() + rpc_client.get_bdevs.side_effect = RPCException("connection error") + + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "connect_device", + side_effect=RPCException("connection error")): + result = storage_node_ops._connect_to_remote_jm_devs( + this_node, jm_ids=["jm-owner"]) + + self.assertEqual(result, []) + # 3 bounded-retry attempts, one get_bdevs call each (remote_bdev + # is empty so the first branch short-circuits without calling). + self.assertEqual(rpc_client.get_bdevs.call_count, 3) + + def test_transient_failure_recovers_on_retry(self): + # A blip that clears within the retry budget must be caught, not + # just tolerated -- the whole point of adding the bounded retry + # instead of degrading on the very first failure. + this_node, rpc_client, db = self._owner_setup() + rpc_client.get_bdevs.side_effect = [ + RPCException("connection error"), + RPCException("connection error"), + {"name": "remote_jm_owner_bdevn1"}, + ] + + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "connect_device", + side_effect=RPCException("connection error")), \ + patch.object(storage_node_ops.time, "sleep"): + result = storage_node_ops._connect_to_remote_jm_devs( + this_node, jm_ids=["jm-owner"]) + + self.assertEqual(len(result), 1) + self.assertEqual(result[0].remote_bdev, "remote_jm_owner_bdevn1") + self.assertEqual(rpc_client.get_bdevs.call_count, 3) + + def test_transient_failure_does_not_block_a_clean_connect(self): + # Once the blip has cleared, the same code path must still succeed + # normally -- the new guard must not swallow a real success too. + this_node, rpc_client, db = self._owner_setup() + rpc_client.get_bdevs.return_value = {"name": "remote_jm_owner_bdevn1"} + + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "connect_device", + return_value="remote_jm_owner_bdevn1"): + result = storage_node_ops._connect_to_remote_jm_devs( + this_node, jm_ids=["jm-owner"]) + + self.assertEqual(len(result), 1) + self.assertEqual(result[0].remote_bdev, "remote_jm_owner_bdevn1") + + class TestShrinkStatusDoesNotDeadlockRemoval(unittest.TestCase): """``node_removal_orchestrate`` holds ``Cluster.STATUS_IN_SHRINK`` for the duration of an attempt, so that the restart phases its replica relocation From 11564e250f96339b37be8373a7c880cc7c6a22f9 Mon Sep 17 00:00:00 2001 From: wmousa Date: Tue, 11 Aug 2026 16:55:38 +0200 Subject: [PATCH 9/9] fix(node-removal): skip already-removed peers with stale jm_ids in phase 5 get_storage_nodes_by_cluster_id returns every node regardless of status, including ones already REMOVED. The JM-device peer-reassignment loop in _decommission_node_devices never filtered on that: an earlier-removed node can still carry the currently-removed node's JM id in its own jm_ids (never cleared on ITS OWN removal), so a later removal's phase 5 would try to "fix" that dead node's JM connections using its own rpc_client -- which points at a pod that no longer exists and can never resolve, let alone connect. 2026-08-11 incident: removing node A, then later removing node B (whose JM device A used to reference) sent phase 5 chasing A's permanently-dead hostname (NameResolutionError -> uncaught RPCException, same class of failure as the bounded-retry fix targets, but that retry can't save a hostname that will never resolve). B's own devices never reached failed/failed_and_migrated because the crash happened before the device loop further down in the same function, and the task still reported "done" on the next attempt (status already REMOVED short-circuits phase 1-4, per the earlier phase-5-resume fix) -- leaving B's devices stuck at unavailable with no task left to retry them. Skip node.status == STATUS_REMOVED outright in that loop -- a removed node's own bookkeeping is dead weight, not something to reconnect. --- simplyblock_core/storage_node_ops.py | 11 +++++++++++ tests/unit/test_node_removal.py | 26 ++++++++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 0342c8c8b0..26df690461 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -4054,6 +4054,17 @@ def _decommission_node_devices(removed_node: StorageNode): device_controller.remove_jm_device(removed_node.jm_device.get_id(), force=True) # look for other nodes who use this JM and replace it for node in db_controller.get_storage_nodes_by_cluster_id(removed_node.cluster_id): + # get_storage_nodes_by_cluster_id returns every node regardless of + # status, including ones already REMOVED. An already-removed node + # can still carry the just-removed node's JM id in its own stale + # jm_ids (never cleared on ITS removal) -- without this guard we'd + # try to "fix" that dead node's JM connections using its own + # rpc_client, which points at a pod that no longer exists and can + # never resolve/connect (2026-08-11 incident: a prior removal's + # leftover jm_ids on 7b8hf sent a later removal's phase 5 chasing + # a permanently-dead hostname). + if node.status == StorageNode.STATUS_REMOVED: + continue if node.jm_ids and removed_node.jm_device.get_id() in node.jm_ids: node.jm_ids.remove(removed_node.jm_device.get_id()) jm_ids = get_sorted_ha_jms(node) diff --git a/tests/unit/test_node_removal.py b/tests/unit/test_node_removal.py index 0f202452f0..7105154ec9 100644 --- a/tests/unit/test_node_removal.py +++ b/tests/unit/test_node_removal.py @@ -803,6 +803,32 @@ def test_complete_when_all_migrated(self): self.assertTrue(ret) dc.device_remove.assert_not_called() + def test_skips_already_removed_peer_with_stale_jm_ids(self): + # 2026-08-11 incident: an earlier-removed node can still carry the + # currently-removed node's JM id in its own stale jm_ids (never + # cleared on ITS OWN removal) -- get_storage_nodes_by_cluster_id + # returns every node regardless of status, including removed ones. + # Must be skipped outright, not "fixed" via its own (permanently + # dead) rpc_client. + cl = _cluster() + removed = _node("n1", n_devices=0, with_jm=True) + removed.jm_ids = [] + stale_peer = _node("stale-peer", status=StorageNode.STATUS_REMOVED) + stale_peer.jm_ids = [removed.jm_device.get_id()] + db = FakeDB(cl, [removed, stale_peer]) + dc = MagicMock() + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "device_controller", dc), \ + patch.object(storage_node_ops, "_connect_to_remote_jm_devs") as connect_mock: + ret = storage_node_ops._decommission_node_devices(removed) + + self.assertTrue(ret) + dc.remove_jm_device.assert_called_once() + connect_mock.assert_not_called() + # The stale peer's own bookkeeping is left alone -- it's dead, not "fixed". + self.assertEqual(stale_peer.jm_ids, [removed.jm_device.get_id()]) + stale_peer.write_to_db.assert_not_called() + # --------------------------------------------------------------------------- # node_removal_orchestrate — phase-5 resume gap