refactor: move ecstore owner layout modules (#3932)

crates/ecstore/src/cluster/control_plane.rs

@@ -0,0 +1,635 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use crate::storage_api_contracts::topology::{
+    CapabilityStatus, DiskCapabilities, TopologyCapabilities, TopologyDisk, TopologyLabels, TopologyPool, TopologySet,
+    TopologySnapshot,
+};
+
+use crate::{
+    endpoints::EndpointServerPools,
+    layout::endpoint::{Endpoint, EndpointType},
+};
+
+const ENDPOINT_TYPE_LABEL: &str = "endpoint_type";
+const LOCAL_ENDPOINT_LABEL: &str = "local";
+const LOCAL_NODE_ID: &str = "local";
+const STORAGE_MEDIA_NOT_REPORTED: &str = "storage media not reported by endpoints";
+const FAILURE_DOMAIN_NOT_REPORTED: &str = "failure domain labels not reported by endpoints";
+const NUMA_NOT_WIRED: &str = "NUMA topology not wired into runtime";
+const PROFILING_NOT_WIRED: &str = "profiling capability not wired into ECStore";
+const PEER_HEALTH_NOT_REPORTED: &str = "peer health not reported by endpoints";
+
+#[derive(Debug, Clone)]
+pub struct ClusterControlPlane {
+    endpoint_pools: EndpointServerPools,
+}
+
+impl ClusterControlPlane {
+    pub fn new(endpoint_pools: EndpointServerPools) -> Self {
+        Self { endpoint_pools }
+    }
+
+    pub fn topology_snapshot(&self) -> TopologySnapshot {
+        topology_snapshot_from_endpoint_pools(&self.endpoint_pools)
+    }
+
+    pub fn membership_snapshot(&self) -> ClusterMembershipSnapshot {
+        membership_snapshot_from_endpoint_pools(&self.endpoint_pools)
+    }
+
+    pub fn pool_state_snapshot(&self) -> ClusterPoolStateSnapshot {
+        pool_state_snapshot_from_endpoint_pools(&self.endpoint_pools)
+    }
+
+    pub fn local_node_storage_snapshot(&self) -> ClusterLocalNodeStorageSnapshot {
+        let membership = self.membership_snapshot();
+        local_node_storage_snapshot_from_membership(&membership)
+    }
+
+    pub fn peer_health_snapshot(&self) -> ClusterPeerHealthSnapshot {
+        let membership = self.membership_snapshot();
+        peer_health_snapshot_from_membership(&membership)
+    }
+
+    pub fn read_snapshot(&self) -> ClusterControlPlaneSnapshot {
+        let membership = self.membership_snapshot();
+
+        ClusterControlPlaneSnapshot {
+            topology: self.topology_snapshot(),
+            pool_state: self.pool_state_snapshot(),
+            local_storage: local_node_storage_snapshot_from_membership(&membership),
+            peer_health: peer_health_snapshot_from_membership(&membership),
+            membership,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClusterControlPlaneSnapshot {
+    pub topology: TopologySnapshot,
+    pub pool_state: ClusterPoolStateSnapshot,
+    pub local_storage: ClusterLocalNodeStorageSnapshot,
+    pub peer_health: ClusterPeerHealthSnapshot,
+    pub membership: ClusterMembershipSnapshot,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ClusterMembershipSnapshot {
+    pub nodes: Vec<ClusterNodeMembership>,
+    pub drives: Vec<ClusterDriveMembership>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClusterNodeMembership {
+    pub node_id: String,
+    pub grid_host: String,
+    pub is_local: bool,
+    pub pools: Vec<usize>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClusterDriveMembership {
+    pub pool_index: usize,
+    pub set_index: usize,
+    pub disk_index: usize,
+    pub node_id: String,
+    pub is_local: bool,
+    pub endpoint_type: ClusterEndpointType,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ClusterPoolStateSnapshot {
+    pub pools: Vec<ClusterPoolState>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClusterPoolState {
+    pub pool_index: usize,
+    pub set_count: usize,
+    pub drives_per_set: usize,
+    pub endpoint_count: usize,
+    pub local_drive_count: usize,
+    pub remote_drive_count: usize,
+    pub legacy: bool,
+    pub endpoint_types: Vec<ClusterEndpointType>,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ClusterLocalNodeStorageSnapshot {
+    pub nodes: Vec<ClusterLocalNodeStorage>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClusterLocalNodeStorage {
+    pub node_id: String,
+    pub pools: Vec<usize>,
+    pub drive_count: usize,
+    pub path_drive_count: usize,
+    pub url_drive_count: usize,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ClusterPeerHealthSnapshot {
+    pub peers: Vec<ClusterPeerHealth>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClusterPeerHealth {
+    pub node_id: String,
+    pub is_local: bool,
+    pub status: CapabilityStatus,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub enum ClusterEndpointType {
+    Path,
+    Url,
+}
+
+pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot {
+    topology_snapshot_from_endpoint_pools_with_capabilities(
+        endpoint_pools,
+        default_topology_capabilities(),
+        default_disk_capabilities(),
+    )
+}
+
+pub fn topology_snapshot_from_endpoint_pools_with_capabilities(
+    endpoint_pools: &EndpointServerPools,
+    capabilities: TopologyCapabilities,
+    disk_capabilities: DiskCapabilities,
+) -> TopologySnapshot {
+    TopologySnapshot {
+        pools: endpoint_pools
+            .as_ref()
+            .iter()
+            .enumerate()
+            .map(|(pool_index, pool)| {
+                let sets =
+                    topology_sets_from_endpoints(pool_index, pool.drives_per_set, pool.endpoints.as_ref(), &disk_capabilities);
+                TopologyPool {
+                    pool_index,
+                    pool_id: None,
+                    labels: TopologyLabels::default(),
+                    sets,
+                }
+            })
+            .collect(),
+        capabilities,
+    }
+}
+
+pub fn membership_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> ClusterMembershipSnapshot {
+    let mut nodes = BTreeMap::<String, ClusterNodeMembership>::new();
+    let mut drives = Vec::new();
+
+    for (pool_index, pool) in endpoint_pools.as_ref().iter().enumerate() {
+        for (endpoint_index, endpoint) in pool.endpoints.as_ref().iter().enumerate() {
+            let (set_index, disk_index) = endpoint_indices(pool_index, endpoint_index, pool.drives_per_set, endpoint);
+            let node_id = endpoint_node_id(endpoint);
+
+            match nodes.entry(node_id.clone()) {
+                std::collections::btree_map::Entry::Vacant(entry) => {
+                    entry.insert(ClusterNodeMembership {
+                        node_id: node_id.clone(),
+                        grid_host: endpoint.grid_host(),
+                        is_local: endpoint.is_local,
+                        pools: vec![pool_index],
+                    });
+                }
+                std::collections::btree_map::Entry::Occupied(mut entry) => {
+                    let node = entry.get_mut();
+                    node.is_local |= endpoint.is_local;
+                    if !node.pools.contains(&pool_index) {
+                        node.pools.push(pool_index);
+                    }
+                }
+            }
+
+            drives.push(ClusterDriveMembership {
+                pool_index,
+                set_index,
+                disk_index,
+                node_id,
+                is_local: endpoint.is_local,
+                endpoint_type: endpoint_type(endpoint),
+            });
+        }
+    }
+
+    ClusterMembershipSnapshot {
+        nodes: nodes.into_values().collect(),
+        drives,
+    }
+}
+
+pub fn pool_state_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> ClusterPoolStateSnapshot {
+    ClusterPoolStateSnapshot {
+        pools: endpoint_pools
+            .as_ref()
+            .iter()
+            .enumerate()
+            .map(|(pool_index, pool)| {
+                let mut endpoint_types = BTreeSet::new();
+                let mut local_drive_count = 0;
+
+                for endpoint in pool.endpoints.as_ref() {
+                    endpoint_types.insert(endpoint_type(endpoint));
+                    if endpoint.is_local {
+                        local_drive_count += 1;
+                    }
+                }
+
+                let endpoint_count = pool.endpoints.as_ref().len();
+                ClusterPoolState {
+                    pool_index,
+                    set_count: pool.set_count,
+                    drives_per_set: pool.drives_per_set,
+                    endpoint_count,
+                    local_drive_count,
+                    remote_drive_count: endpoint_count.saturating_sub(local_drive_count),
+                    legacy: pool.legacy,
+                    endpoint_types: endpoint_types.into_iter().collect(),
+                }
+            })
+            .collect(),
+    }
+}
+
+pub fn local_node_storage_snapshot_from_membership(membership: &ClusterMembershipSnapshot) -> ClusterLocalNodeStorageSnapshot {
+    ClusterLocalNodeStorageSnapshot {
+        nodes: membership
+            .nodes
+            .iter()
+            .filter(|node| node.is_local)
+            .map(|node| {
+                let mut path_drive_count = 0;
+                let mut url_drive_count = 0;
+
+                for drive in membership.drives.iter().filter(|drive| drive.node_id == node.node_id) {
+                    match drive.endpoint_type {
+                        ClusterEndpointType::Path => path_drive_count += 1,
+                        ClusterEndpointType::Url => url_drive_count += 1,
+                    }
+                }
+
+                ClusterLocalNodeStorage {
+                    node_id: node.node_id.clone(),
+                    pools: node.pools.clone(),
+                    drive_count: path_drive_count + url_drive_count,
+                    path_drive_count,
+                    url_drive_count,
+                }
+            })
+            .collect(),
+    }
+}
+
+pub fn peer_health_snapshot_from_membership(membership: &ClusterMembershipSnapshot) -> ClusterPeerHealthSnapshot {
+    ClusterPeerHealthSnapshot {
+        peers: membership
+            .nodes
+            .iter()
+            .map(|node| ClusterPeerHealth {
+                node_id: node.node_id.clone(),
+                is_local: node.is_local,
+                status: CapabilityStatus::unknown().with_reason(PEER_HEALTH_NOT_REPORTED),
+            })
+            .collect(),
+    }
+}
+
+fn topology_sets_from_endpoints(
+    pool_index: usize,
+    drives_per_set: usize,
+    endpoints: &[Endpoint],
+    disk_capabilities: &DiskCapabilities,
+) -> Vec<TopologySet> {
+    let mut sets = BTreeMap::<usize, Vec<TopologyDisk>>::new();
+
+    for (endpoint_index, endpoint) in endpoints.iter().enumerate() {
+        let (set_index, disk_index) = endpoint_indices(pool_index, endpoint_index, drives_per_set, endpoint);
+        sets.entry(set_index).or_default().push(topology_disk_from_endpoint(
+            pool_index,
+            set_index,
+            disk_index,
+            endpoint,
+            disk_capabilities,
+        ));
+    }
+
+    sets.into_iter()
+        .map(|(set_index, mut disks)| {
+            disks.sort_by_key(|disk| disk.disk_index);
+            TopologySet {
+                pool_index,
+                set_index,
+                set_id: None,
+                labels: TopologyLabels::default(),
+                disks,
+            }
+        })
+        .collect()
+}
+
+fn topology_disk_from_endpoint(
+    pool_index: usize,
+    set_index: usize,
+    disk_index: usize,
+    endpoint: &Endpoint,
+    disk_capabilities: &DiskCapabilities,
+) -> TopologyDisk {
+    TopologyDisk {
+        pool_index,
+        set_index,
+        disk_index,
+        disk_id: endpoint_disk_id(endpoint),
+        labels: endpoint_labels(endpoint),
+        capabilities: disk_capabilities.clone(),
+    }
+}
+
+fn endpoint_indices(pool_index: usize, endpoint_index: usize, drives_per_set: usize, endpoint: &Endpoint) -> (usize, usize) {
+    let safe_drives_per_set = drives_per_set.max(1);
+    let set_index = non_negative_index(endpoint.set_idx).unwrap_or(endpoint_index / safe_drives_per_set);
+    let disk_index = non_negative_index(endpoint.disk_idx).unwrap_or(endpoint_index % safe_drives_per_set);
+
+    debug_assert_eq!(non_negative_index(endpoint.pool_idx).unwrap_or(pool_index), pool_index);
+    (set_index, disk_index)
+}
+
+fn endpoint_disk_id(endpoint: &Endpoint) -> Option<String> {
+    let host_port = endpoint.host_port();
+    if host_port.is_empty() { None } else { Some(host_port) }
+}
+
+fn endpoint_node_id(endpoint: &Endpoint) -> String {
+    endpoint_disk_id(endpoint).unwrap_or_else(|| LOCAL_NODE_ID.to_owned())
+}
+
+fn endpoint_labels(endpoint: &Endpoint) -> TopologyLabels {
+    let mut additional = BTreeMap::new();
+    additional.insert(ENDPOINT_TYPE_LABEL.to_owned(), endpoint_type_label(endpoint).to_owned());
+    additional.insert(LOCAL_ENDPOINT_LABEL.to_owned(), endpoint.is_local.to_string());
+
+    TopologyLabels {
+        node: Some(endpoint_node_id(endpoint)),
+        additional,
+        ..TopologyLabels::default()
+    }
+}
+
+fn endpoint_type(endpoint: &Endpoint) -> ClusterEndpointType {
+    match endpoint.get_type() {
+        EndpointType::Path => ClusterEndpointType::Path,
+        EndpointType::Url => ClusterEndpointType::Url,
+    }
+}
+
+fn endpoint_type_label(endpoint: &Endpoint) -> &'static str {
+    match endpoint_type(endpoint) {
+        ClusterEndpointType::Path => "path",
+        ClusterEndpointType::Url => "url",
+    }
+}
+
+fn non_negative_index(index: i32) -> Option<usize> {
+    usize::try_from(index).ok()
+}
+
+fn default_topology_capabilities() -> TopologyCapabilities {
+    TopologyCapabilities {
+        profiling: CapabilityStatus::unknown().with_reason(PROFILING_NOT_WIRED),
+        numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED),
+        failure_domain_labels: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED),
+        media_labels: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED),
+    }
+}
+
+fn default_disk_capabilities() -> DiskCapabilities {
+    DiskCapabilities {
+        media_type: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED),
+        failure_domain: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED),
+        numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED),
+        profiling: CapabilityStatus::unknown().with_reason(PROFILING_NOT_WIRED),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::BTreeSet;
+
+    use crate::endpoints::{Endpoints, PoolEndpoints};
+
+    #[test]
+    fn topology_snapshot_maps_endpoint_sets_without_local_paths() {
+        let endpoint_pools = sample_path_endpoint_pools();
+        let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools);
+
+        assert_eq!(snapshot.pools.len(), 1);
+        assert_eq!(snapshot.pools[0].sets.len(), 2);
+        assert_eq!(snapshot.pools[0].sets[0].disks.len(), 2);
+        assert_eq!(snapshot.pools[0].sets[1].disks.len(), 2);
+        assert_eq!(snapshot.pools[0].sets[1].disks[1].disk_index, 1);
+        assert_eq!(
+            snapshot.pools[0].sets[0].disks[0]
+                .labels
+                .additional
+                .get(ENDPOINT_TYPE_LABEL)
+                .map(String::as_str),
+            Some("path")
+        );
+        assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some(LOCAL_NODE_ID));
+
+        let encoded = serde_json::to_string(&snapshot).expect("serialize topology snapshot");
+        assert!(!encoded.contains("/tmp/rustfs-cluster-control-plane"));
+    }
+
+    #[test]
+    fn topology_snapshot_uses_url_hosts_as_disk_ids() {
+        let endpoint_pools = sample_url_endpoint_pools();
+        let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools);
+
+        assert_eq!(snapshot.pools[0].sets[0].disks[0].disk_id.as_deref(), Some("node1.example:9000"));
+        assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some("node1.example:9000"));
+        assert_eq!(
+            snapshot.pools[0].sets[0].disks[0]
+                .labels
+                .additional
+                .get(ENDPOINT_TYPE_LABEL)
+                .map(String::as_str),
+            Some("url")
+        );
+    }
+
+    #[test]
+    fn membership_snapshot_groups_nodes_and_drives() {
+        let endpoint_pools = sample_url_endpoint_pools();
+        let snapshot = membership_snapshot_from_endpoint_pools(&endpoint_pools);
+
+        assert_eq!(snapshot.nodes.len(), 2);
+        assert_eq!(snapshot.drives.len(), 4);
+        assert_eq!(snapshot.nodes[0].node_id, "node1.example:9000");
+        assert_eq!(snapshot.nodes[0].pools, vec![0]);
+        assert_eq!(snapshot.drives[2].node_id, "node2.example:9000");
+        assert_eq!(snapshot.drives[2].set_index, 1);
+        assert_eq!(snapshot.drives[2].endpoint_type, ClusterEndpointType::Url);
+    }
+
+    #[test]
+    fn pool_state_snapshot_counts_local_remote_drives_and_endpoint_types() {
+        let endpoint_pools = sample_mixed_endpoint_pools();
+        let snapshot = pool_state_snapshot_from_endpoint_pools(&endpoint_pools);
+
+        assert_eq!(snapshot.pools.len(), 1);
+        assert_eq!(snapshot.pools[0].set_count, 2);
+        assert_eq!(snapshot.pools[0].drives_per_set, 2);
+        assert_eq!(snapshot.pools[0].endpoint_count, 4);
+        assert_eq!(snapshot.pools[0].local_drive_count, 3);
+        assert_eq!(snapshot.pools[0].remote_drive_count, 1);
+        assert_eq!(
+            snapshot.pools[0].endpoint_types,
+            vec![ClusterEndpointType::Path, ClusterEndpointType::Url]
+        );
+    }
+
+    #[test]
+    fn local_node_storage_snapshot_keeps_only_local_drive_counts() {
+        let membership = membership_snapshot_from_endpoint_pools(&sample_mixed_endpoint_pools());
+        let snapshot = local_node_storage_snapshot_from_membership(&membership);
+
+        assert_eq!(snapshot.nodes.len(), 2);
+        assert_eq!(snapshot.nodes[0].node_id, LOCAL_NODE_ID);
+        assert_eq!(snapshot.nodes[0].drive_count, 2);
+        assert_eq!(snapshot.nodes[0].path_drive_count, 2);
+        assert_eq!(snapshot.nodes[0].url_drive_count, 0);
+        assert_eq!(snapshot.nodes[1].node_id, "node1.example:9000");
+        assert_eq!(snapshot.nodes[1].drive_count, 1);
+        assert_eq!(snapshot.nodes[1].path_drive_count, 0);
+        assert_eq!(snapshot.nodes[1].url_drive_count, 1);
+    }
+
+    #[test]
+    fn peer_health_snapshot_reports_static_unknown_status() {
+        let membership = membership_snapshot_from_endpoint_pools(&sample_url_endpoint_pools());
+        let snapshot = peer_health_snapshot_from_membership(&membership);
+
+        assert_eq!(snapshot.peers.len(), 2);
+        assert_eq!(snapshot.peers[0].node_id, "node1.example:9000");
+        assert!(snapshot.peers[0].is_local);
+        assert_eq!(snapshot.peers[0].status.reason.as_deref(), Some(PEER_HEALTH_NOT_REPORTED));
+        assert_eq!(snapshot.peers[1].node_id, "node2.example:9000");
+        assert!(!snapshot.peers[1].is_local);
+    }
+
+    #[test]
+    fn control_plane_read_snapshot_combines_topology_and_membership() {
+        let control_plane = ClusterControlPlane::new(sample_mixed_endpoint_pools());
+        let snapshot = control_plane.read_snapshot();
+        let node_ids = snapshot
+            .membership
+            .nodes
+            .iter()
+            .map(|node| node.node_id.as_str())
+            .collect::<BTreeSet<_>>();
+
+        assert_eq!(snapshot.topology.pools[0].sets.len(), 2);
+        assert_eq!(snapshot.pool_state.pools[0].endpoint_count, 4);
+        assert_eq!(snapshot.local_storage.nodes.len(), 2);
+        assert_eq!(snapshot.peer_health.peers.len(), 3);
+        assert_eq!(node_ids, BTreeSet::from([LOCAL_NODE_ID, "node1.example:9000", "node2.example:9000"]));
+    }
+
+    fn sample_path_endpoint_pools() -> EndpointServerPools {
+        let endpoints = (0..4)
+            .map(|index| {
+                let mut endpoint =
+                    Endpoint::try_from(format!("/tmp/rustfs-cluster-control-plane-{index}").as_str()).expect("local endpoint");
+                endpoint.set_pool_index(0);
+                endpoint.set_set_index(index / 2);
+                endpoint.set_disk_index(index % 2);
+                endpoint
+            })
+            .collect::<Vec<_>>();
+
+        EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 2,
+            drives_per_set: 2,
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: "/tmp/rustfs-cluster-control-plane-{0...3}".to_owned(),
+            platform: "OS: test | Arch: test".to_owned(),
+        }])
+    }
+
+    fn sample_url_endpoint_pools() -> EndpointServerPools {
+        let endpoints = (0..4)
+            .map(|index| {
+                let host = if index < 2 { "node1.example" } else { "node2.example" };
+                let mut endpoint =
+                    Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint");
+                endpoint.set_pool_index(0);
+                endpoint.set_set_index(index / 2);
+                endpoint.set_disk_index(index % 2);
+                endpoint.is_local = index < 2;
+                endpoint
+            })
+            .collect::<Vec<_>>();
+
+        EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 2,
+            drives_per_set: 2,
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: "http://node{1...2}.example:9000/export{0...3}".to_owned(),
+            platform: "OS: test | Arch: test".to_owned(),
+        }])
+    }
+
+    fn sample_mixed_endpoint_pools() -> EndpointServerPools {
+        let mut endpoints = Vec::new();
+
+        for index in 0..2 {
+            let mut endpoint =
+                Endpoint::try_from(format!("/tmp/rustfs-cluster-control-plane-{index}").as_str()).expect("local endpoint");
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(index);
+            endpoints.push(endpoint);
+        }
+
+        for index in 2..4 {
+            let host = if index == 2 { "node1.example" } else { "node2.example" };
+            let mut endpoint = Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint");
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(1);
+            endpoint.set_disk_index(index - 2);
+            endpoint.is_local = index == 2;
+            endpoints.push(endpoint);
+        }
+
+        EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 2,
+            drives_per_set: 2,
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: "mixed-test-endpoints".to_owned(),
+            platform: "OS: test | Arch: test".to_owned(),
+        }])
+    }
+}

crates/ecstore/src/cluster/mod.rs

@@ -12,624 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::{BTreeMap, BTreeSet};
+mod control_plane;
 
-use crate::storage_api_contracts::topology::{
-    CapabilityStatus, DiskCapabilities, TopologyCapabilities, TopologyDisk, TopologyLabels, TopologyPool, TopologySet,
-    TopologySnapshot,
+pub use control_plane::{
+    ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage,
+    ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth,
+    ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, local_node_storage_snapshot_from_membership,
+    membership_snapshot_from_endpoint_pools, peer_health_snapshot_from_membership, pool_state_snapshot_from_endpoint_pools,
+    topology_snapshot_from_endpoint_pools, topology_snapshot_from_endpoint_pools_with_capabilities,
 };
-
-use crate::{
-    endpoints::EndpointServerPools,
-    layout::endpoint::{Endpoint, EndpointType},
-};
-
-const ENDPOINT_TYPE_LABEL: &str = "endpoint_type";
-const LOCAL_ENDPOINT_LABEL: &str = "local";
-const LOCAL_NODE_ID: &str = "local";
-const STORAGE_MEDIA_NOT_REPORTED: &str = "storage media not reported by endpoints";
-const FAILURE_DOMAIN_NOT_REPORTED: &str = "failure domain labels not reported by endpoints";
-const NUMA_NOT_WIRED: &str = "NUMA topology not wired into runtime";
-const PROFILING_NOT_WIRED: &str = "profiling capability not wired into ECStore";
-const PEER_HEALTH_NOT_REPORTED: &str = "peer health not reported by endpoints";
-
-#[derive(Debug, Clone)]
-pub struct ClusterControlPlane {
-    endpoint_pools: EndpointServerPools,
-}
-
-impl ClusterControlPlane {
-    pub fn new(endpoint_pools: EndpointServerPools) -> Self {
-        Self { endpoint_pools }
-    }
-
-    pub fn topology_snapshot(&self) -> TopologySnapshot {
-        topology_snapshot_from_endpoint_pools(&self.endpoint_pools)
-    }
-
-    pub fn membership_snapshot(&self) -> ClusterMembershipSnapshot {
-        membership_snapshot_from_endpoint_pools(&self.endpoint_pools)
-    }
-
-    pub fn pool_state_snapshot(&self) -> ClusterPoolStateSnapshot {
-        pool_state_snapshot_from_endpoint_pools(&self.endpoint_pools)
-    }
-
-    pub fn local_node_storage_snapshot(&self) -> ClusterLocalNodeStorageSnapshot {
-        let membership = self.membership_snapshot();
-        local_node_storage_snapshot_from_membership(&membership)
-    }
-
-    pub fn peer_health_snapshot(&self) -> ClusterPeerHealthSnapshot {
-        let membership = self.membership_snapshot();
-        peer_health_snapshot_from_membership(&membership)
-    }
-
-    pub fn read_snapshot(&self) -> ClusterControlPlaneSnapshot {
-        let membership = self.membership_snapshot();
-
-        ClusterControlPlaneSnapshot {
-            topology: self.topology_snapshot(),
-            pool_state: self.pool_state_snapshot(),
-            local_storage: local_node_storage_snapshot_from_membership(&membership),
-            peer_health: peer_health_snapshot_from_membership(&membership),
-            membership,
-        }
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ClusterControlPlaneSnapshot {
-    pub topology: TopologySnapshot,
-    pub pool_state: ClusterPoolStateSnapshot,
-    pub local_storage: ClusterLocalNodeStorageSnapshot,
-    pub peer_health: ClusterPeerHealthSnapshot,
-    pub membership: ClusterMembershipSnapshot,
-}
-
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct ClusterMembershipSnapshot {
-    pub nodes: Vec<ClusterNodeMembership>,
-    pub drives: Vec<ClusterDriveMembership>,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ClusterNodeMembership {
-    pub node_id: String,
-    pub grid_host: String,
-    pub is_local: bool,
-    pub pools: Vec<usize>,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ClusterDriveMembership {
-    pub pool_index: usize,
-    pub set_index: usize,
-    pub disk_index: usize,
-    pub node_id: String,
-    pub is_local: bool,
-    pub endpoint_type: ClusterEndpointType,
-}
-
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct ClusterPoolStateSnapshot {
-    pub pools: Vec<ClusterPoolState>,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ClusterPoolState {
-    pub pool_index: usize,
-    pub set_count: usize,
-    pub drives_per_set: usize,
-    pub endpoint_count: usize,
-    pub local_drive_count: usize,
-    pub remote_drive_count: usize,
-    pub legacy: bool,
-    pub endpoint_types: Vec<ClusterEndpointType>,
-}
-
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct ClusterLocalNodeStorageSnapshot {
-    pub nodes: Vec<ClusterLocalNodeStorage>,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ClusterLocalNodeStorage {
-    pub node_id: String,
-    pub pools: Vec<usize>,
-    pub drive_count: usize,
-    pub path_drive_count: usize,
-    pub url_drive_count: usize,
-}
-
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct ClusterPeerHealthSnapshot {
-    pub peers: Vec<ClusterPeerHealth>,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct ClusterPeerHealth {
-    pub node_id: String,
-    pub is_local: bool,
-    pub status: CapabilityStatus,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
-pub enum ClusterEndpointType {
-    Path,
-    Url,
-}
-
-pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot {
-    topology_snapshot_from_endpoint_pools_with_capabilities(
-        endpoint_pools,
-        default_topology_capabilities(),
-        default_disk_capabilities(),
-    )
-}
-
-pub fn topology_snapshot_from_endpoint_pools_with_capabilities(
-    endpoint_pools: &EndpointServerPools,
-    capabilities: TopologyCapabilities,
-    disk_capabilities: DiskCapabilities,
-) -> TopologySnapshot {
-    TopologySnapshot {
-        pools: endpoint_pools
-            .as_ref()
-            .iter()
-            .enumerate()
-            .map(|(pool_index, pool)| {
-                let sets =
-                    topology_sets_from_endpoints(pool_index, pool.drives_per_set, pool.endpoints.as_ref(), &disk_capabilities);
-                TopologyPool {
-                    pool_index,
-                    pool_id: None,
-                    labels: TopologyLabels::default(),
-                    sets,
-                }
-            })
-            .collect(),
-        capabilities,
-    }
-}
-
-pub fn membership_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> ClusterMembershipSnapshot {
-    let mut nodes = BTreeMap::<String, ClusterNodeMembership>::new();
-    let mut drives = Vec::new();
-
-    for (pool_index, pool) in endpoint_pools.as_ref().iter().enumerate() {
-        for (endpoint_index, endpoint) in pool.endpoints.as_ref().iter().enumerate() {
-            let (set_index, disk_index) = endpoint_indices(pool_index, endpoint_index, pool.drives_per_set, endpoint);
-            let node_id = endpoint_node_id(endpoint);
-
-            match nodes.entry(node_id.clone()) {
-                std::collections::btree_map::Entry::Vacant(entry) => {
-                    entry.insert(ClusterNodeMembership {
-                        node_id: node_id.clone(),
-                        grid_host: endpoint.grid_host(),
-                        is_local: endpoint.is_local,
-                        pools: vec![pool_index],
-                    });
-                }
-                std::collections::btree_map::Entry::Occupied(mut entry) => {
-                    let node = entry.get_mut();
-                    node.is_local |= endpoint.is_local;
-                    if !node.pools.contains(&pool_index) {
-                        node.pools.push(pool_index);
-                    }
-                }
-            }
-
-            drives.push(ClusterDriveMembership {
-                pool_index,
-                set_index,
-                disk_index,
-                node_id,
-                is_local: endpoint.is_local,
-                endpoint_type: endpoint_type(endpoint),
-            });
-        }
-    }
-
-    ClusterMembershipSnapshot {
-        nodes: nodes.into_values().collect(),
-        drives,
-    }
-}
-
-pub fn pool_state_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> ClusterPoolStateSnapshot {
-    ClusterPoolStateSnapshot {
-        pools: endpoint_pools
-            .as_ref()
-            .iter()
-            .enumerate()
-            .map(|(pool_index, pool)| {
-                let mut endpoint_types = BTreeSet::new();
-                let mut local_drive_count = 0;
-
-                for endpoint in pool.endpoints.as_ref() {
-                    endpoint_types.insert(endpoint_type(endpoint));
-                    if endpoint.is_local {
-                        local_drive_count += 1;
-                    }
-                }
-
-                let endpoint_count = pool.endpoints.as_ref().len();
-                ClusterPoolState {
-                    pool_index,
-                    set_count: pool.set_count,
-                    drives_per_set: pool.drives_per_set,
-                    endpoint_count,
-                    local_drive_count,
-                    remote_drive_count: endpoint_count.saturating_sub(local_drive_count),
-                    legacy: pool.legacy,
-                    endpoint_types: endpoint_types.into_iter().collect(),
-                }
-            })
-            .collect(),
-    }
-}
-
-pub fn local_node_storage_snapshot_from_membership(membership: &ClusterMembershipSnapshot) -> ClusterLocalNodeStorageSnapshot {
-    ClusterLocalNodeStorageSnapshot {
-        nodes: membership
-            .nodes
-            .iter()
-            .filter(|node| node.is_local)
-            .map(|node| {
-                let mut path_drive_count = 0;
-                let mut url_drive_count = 0;
-
-                for drive in membership.drives.iter().filter(|drive| drive.node_id == node.node_id) {
-                    match drive.endpoint_type {
-                        ClusterEndpointType::Path => path_drive_count += 1,
-                        ClusterEndpointType::Url => url_drive_count += 1,
-                    }
-                }
-
-                ClusterLocalNodeStorage {
-                    node_id: node.node_id.clone(),
-                    pools: node.pools.clone(),
-                    drive_count: path_drive_count + url_drive_count,
-                    path_drive_count,
-                    url_drive_count,
-                }
-            })
-            .collect(),
-    }
-}
-
-pub fn peer_health_snapshot_from_membership(membership: &ClusterMembershipSnapshot) -> ClusterPeerHealthSnapshot {
-    ClusterPeerHealthSnapshot {
-        peers: membership
-            .nodes
-            .iter()
-            .map(|node| ClusterPeerHealth {
-                node_id: node.node_id.clone(),
-                is_local: node.is_local,
-                status: CapabilityStatus::unknown().with_reason(PEER_HEALTH_NOT_REPORTED),
-            })
-            .collect(),
-    }
-}
-
-fn topology_sets_from_endpoints(
-    pool_index: usize,
-    drives_per_set: usize,
-    endpoints: &[Endpoint],
-    disk_capabilities: &DiskCapabilities,
-) -> Vec<TopologySet> {
-    let mut sets = BTreeMap::<usize, Vec<TopologyDisk>>::new();
-
-    for (endpoint_index, endpoint) in endpoints.iter().enumerate() {
-        let (set_index, disk_index) = endpoint_indices(pool_index, endpoint_index, drives_per_set, endpoint);
-        sets.entry(set_index).or_default().push(topology_disk_from_endpoint(
-            pool_index,
-            set_index,
-            disk_index,
-            endpoint,
-            disk_capabilities,
-        ));
-    }
-
-    sets.into_iter()
-        .map(|(set_index, mut disks)| {
-            disks.sort_by_key(|disk| disk.disk_index);
-            TopologySet {
-                pool_index,
-                set_index,
-                set_id: None,
-                labels: TopologyLabels::default(),
-                disks,
-            }
-        })
-        .collect()
-}
-
-fn topology_disk_from_endpoint(
-    pool_index: usize,
-    set_index: usize,
-    disk_index: usize,
-    endpoint: &Endpoint,
-    disk_capabilities: &DiskCapabilities,
-) -> TopologyDisk {
-    TopologyDisk {
-        pool_index,
-        set_index,
-        disk_index,
-        disk_id: endpoint_disk_id(endpoint),
-        labels: endpoint_labels(endpoint),
-        capabilities: disk_capabilities.clone(),
-    }
-}
-
-fn endpoint_indices(pool_index: usize, endpoint_index: usize, drives_per_set: usize, endpoint: &Endpoint) -> (usize, usize) {
-    let safe_drives_per_set = drives_per_set.max(1);
-    let set_index = non_negative_index(endpoint.set_idx).unwrap_or(endpoint_index / safe_drives_per_set);
-    let disk_index = non_negative_index(endpoint.disk_idx).unwrap_or(endpoint_index % safe_drives_per_set);
-
-    debug_assert_eq!(non_negative_index(endpoint.pool_idx).unwrap_or(pool_index), pool_index);
-    (set_index, disk_index)
-}
-
-fn endpoint_disk_id(endpoint: &Endpoint) -> Option<String> {
-    let host_port = endpoint.host_port();
-    if host_port.is_empty() { None } else { Some(host_port) }
-}
-
-fn endpoint_node_id(endpoint: &Endpoint) -> String {
-    endpoint_disk_id(endpoint).unwrap_or_else(|| LOCAL_NODE_ID.to_owned())
-}
-
-fn endpoint_labels(endpoint: &Endpoint) -> TopologyLabels {
-    let mut additional = BTreeMap::new();
-    additional.insert(ENDPOINT_TYPE_LABEL.to_owned(), endpoint_type_label(endpoint).to_owned());
-    additional.insert(LOCAL_ENDPOINT_LABEL.to_owned(), endpoint.is_local.to_string());
-
-    TopologyLabels {
-        node: Some(endpoint_node_id(endpoint)),
-        additional,
-        ..TopologyLabels::default()
-    }
-}
-
-fn endpoint_type(endpoint: &Endpoint) -> ClusterEndpointType {
-    match endpoint.get_type() {
-        EndpointType::Path => ClusterEndpointType::Path,
-        EndpointType::Url => ClusterEndpointType::Url,
-    }
-}
-
-fn endpoint_type_label(endpoint: &Endpoint) -> &'static str {
-    match endpoint_type(endpoint) {
-        ClusterEndpointType::Path => "path",
-        ClusterEndpointType::Url => "url",
-    }
-}
-
-fn non_negative_index(index: i32) -> Option<usize> {
-    usize::try_from(index).ok()
-}
-
-fn default_topology_capabilities() -> TopologyCapabilities {
-    TopologyCapabilities {
-        profiling: CapabilityStatus::unknown().with_reason(PROFILING_NOT_WIRED),
-        numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED),
-        failure_domain_labels: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED),
-        media_labels: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED),
-    }
-}
-
-fn default_disk_capabilities() -> DiskCapabilities {
-    DiskCapabilities {
-        media_type: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED),
-        failure_domain: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED),
-        numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED),
-        profiling: CapabilityStatus::unknown().with_reason(PROFILING_NOT_WIRED),
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::collections::BTreeSet;
-
-    use crate::endpoints::{Endpoints, PoolEndpoints};
-
-    #[test]
-    fn topology_snapshot_maps_endpoint_sets_without_local_paths() {
-        let endpoint_pools = sample_path_endpoint_pools();
-        let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools);
-
-        assert_eq!(snapshot.pools.len(), 1);
-        assert_eq!(snapshot.pools[0].sets.len(), 2);
-        assert_eq!(snapshot.pools[0].sets[0].disks.len(), 2);
-        assert_eq!(snapshot.pools[0].sets[1].disks.len(), 2);
-        assert_eq!(snapshot.pools[0].sets[1].disks[1].disk_index, 1);
-        assert_eq!(
-            snapshot.pools[0].sets[0].disks[0]
-                .labels
-                .additional
-                .get(ENDPOINT_TYPE_LABEL)
-                .map(String::as_str),
-            Some("path")
-        );
-        assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some(LOCAL_NODE_ID));
-
-        let encoded = serde_json::to_string(&snapshot).expect("serialize topology snapshot");
-        assert!(!encoded.contains("/tmp/rustfs-cluster-control-plane"));
-    }
-
-    #[test]
-    fn topology_snapshot_uses_url_hosts_as_disk_ids() {
-        let endpoint_pools = sample_url_endpoint_pools();
-        let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools);
-
-        assert_eq!(snapshot.pools[0].sets[0].disks[0].disk_id.as_deref(), Some("node1.example:9000"));
-        assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some("node1.example:9000"));
-        assert_eq!(
-            snapshot.pools[0].sets[0].disks[0]
-                .labels
-                .additional
-                .get(ENDPOINT_TYPE_LABEL)
-                .map(String::as_str),
-            Some("url")
-        );
-    }
-
-    #[test]
-    fn membership_snapshot_groups_nodes_and_drives() {
-        let endpoint_pools = sample_url_endpoint_pools();
-        let snapshot = membership_snapshot_from_endpoint_pools(&endpoint_pools);
-
-        assert_eq!(snapshot.nodes.len(), 2);
-        assert_eq!(snapshot.drives.len(), 4);
-        assert_eq!(snapshot.nodes[0].node_id, "node1.example:9000");
-        assert_eq!(snapshot.nodes[0].pools, vec![0]);
-        assert_eq!(snapshot.drives[2].node_id, "node2.example:9000");
-        assert_eq!(snapshot.drives[2].set_index, 1);
-        assert_eq!(snapshot.drives[2].endpoint_type, ClusterEndpointType::Url);
-    }
-
-    #[test]
-    fn pool_state_snapshot_counts_local_remote_drives_and_endpoint_types() {
-        let endpoint_pools = sample_mixed_endpoint_pools();
-        let snapshot = pool_state_snapshot_from_endpoint_pools(&endpoint_pools);
-
-        assert_eq!(snapshot.pools.len(), 1);
-        assert_eq!(snapshot.pools[0].set_count, 2);
-        assert_eq!(snapshot.pools[0].drives_per_set, 2);
-        assert_eq!(snapshot.pools[0].endpoint_count, 4);
-        assert_eq!(snapshot.pools[0].local_drive_count, 3);
-        assert_eq!(snapshot.pools[0].remote_drive_count, 1);
-        assert_eq!(
-            snapshot.pools[0].endpoint_types,
-            vec![ClusterEndpointType::Path, ClusterEndpointType::Url]
-        );
-    }
-
-    #[test]
-    fn local_node_storage_snapshot_keeps_only_local_drive_counts() {
-        let membership = membership_snapshot_from_endpoint_pools(&sample_mixed_endpoint_pools());
-        let snapshot = local_node_storage_snapshot_from_membership(&membership);
-
-        assert_eq!(snapshot.nodes.len(), 2);
-        assert_eq!(snapshot.nodes[0].node_id, LOCAL_NODE_ID);
-        assert_eq!(snapshot.nodes[0].drive_count, 2);
-        assert_eq!(snapshot.nodes[0].path_drive_count, 2);
-        assert_eq!(snapshot.nodes[0].url_drive_count, 0);
-        assert_eq!(snapshot.nodes[1].node_id, "node1.example:9000");
-        assert_eq!(snapshot.nodes[1].drive_count, 1);
-        assert_eq!(snapshot.nodes[1].path_drive_count, 0);
-        assert_eq!(snapshot.nodes[1].url_drive_count, 1);
-    }
-
-    #[test]
-    fn peer_health_snapshot_reports_static_unknown_status() {
-        let membership = membership_snapshot_from_endpoint_pools(&sample_url_endpoint_pools());
-        let snapshot = peer_health_snapshot_from_membership(&membership);
-
-        assert_eq!(snapshot.peers.len(), 2);
-        assert_eq!(snapshot.peers[0].node_id, "node1.example:9000");
-        assert!(snapshot.peers[0].is_local);
-        assert_eq!(snapshot.peers[0].status.reason.as_deref(), Some(PEER_HEALTH_NOT_REPORTED));
-        assert_eq!(snapshot.peers[1].node_id, "node2.example:9000");
-        assert!(!snapshot.peers[1].is_local);
-    }
-
-    #[test]
-    fn control_plane_read_snapshot_combines_topology_and_membership() {
-        let control_plane = ClusterControlPlane::new(sample_mixed_endpoint_pools());
-        let snapshot = control_plane.read_snapshot();
-        let node_ids = snapshot
-            .membership
-            .nodes
-            .iter()
-            .map(|node| node.node_id.as_str())
-            .collect::<BTreeSet<_>>();
-
-        assert_eq!(snapshot.topology.pools[0].sets.len(), 2);
-        assert_eq!(snapshot.pool_state.pools[0].endpoint_count, 4);
-        assert_eq!(snapshot.local_storage.nodes.len(), 2);
-        assert_eq!(snapshot.peer_health.peers.len(), 3);
-        assert_eq!(node_ids, BTreeSet::from([LOCAL_NODE_ID, "node1.example:9000", "node2.example:9000"]));
-    }
-
-    fn sample_path_endpoint_pools() -> EndpointServerPools {
-        let endpoints = (0..4)
-            .map(|index| {
-                let mut endpoint =
-                    Endpoint::try_from(format!("/tmp/rustfs-cluster-control-plane-{index}").as_str()).expect("local endpoint");
-                endpoint.set_pool_index(0);
-                endpoint.set_set_index(index / 2);
-                endpoint.set_disk_index(index % 2);
-                endpoint
-            })
-            .collect::<Vec<_>>();
-
-        EndpointServerPools::from(vec![PoolEndpoints {
-            legacy: false,
-            set_count: 2,
-            drives_per_set: 2,
-            endpoints: Endpoints::from(endpoints),
-            cmd_line: "/tmp/rustfs-cluster-control-plane-{0...3}".to_owned(),
-            platform: "OS: test | Arch: test".to_owned(),
-        }])
-    }
-
-    fn sample_url_endpoint_pools() -> EndpointServerPools {
-        let endpoints = (0..4)
-            .map(|index| {
-                let host = if index < 2 { "node1.example" } else { "node2.example" };
-                let mut endpoint =
-                    Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint");
-                endpoint.set_pool_index(0);
-                endpoint.set_set_index(index / 2);
-                endpoint.set_disk_index(index % 2);
-                endpoint.is_local = index < 2;
-                endpoint
-            })
-            .collect::<Vec<_>>();
-
-        EndpointServerPools::from(vec![PoolEndpoints {
-            legacy: false,
-            set_count: 2,
-            drives_per_set: 2,
-            endpoints: Endpoints::from(endpoints),
-            cmd_line: "http://node{1...2}.example:9000/export{0...3}".to_owned(),
-            platform: "OS: test | Arch: test".to_owned(),
-        }])
-    }
-
-    fn sample_mixed_endpoint_pools() -> EndpointServerPools {
-        let mut endpoints = Vec::new();
-
-        for index in 0..2 {
-            let mut endpoint =
-                Endpoint::try_from(format!("/tmp/rustfs-cluster-control-plane-{index}").as_str()).expect("local endpoint");
-            endpoint.set_pool_index(0);
-            endpoint.set_set_index(0);
-            endpoint.set_disk_index(index);
-            endpoints.push(endpoint);
-        }
-
-        for index in 2..4 {
-            let host = if index == 2 { "node1.example" } else { "node2.example" };
-            let mut endpoint = Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint");
-            endpoint.set_pool_index(0);
-            endpoint.set_set_index(1);
-            endpoint.set_disk_index(index - 2);
-            endpoint.is_local = index == 2;
-            endpoints.push(endpoint);
-        }
-
-        EndpointServerPools::from(vec![PoolEndpoints {
-            legacy: false,
-            set_count: 2,
-            drives_per_set: 2,
-            endpoints: Endpoints::from(endpoints),
-            cmd_line: "mixed-test-endpoints".to_owned(),
-            platform: "OS: test | Arch: test".to_owned(),
-        }])
-    }
-}

crates/ecstore/src/core/sets.rs

@@ -1090,9 +1090,36 @@ async fn init_storage_disks_with_errors(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::endpoints::SetupType;
     use crate::layout::endpoint::Endpoint;
     use crate::storage_api_contracts::heal::HealOperations as _;
     use crate::storage_api_contracts::list::ListOperations as _;
+    use rustfs_lock::client::local::LocalClient;
+    use serial_test::serial;
+
+    struct SetupTypeGuard {
+        previous: SetupType,
+    }
+
+    impl SetupTypeGuard {
+        async fn switch_to(next: SetupType) -> Self {
+            let previous = runtime_sources::current_setup_type().await;
+            runtime_sources::set_setup_type(next).await;
+            Self { previous }
+        }
+    }
+
+    impl Drop for SetupTypeGuard {
+        fn drop(&mut self) {
+            let previous = self.previous.clone();
+            let handle = tokio::runtime::Handle::current();
+            tokio::task::block_in_place(|| {
+                handle.block_on(async move {
+                    runtime_sources::set_setup_type(previous).await;
+                });
+            });
+        }
+    }
 
     #[test]
     fn test_apply_delete_objects_results_preserves_original_order_for_out_of_order_batches() {
@@ -1203,8 +1230,10 @@ mod tests {
         assert_eq!(result, (Some(3), Some(1), Some(0)));
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
     async fn sets_list_objects_v2_lists_objects_within_the_pool() {
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::Erasure).await;
         let format = FormatV3::new(1, 2);
         let mut endpoints = Vec::new();
         let mut disks = Vec::new();
@@ -1247,7 +1276,7 @@ mod tests {
             0,
             endpoints.clone(),
             format.clone(),
-            Vec::new(),
+            vec![Arc::new(LocalClient::new()), Arc::new(LocalClient::new())],
         )
         .await;
 
@@ -1272,23 +1301,26 @@ mod tests {
             exit_signal: None,
         });
 
-        sets.make_bucket("bucket", &MakeBucketOptions::default())
+        let bucket = format!("bucket-{}", Uuid::new_v4().simple());
+        let object = format!("object-{}", Uuid::new_v4().simple());
+
+        sets.make_bucket(&bucket, &MakeBucketOptions::default())
             .await
             .expect("bucket should be created");
 
         let mut reader = PutObjReader::from_vec(b"hello".to_vec());
-        sets.put_object("bucket", "object", &mut reader, &ObjectOptions::default())
+        sets.put_object(&bucket, &object, &mut reader, &ObjectOptions::default())
             .await
             .expect("object should be written");
 
         let result = sets
             .clone()
-            .list_objects_v2("bucket", "", None, None, 1000, false, None, false)
+            .list_objects_v2(&bucket, "", None, None, 1000, false, None, false)
             .await
             .expect("pool-level listing should succeed");
 
         assert_eq!(result.objects.len(), 1);
-        assert_eq!(result.objects[0].name, "object");
+        assert_eq!(result.objects[0].name, object);
     }
 
     #[tokio::test]

crates/ecstore/src/lib.rs

@@ -15,36 +15,51 @@
 
 extern crate core;
 
+#[path = "diagnostics/admin_server_info.rs"]
 mod admin_server_info;
 pub mod api;
+#[path = "services/batch_processor.rs"]
 mod batch_processor;
+#[path = "io_support/bitrot.rs"]
 mod bitrot;
 mod bucket;
 mod cache_value;
 mod cluster;
+#[path = "io_support/compress.rs"]
 mod compress;
 mod config;
 mod data_movement;
+#[path = "data_movement/backpressure.rs"]
 mod data_movement_backpressure;
 mod data_usage;
 mod disk;
+#[path = "layout/disks_layout_facade.rs"]
 mod disks_layout;
+#[path = "layout/endpoints_facade.rs"]
 mod endpoints;
 mod erasure_codec;
 mod erasure_coding;
 mod error;
+#[path = "diagnostics/get.rs"]
 mod get_diagnostics;
+#[path = "runtime/global.rs"]
 mod global;
 pub(crate) mod layout;
+#[path = "services/metrics_realtime.rs"]
 mod metrics_realtime;
+#[path = "services/notification_sys.rs"]
 mod notification_sys;
 mod object_api;
+#[path = "core/pools.rs"]
 mod pools;
 mod rebalance;
+#[path = "io_support/rio.rs"]
 mod rio;
 mod rpc;
+#[path = "runtime/sources.rs"]
 mod runtime_sources;
 mod set_disk;
+#[path = "core/sets.rs"]
 mod sets;
 mod storage_api_contracts;
 mod store;
@@ -58,10 +73,13 @@ mod store_utils;
 // pub mod checksum;
 mod client;
 mod event;
+#[path = "services/event_notification.rs"]
 mod event_notification;
 #[cfg(test)]
+#[path = "core/pools_test.rs"]
 mod pools_test;
 #[cfg(test)]
+#[path = "core/store_test.rs"]
 mod store_test;
 mod tier;
 

docs/architecture/migration-progress.md

@@ -5,20 +5,27 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 ## Current Context
 
 - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
-- Branch: `overtrue/arch-ecstore-store-init-layout`
+- Branch: `overtrue/arch-ecstore-cluster-control-plane-layout`
 - Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126/API-127/API-128/API-129/API-130/API-131/API-132/API-133/API-134/API-135/API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154/API-155/API-156/API-157/API-158/API-159/API-160/API-161/API-162/API-163/API-164/API-165/API-166/API-167/API-168/API-169/API-170/API-171/API-172/API-173/API-174/API-175/API-176/API-177/API-178/API-179/API-180/API-181/API-182/API-183/API-184/API-185/API-186/API-187/API-188/API-189/API-190/API-191/API-192/API-193/API-194/API-195/API-196/API-197/API-198/API-199/API-200/API-201/API-202/API-203/API-204/API-205/API-206/API-207/API-208/API-209/API-210/API-211/API-212/API-213/API-214/API-215/API-216/API-217/API-218/API-219/API-220/API-221/API-222/API-223/API-224/API-225/API-226/API-227/API-228/API-229/API-230/API-231/API-232/API-233/API-234/API-235/API-236/API-237/API-238/API-239/API-240/API-241/API-242/API-243/API-244/API-245/API-246/API-247/API-248/API-249/API-250/API-251/API-252/API-253/API-254/CTX-002`.
 - Current baseline also includes API-255 from PR #3923, API-256 from PR
   #3925, and CFG-009 from PR #3927.
-- Current phase PR: E-019/E-STORE-003 ECStore store init support module owner layout.
-- Based on: E-018 branch while PR #3929 is pending; rebase onto `origin/main`
-  after E-017 and E-018 merge before opening this PR.
-- PR type for this branch: `pure-move`
-- Runtime behavior changes: none expected for E-019; this is a pure module file
-  move that keeps the existing `store_init` module path.
-- Rust code changes: move ECStore store init support module into the `store/`
-  owner directory without function-body changes.
+- Current phase PR: E-028/E-CLUSTER-001 ECStore cluster control-plane owner layout.
+- Based on: E-027 branch while PR #3930 is pending; rebase onto `origin/main`
+  after E-017 through E-027 merge before opening this PR.
+- PR type for this branch: `pure-move`.
+- Runtime behavior changes: none expected for E-028; production changes are a
+  module file move that keeps the existing `cluster` facade paths.
+- Rust code changes: move ECStore cluster control-plane implementation under
+  the `cluster/control_plane` owner module without function-body changes.
 - CI/script changes: reject restoring ECStore root `store.rs`, `set_disk.rs`,
-  `store_list_objects.rs`, `store_utils.rs`, or `store_init.rs`;
+  `store_list_objects.rs`, `store_utils.rs`, `store_init.rs`,
+  `disks_layout.rs`, `endpoints.rs`, `storage_api_contracts.rs`,
+  `batch_processor.rs`, `event_notification.rs`, `metrics_realtime.rs`, or
+  `notification_sys.rs`, `data_movement.rs`, or
+  `data_movement_backpressure.rs`, `admin_server_info.rs`, `data_usage.rs`,
+  `get_diagnostics.rs`, `global.rs`, `runtime_sources.rs`, `bitrot.rs`,
+  `compress.rs`, `rio.rs`, `error.rs`, `rebalance.rs`, `pools.rs`,
+  `sets.rs`, `pools_test.rs`, or `store_test.rs`;
   lock completed config model ownership against restoring
   old `rustfs_ecstore::config` model/accessor compatibility paths, and lock
   ECStore public facades against re-exporting the moved server-config symbols;
@@ -54,8 +61,17 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
   files after API-256, and reject restoring old ECStore server-config model or
   global accessor compatibility paths after CFG-009, reject restoring ECStore
   root `store.rs` or `set_disk.rs` after E-017, reject restoring ECStore root
-  store support modules after E-018, and reject restoring ECStore root
-  `store_init.rs` after E-019.
+  store support modules after E-018, reject restoring ECStore root
+  `store_init.rs` after E-019, reject restoring ECStore root layout facade
+  and storage API contract support modules after E-020, and reject restoring
+  ECStore root service runtime modules after E-021, and reject restoring
+  ECStore root data movement modules after E-022, and reject restoring ECStore
+  root usage and diagnostics modules after E-023, and reject restoring ECStore
+  root runtime global modules after E-024, and reject restoring ECStore root
+  I/O support modules after E-025, and reject restoring ECStore root error and
+  rebalance facade modules after E-026, and reject restoring ECStore root core
+  runtime/test modules after E-027, and keep the ECStore cluster root module as
+  a re-export-only owner facade after E-028.
 
 ## Phase 0 Tasks
 
@@ -3932,6 +3948,158 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
     formatting, diff hygiene, Rust risk scan, branch freshness check,
     pre-commit quality gate, and three-expert review.
 
+- [x] `E-020/E-LAYOUT-001` Move ECStore layout facade and contract support modules under owners.
+  - Do: move `crates/ecstore/src/disks_layout.rs` to
+    `crates/ecstore/src/layout/disks_layout_facade.rs`,
+    `crates/ecstore/src/endpoints.rs` to
+    `crates/ecstore/src/layout/endpoints_facade.rs`, and
+  `crates/ecstore/src/storage_api_contracts.rs` to
+  `crates/ecstore/src/storage_api_contracts/mod.rs` while preserving the
+  existing crate module names; harden the pool-level list-object test against
+  parallel global lock-state pollution.
+  - Acceptance: `crate::disks_layout`, `crate::endpoints`, and
+    `crate::storage_api_contracts` continue to resolve through the same module
+    names, and migration rules reject restoring the old root files.
+  - Must preserve: disk layout facade re-exports, endpoint pool facade
+    re-exports, storage API contract domain aliases, ECStore internal contract
+    imports, pool-level list-object behavior, and all runtime side effects.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
+- [x] `E-021/E-SERVICES-001` Move ECStore service runtime modules under the services owner.
+  - Do: move `crates/ecstore/src/batch_processor.rs`,
+    `crates/ecstore/src/event_notification.rs`,
+    `crates/ecstore/src/metrics_realtime.rs`, and
+    `crates/ecstore/src/notification_sys.rs` under
+    `crates/ecstore/src/services/` while preserving the existing crate module
+    names through path declarations; harden the pool-level list-object test
+    against parallel global setup-state pollution.
+  - Acceptance: `crate::batch_processor`, `crate::event_notification`,
+    `crate::metrics_realtime`, and `crate::notification_sys` continue to
+    resolve through the same module names, and migration rules reject restoring
+    the old root files.
+  - Must preserve: batch job serialization/status behavior, notification event
+    registration/dispatch, realtime metrics snapshots, notification system
+    lifecycle, peer event behavior, pool-level list-object behavior, and all
+    runtime side effects.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
+- [x] `E-022/E-DATA-MOVEMENT-001` Move ECStore data movement modules under their owner.
+  - Do: move `crates/ecstore/src/data_movement.rs` to
+    `crates/ecstore/src/data_movement/mod.rs` and
+    `crates/ecstore/src/data_movement_backpressure.rs` to
+    `crates/ecstore/src/data_movement/backpressure.rs` while preserving the
+    existing crate module names.
+  - Acceptance: `crate::data_movement` and
+    `crate::data_movement_backpressure` continue to resolve through the same
+    module names, and migration rules reject restoring the old root files.
+  - Must preserve: decommission/rebalance data movement copy semantics,
+    overwrite-resume behavior, checksum propagation, abort flags, backpressure
+    admission policy, metrics, cancellation behavior, and all runtime side
+    effects.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
+- [x] `E-023/E-USAGE-DIAGNOSTICS-001` Move ECStore usage and diagnostics modules under owners.
+  - Do: move `crates/ecstore/src/data_usage.rs` to
+    `crates/ecstore/src/data_usage/mod.rs`,
+    `crates/ecstore/src/admin_server_info.rs` to
+    `crates/ecstore/src/diagnostics/admin_server_info.rs`, and
+    `crates/ecstore/src/get_diagnostics.rs` to
+    `crates/ecstore/src/diagnostics/get.rs` while preserving the existing
+    crate module names.
+  - Acceptance: `crate::data_usage`, `crate::admin_server_info`, and
+    `crate::get_diagnostics` continue to resolve through the same module names,
+    and migration rules reject restoring the old root files.
+  - Must preserve: usage cache paths, local usage snapshots, bucket usage
+    aggregation, server info assembly, GET diagnostic labels, diagnostics error
+    classification, and all runtime side effects.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
+- [x] `E-024/E-RUNTIME-001` Move ECStore runtime global modules under the runtime owner.
+  - Do: move `crates/ecstore/src/global.rs` to
+    `crates/ecstore/src/runtime/global.rs` and
+    `crates/ecstore/src/runtime_sources.rs` to
+    `crates/ecstore/src/runtime/sources.rs` while preserving the existing
+    crate module names through path declarations.
+  - Acceptance: `crate::global` and `crate::runtime_sources` continue to
+    resolve through the same module names, and migration rules reject restoring
+    the old root files.
+  - Must preserve: process-global state ownership, setup-type flags, endpoint
+    globals, object-store publication, lock-client publication, runtime source
+    adapters, test reset helpers, and all runtime side effects.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
+- [x] `E-025/E-IO-001` Move ECStore I/O support modules under the IO owner.
+  - Do: move `crates/ecstore/src/bitrot.rs`,
+    `crates/ecstore/src/compress.rs`, and `crates/ecstore/src/rio.rs` under
+    `crates/ecstore/src/io_support/` while preserving the existing crate module
+    names through path declarations.
+  - Acceptance: `crate::bitrot`, `crate::compress`, and `crate::rio` continue
+    to resolve through the same module names, and migration rules reject
+    restoring the old root files.
+  - Must preserve: bitrot reader construction, disk compression matching,
+    compression config defaults/env handling, RIO backend selection, encryption
+    and compression metadata behavior, reader wrappers, and all runtime side
+    effects.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
+- [x] `E-026/E-CORE-001` Move ECStore error and rebalance facades under owners.
+  - Do: move `crates/ecstore/src/error.rs` to
+    `crates/ecstore/src/error/mod.rs` and `crates/ecstore/src/rebalance.rs` to
+    `crates/ecstore/src/rebalance/mod.rs` while preserving the existing crate
+    module names through directory module resolution.
+  - Acceptance: `crate::error` and `crate::rebalance` continue to resolve
+    through the same module names, and migration rules reject restoring the old
+    root files.
+  - Must preserve: storage error variants/conversions/classifiers, S3 error
+    mapping, rebalance metadata exports, rebalance orchestration constants,
+    rebalance tests, and all runtime side effects.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
+- [x] `E-027/E-CORE-001` Move ECStore core runtime and test harness modules under owner.
+  - Do: move `crates/ecstore/src/pools.rs` and
+    `crates/ecstore/src/sets.rs` under `crates/ecstore/src/core/`, and move
+    the root ECStore core harness files `pools_test.rs` and `store_test.rs`
+    under the same owner while preserving crate module names through path
+    declarations.
+  - Acceptance: `crate::pools`, `crate::sets`, and the existing test modules
+    continue to resolve through the same module names, and migration rules
+    reject restoring the old root files.
+  - Must preserve: pool orchestration, set routing, ECStore object API
+    behavior, test harness setup/teardown, lock/setup isolation, and all
+    runtime side effects.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
+- [x] `E-028/E-CLUSTER-001` Move ECStore cluster control-plane implementation under owner.
+  - Do: move `crates/ecstore/src/cluster/mod.rs` implementation into
+    `crates/ecstore/src/cluster/control_plane.rs` and keep
+    `cluster/mod.rs` as a re-export-only facade.
+  - Acceptance: `rustfs_ecstore::api::cluster::*` and existing
+    `crate::cluster::*` consumers keep the same symbols, and migration rules
+    reject restoring control-plane implementation declarations to the cluster
+    root module.
+  - Must preserve: topology snapshots, membership grouping, pool-state
+    reporting, local-node storage projection, peer-health unknown status,
+    read-only facade behavior, and all cluster tests.
+  - Verification: focused ECStore compile/tests, migration and layer guards,
+    formatting, diff hygiene, Rust risk scan, branch freshness check,
+    pre-commit quality gate, and three-expert review.
+
 - [x] `API-129` Route RustFS internal ECStore consumers through owner boundary.
   - Do: expose crate-local ECStore facade module aliases from
     `rustfs/src/storage/mod.rs` and migrate RustFS startup, server, capacity,
@@ -5786,6 +5954,33 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 
 | Expert | Status | Notes |
 |---|---|---|
+| Quality/architecture | pass | E-028 moves the ECStore cluster control-plane implementation into an owner submodule and leaves the root module as a facade. |
+| Migration preservation | pass | `crate::cluster::*` and `rustfs_ecstore::api::cluster::*` stay stable through root re-exports; the moved implementation is a 100% rename. |
+| Testing/verification | pass | ECStore cluster tests, ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
+| Quality/architecture | pass | E-027 moves ECStore pools, sets, and root core test harness modules under the core owner directory and extends the root-file guard. |
+| Migration preservation | pass | `crate::pools`, `crate::sets`, `pools_test`, and `store_test` stay stable through path declarations; moved files are 100% renames. |
+| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
+| Quality/architecture | pass | E-026 moves ECStore error and rebalance facades under owner directories and extends the root-file guard. |
+| Migration preservation | pass | `crate::error` and `crate::rebalance` stay stable through directory module resolution; moved files are 100% renames. |
+| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
+| Quality/architecture | pass | E-025 moves ECStore I/O support modules under the `io_support` owner directory and extends the root-file guard. |
+| Migration preservation | pass | `crate::bitrot`, `crate::compress`, and `crate::rio` stay stable through path declarations; moved files are 100% renames. |
+| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
+| Quality/architecture | pass | E-024 moves ECStore runtime global state and runtime-source adapters under the `runtime` owner directory and extends the root-file guard. |
+| Migration preservation | pass | `crate::global` and `crate::runtime_sources` stay stable through path declarations; moved files are 100% renames. |
+| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
+| Quality/architecture | pass | E-023 moves ECStore usage and diagnostics/status modules under owner directories and extends the root-file guard. |
+| Migration preservation | pass | `crate::data_usage`, `crate::admin_server_info`, and `crate::get_diagnostics` stay stable through directory module resolution or path declarations; moved files are 100% renames. |
+| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
+| Quality/architecture | pass | E-022 moves ECStore data movement modules under the `data_movement` owner directory and extends the root-file guard. |
+| Migration preservation | pass | `crate::data_movement` and `crate::data_movement_backpressure` stay stable through directory module resolution and a path declaration; moved files are 100% renames. |
+| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
+| Quality/architecture | pass | E-021 moves ECStore service runtime modules under the `services` owner directory, extends the root-file guard, and keeps setup-state test hardening local to `sets` coverage. |
+| Migration preservation | pass | `crate::batch_processor`, `crate::event_notification`, `crate::metrics_realtime`, and `crate::notification_sys` stay stable through path declarations; moved files are 100% renames and the `sets` test restores setup state. |
+| Testing/verification | pass | ECStore target test, default ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
+| Quality/architecture | pass | E-020 moves ECStore disk-layout/endpoints facade shims and storage API contracts under owner directories, extends the root-file guard, and keeps test hardening local to `sets` coverage. |
+| Migration preservation | pass | `crate::disks_layout`, `crate::endpoints`, and `crate::storage_api_contracts` stay stable through module path declarations or directory module resolution; the `sets` test now owns its lock clients and resource names. |
+| Testing/verification | pass | ECStore target test, default ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |
 | Quality/architecture | pass | E-019 moves ECStore store init format support under the store owner directory and extends the existing root support guard. |
 | Migration preservation | pass | `crate::store_init` stays stable through a path declaration, and the moved file is a 100% rename with no function-body changes. |
 | Testing/verification | pass | ECStore tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. |

scripts/check_architecture_migration_rules.sh

@@ -132,6 +132,15 @@ EXTERNAL_ECSTORE_API_BOUNDARY_HITS_FILE="${TMP_DIR}/external_ecstore_api_boundar
 LEGACY_ECSTORE_CONFIG_MODEL_HITS_FILE="${TMP_DIR}/legacy_ecstore_config_model_hits.txt"
 ECSTORE_ROOT_STORE_SET_DISK_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_store_set_disk_module_hits.txt"
 ECSTORE_ROOT_STORE_SUPPORT_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_store_support_module_hits.txt"
+ECSTORE_ROOT_LAYOUT_CONTRACT_SUPPORT_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_layout_contract_support_module_hits.txt"
+ECSTORE_ROOT_SERVICE_RUNTIME_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_service_runtime_module_hits.txt"
+ECSTORE_ROOT_DATA_MOVEMENT_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_data_movement_module_hits.txt"
+ECSTORE_ROOT_USAGE_DIAGNOSTICS_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_usage_diagnostics_module_hits.txt"
+ECSTORE_ROOT_RUNTIME_GLOBAL_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_runtime_global_module_hits.txt"
+ECSTORE_ROOT_IO_SUPPORT_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_io_support_module_hits.txt"
+ECSTORE_ROOT_ERROR_REBALANCE_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_error_rebalance_module_hits.txt"
+ECSTORE_ROOT_CORE_RUNTIME_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_core_runtime_module_hits.txt"
+ECSTORE_CLUSTER_ROOT_IMPL_HITS_FILE="${TMP_DIR}/ecstore_cluster_root_impl_hits.txt"
 ALL_STORAGE_COMPAT_SELF_FACADE_PATH_HITS_FILE="${TMP_DIR}/all_storage_compat_self_facade_path_hits.txt"
 RUSTFS_LOCAL_COMPAT_OWNER_SELF_PATH_HITS_FILE="${TMP_DIR}/rustfs_local_compat_owner_self_path_hits.txt"
 RUSTFS_ROOT_COMPAT_RELATIVE_CONSUMER_HITS_FILE="${TMP_DIR}/rustfs_root_compat_relative_consumer_hits.txt"
@@ -700,17 +709,21 @@ fi
   rg -n --with-filename '^use rustfs_storage_api|^pub use rustfs_storage_api|rustfs_storage_api::' \
     crates/ecstore/src \
     --glob '*.rs' \
-    --glob '!storage_api_contracts.rs' || true
+    --glob '!crates/ecstore/src/storage_api_contracts/mod.rs' \
+    --glob '!crates/ecstore/src/storage_api_contracts/**' \
+    --glob '!storage_api_contracts.rs' \
+    --glob '!storage_api_contracts/mod.rs' \
+    --glob '!storage_api_contracts/**' || true
 ) >"$ECSTORE_DIRECT_STORAGE_API_SOURCE_HITS_FILE"
 
 if [[ -s "$ECSTORE_DIRECT_STORAGE_API_SOURCE_HITS_FILE" ]]; then
-  report_failure "ECStore modules must route storage-api symbols through crates/ecstore/src/storage_api_contracts.rs: $(paste -sd '; ' "$ECSTORE_DIRECT_STORAGE_API_SOURCE_HITS_FILE")"
+  report_failure "ECStore modules must route storage-api symbols through crates/ecstore/src/storage_api_contracts: $(paste -sd '; ' "$ECSTORE_DIRECT_STORAGE_API_SOURCE_HITS_FILE")"
 fi
 
 (
   cd "$ROOT_DIR"
   rg -n --with-filename '^pub(?:\(crate\))? use rustfs_storage_api' \
-    crates/ecstore/src/storage_api_contracts.rs || true
+    crates/ecstore/src/storage_api_contracts/mod.rs || true
 ) >"$ECSTORE_STORAGE_API_ROOT_REEXPORT_HITS_FILE"
 
 if [[ -s "$ECSTORE_STORAGE_API_ROOT_REEXPORT_HITS_FILE" ]]; then
@@ -723,11 +736,19 @@ fi
     rg -n -U --with-filename 'storage_api_contracts::\{\s*[A-Z]' \
       crates/ecstore/src \
       --glob '*.rs' \
-      --glob '!storage_api_contracts.rs' || true
+      --glob '!crates/ecstore/src/storage_api_contracts/mod.rs' \
+      --glob '!crates/ecstore/src/storage_api_contracts/**' \
+      --glob '!storage_api_contracts.rs' \
+      --glob '!storage_api_contracts/mod.rs' \
+      --glob '!storage_api_contracts/**' || true
     rg -n --with-filename 'storage_api_contracts::[A-Z]' \
       crates/ecstore/src \
       --glob '*.rs' \
-      --glob '!storage_api_contracts.rs' || true
+      --glob '!crates/ecstore/src/storage_api_contracts/mod.rs' \
+      --glob '!crates/ecstore/src/storage_api_contracts/**' \
+      --glob '!storage_api_contracts.rs' \
+      --glob '!storage_api_contracts/mod.rs' \
+      --glob '!storage_api_contracts/**' || true
   }
 ) >"$ECSTORE_STORAGE_API_ROOT_CONSUMER_HITS_FILE"
 
@@ -2805,6 +2826,125 @@ if [[ -s "$ECSTORE_ROOT_STORE_SUPPORT_MODULE_HITS_FILE" ]]; then
   report_failure "ECStore store support modules must stay under the store owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_STORE_SUPPORT_MODULE_HITS_FILE")"
 fi
 
+(
+  cd "$ROOT_DIR"
+  {
+    [[ -e crates/ecstore/src/disks_layout.rs ]] && printf '%s\n' 'crates/ecstore/src/disks_layout.rs'
+    [[ -e crates/ecstore/src/endpoints.rs ]] && printf '%s\n' 'crates/ecstore/src/endpoints.rs'
+    [[ -e crates/ecstore/src/storage_api_contracts.rs ]] && printf '%s\n' 'crates/ecstore/src/storage_api_contracts.rs'
+    true
+  }
+) >"$ECSTORE_ROOT_LAYOUT_CONTRACT_SUPPORT_MODULE_HITS_FILE"
+
+if [[ -s "$ECSTORE_ROOT_LAYOUT_CONTRACT_SUPPORT_MODULE_HITS_FILE" ]]; then
+  report_failure "ECStore layout facade and storage API contract modules must stay under owner directories: $(paste -sd '; ' "$ECSTORE_ROOT_LAYOUT_CONTRACT_SUPPORT_MODULE_HITS_FILE")"
+fi
+
+(
+  cd "$ROOT_DIR"
+  {
+    [[ -e crates/ecstore/src/batch_processor.rs ]] && printf '%s\n' 'crates/ecstore/src/batch_processor.rs'
+    [[ -e crates/ecstore/src/event_notification.rs ]] && printf '%s\n' 'crates/ecstore/src/event_notification.rs'
+    [[ -e crates/ecstore/src/metrics_realtime.rs ]] && printf '%s\n' 'crates/ecstore/src/metrics_realtime.rs'
+    [[ -e crates/ecstore/src/notification_sys.rs ]] && printf '%s\n' 'crates/ecstore/src/notification_sys.rs'
+    true
+  }
+) >"$ECSTORE_ROOT_SERVICE_RUNTIME_MODULE_HITS_FILE"
+
+if [[ -s "$ECSTORE_ROOT_SERVICE_RUNTIME_MODULE_HITS_FILE" ]]; then
+  report_failure "ECStore service runtime modules must stay under the services owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_SERVICE_RUNTIME_MODULE_HITS_FILE")"
+fi
+
+(
+  cd "$ROOT_DIR"
+  {
+    [[ -e crates/ecstore/src/data_movement.rs ]] && printf '%s\n' 'crates/ecstore/src/data_movement.rs'
+    [[ -e crates/ecstore/src/data_movement_backpressure.rs ]] && printf '%s\n' 'crates/ecstore/src/data_movement_backpressure.rs'
+    true
+  }
+) >"$ECSTORE_ROOT_DATA_MOVEMENT_MODULE_HITS_FILE"
+
+if [[ -s "$ECSTORE_ROOT_DATA_MOVEMENT_MODULE_HITS_FILE" ]]; then
+  report_failure "ECStore data movement modules must stay under the data_movement owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_DATA_MOVEMENT_MODULE_HITS_FILE")"
+fi
+
+(
+  cd "$ROOT_DIR"
+  {
+    [[ -e crates/ecstore/src/admin_server_info.rs ]] && printf '%s\n' 'crates/ecstore/src/admin_server_info.rs'
+    [[ -e crates/ecstore/src/data_usage.rs ]] && printf '%s\n' 'crates/ecstore/src/data_usage.rs'
+    [[ -e crates/ecstore/src/get_diagnostics.rs ]] && printf '%s\n' 'crates/ecstore/src/get_diagnostics.rs'
+    true
+  }
+) >"$ECSTORE_ROOT_USAGE_DIAGNOSTICS_MODULE_HITS_FILE"
+
+if [[ -s "$ECSTORE_ROOT_USAGE_DIAGNOSTICS_MODULE_HITS_FILE" ]]; then
+  report_failure "ECStore usage and diagnostics modules must stay under owner directories: $(paste -sd '; ' "$ECSTORE_ROOT_USAGE_DIAGNOSTICS_MODULE_HITS_FILE")"
+fi
+
+(
+  cd "$ROOT_DIR"
+  {
+    [[ -e crates/ecstore/src/global.rs ]] && printf '%s\n' 'crates/ecstore/src/global.rs'
+    [[ -e crates/ecstore/src/runtime_sources.rs ]] && printf '%s\n' 'crates/ecstore/src/runtime_sources.rs'
+    true
+  }
+) >"$ECSTORE_ROOT_RUNTIME_GLOBAL_MODULE_HITS_FILE"
+
+if [[ -s "$ECSTORE_ROOT_RUNTIME_GLOBAL_MODULE_HITS_FILE" ]]; then
+  report_failure "ECStore runtime global modules must stay under the runtime owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_RUNTIME_GLOBAL_MODULE_HITS_FILE")"
+fi
+
+(
+  cd "$ROOT_DIR"
+  {
+    [[ -e crates/ecstore/src/bitrot.rs ]] && printf '%s\n' 'crates/ecstore/src/bitrot.rs'
+    [[ -e crates/ecstore/src/compress.rs ]] && printf '%s\n' 'crates/ecstore/src/compress.rs'
+    [[ -e crates/ecstore/src/rio.rs ]] && printf '%s\n' 'crates/ecstore/src/rio.rs'
+    true
+  }
+) >"$ECSTORE_ROOT_IO_SUPPORT_MODULE_HITS_FILE"
+
+if [[ -s "$ECSTORE_ROOT_IO_SUPPORT_MODULE_HITS_FILE" ]]; then
+  report_failure "ECStore I/O support modules must stay under the io_support owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_IO_SUPPORT_MODULE_HITS_FILE")"
+fi
+
+(
+  cd "$ROOT_DIR"
+  {
+    [[ -e crates/ecstore/src/error.rs ]] && printf '%s\n' 'crates/ecstore/src/error.rs'
+    [[ -e crates/ecstore/src/rebalance.rs ]] && printf '%s\n' 'crates/ecstore/src/rebalance.rs'
+    true
+  }
+) >"$ECSTORE_ROOT_ERROR_REBALANCE_MODULE_HITS_FILE"
+
+if [[ -s "$ECSTORE_ROOT_ERROR_REBALANCE_MODULE_HITS_FILE" ]]; then
+  report_failure "ECStore error and rebalance modules must stay under owner directories: $(paste -sd '; ' "$ECSTORE_ROOT_ERROR_REBALANCE_MODULE_HITS_FILE")"
+fi
+
+(
+  cd "$ROOT_DIR"
+  {
+    [[ -e crates/ecstore/src/pools.rs ]] && printf '%s\n' 'crates/ecstore/src/pools.rs'
+    [[ -e crates/ecstore/src/sets.rs ]] && printf '%s\n' 'crates/ecstore/src/sets.rs'
+    [[ -e crates/ecstore/src/pools_test.rs ]] && printf '%s\n' 'crates/ecstore/src/pools_test.rs'
+    [[ -e crates/ecstore/src/store_test.rs ]] && printf '%s\n' 'crates/ecstore/src/store_test.rs'
+    true
+  }
+) >"$ECSTORE_ROOT_CORE_RUNTIME_MODULE_HITS_FILE"
+
+if [[ -s "$ECSTORE_ROOT_CORE_RUNTIME_MODULE_HITS_FILE" ]]; then
+  report_failure "ECStore core runtime modules must stay under the core owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_CORE_RUNTIME_MODULE_HITS_FILE")"
+fi
+
+rg -n --with-filename 'pub (struct|enum|fn) Cluster|pub fn (topology|membership|pool_state|local_node_storage|peer_health)_snapshot' \
+  "${ROOT_DIR}/crates/ecstore/src/cluster/mod.rs" \
+  >"$ECSTORE_CLUSTER_ROOT_IMPL_HITS_FILE" || true
+
+if [[ -s "$ECSTORE_CLUSTER_ROOT_IMPL_HITS_FILE" ]]; then
+  report_failure "ECStore cluster root module must only re-export control-plane owner symbols: $(paste -sd '; ' "$ECSTORE_CLUSTER_ROOT_IMPL_HITS_FILE")"
+fi
+
 cat >"$ECSTORE_COMPAT_PASSTHROUGH_EXPECTED_FILE" <<'EOF'
 EOF
 sort -o "$ECSTORE_COMPAT_PASSTHROUGH_EXPECTED_FILE" "$ECSTORE_COMPAT_PASSTHROUGH_EXPECTED_FILE"