`9cc8ddf`

integrate coordinator client for distributed node management

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 8 months ago

SHA: 9cc8ddf6fdf825ad2bd7e9768e4895201dffde2c
Parents: de09132
Tree: 3b74901

14 changed files

Status	File	+	-
M	`Cargo.toml`	5	0
A	`Dockerfile.test`	48	0
M	`build.rs`	13	0
A	`config.test.yaml`	25	0
A	`docker-compose.test.yml`	90	0
M	`src/config.rs`	1	1
A	`src/coordinator/client.rs`	339	0
A	`src/coordinator/mod.rs`	260	0
A	`src/coordinator/types.rs`	215	0
M	`src/lib.rs`	1	0
M	`src/main.rs`	1	0
M	`src/node_manager.rs`	227	25
M	`src/storage/encrypted_chunk_store.rs`	2	2
M	`src/storage/storage_manager.rs`	17	0

Cargo.tomlmodified

  futures = "0.3"
  async-trait = "0.1"
++# Coordinator integration
++uuid = { version = "1.6", features = ["v4"] }
++chrono = { version = "0.4", features = ["serde"] }
++hyper = "1.0"
++
  [build-dependencies]
  tonic-build = "0.12"

Dockerfile.testadded

++# Test Dockerfile for ZephyrFS Node
++FROM rust:1.75-slim as builder
++
++# Install dependencies
++RUN apt-get update && apt-get install -y \
++    pkg-config \
++    libssl-dev \
++    protobuf-compiler \
++    && rm -rf /var/lib/apt/lists/*
++
++WORKDIR /app
++
++# Copy source code
++COPY . .
++
++# Build the application
++RUN cargo build --release
++
++# Runtime stage
++FROM debian:bookworm-slim
++
++# Install runtime dependencies
++RUN apt-get update && apt-get install -y \
++    ca-certificates \
++    wget \
++    && rm -rf /var/lib/apt/lists/*
++
++WORKDIR /app
++
++# Copy binary from builder
++COPY --from=builder /app/target/release/zephyrfs-node .
++
++# Create data directory
++RUN mkdir -p /data
++
++# Set environment variables
++ENV RUST_LOG=debug
++ENV RUST_BACKTRACE=1
++
++# Expose ports
++EXPOSE 4001 8080
++
++# Health check
++HEALTHCHECK --interval=30s --timeout=10s --start-period=30s --retries=3 \
++    CMD ./zephyrfs-node health-check || exit 1
++
++# Run the node
++CMD ["./zephyrfs-node", "start"]

build.rsmodified

  fn main() -> Result<(), Box<dyn std::error::Error>> {
++    // Generate protobuf code for node service (existing)
      tonic_build::configure()
          .build_server(true)
          .build_client(true)
              &["../zephyrfs-proto/protobuff/node.proto"],
              &["../zephyrfs-proto/protobuff"],
          )?;
++
++    // Generate protobuf code for coordinator service (new)
++    tonic_build::configure()
++        .build_server(false) // We only need the client
++        .build_client(true)
++        .compile(
++            &["../zephyrfs-proto/protobuff/coordinator.proto"],
++            &["../zephyrfs-proto/protobuff"],
++        )?;
++
++    println!("cargo:rerun-if-changed=../zephyrfs-proto/protobuff/coordinator.proto");
++    println!("cargo:rerun-if-changed=../zephyrfs-proto/protobuff/node.proto");
      Ok(())
+ }

config.test.yamladded

++# Test configuration for ZephyrFS Node
++network:
++  p2p_port: 4001
++  api_port: 8080
++  enable_mdns: true
++  bootstrap_peers: []
++  max_peers: 50
++  enable_nat_traversal: false  # Disabled for Docker testing
++
++storage:
++  data_dir: "/data"
++  max_storage: 1073741824  # 1GB
++  chunk_size: 1048576      # 1MB
++  encrypt_at_rest: true
++
++coordinator:
++  url: "http://coordinator:8080"
++  timeout: 30
++  heartbeat_interval: 10
++
++security:
++  strict_tls: false        # Disabled for local testing
++  min_peer_reputation: 0.0
++  max_unknown_peers: 10
++  enable_rate_limiting: false

docker-compose.test.ymladded

++version: '3.8'
++
++services:
++  # ZephyrFS Coordinator
++  coordinator:
++    build:
++      context: ../zephyrfs-coordinator
++      dockerfile: Dockerfile
++    ports:
++      - "8080:8080"  # gRPC port
++      - "8090:8090"  # HTTP API port
++      - "8091:8091"  # Metrics port
++    environment:
++      - LOG_LEVEL=debug
++    volumes:
++      - coordinator_data:/data
++    networks:
++      - zephyr_network
++    healthcheck:
++      test: ["CMD", "wget", "--spider", "-q", "http://localhost:8091/health"]
++      interval: 10s
++      timeout: 5s
++      retries: 3
++
++  # ZephyrFS Node 1
++  node1:
++    build:
++      context: .
++      dockerfile: Dockerfile.test
++    depends_on:
++      coordinator:
++        condition: service_healthy
++    environment:
++      - ZEPHYR_COORDINATOR_URL=http://coordinator:8080
++      - ZEPHYR_P2P_PORT=4001
++      - ZEPHYR_API_PORT=8081
++      - ZEPHYR_NODE_ID=node-1
++      - LOG_LEVEL=debug
++    volumes:
++      - node1_data:/data
++    networks:
++      - zephyr_network
++
++  # ZephyrFS Node 2
++  node2:
++    build:
++      context: .
++      dockerfile: Dockerfile.test
++    depends_on:
++      coordinator:
++        condition: service_healthy
++    environment:
++      - ZEPHYR_COORDINATOR_URL=http://coordinator:8080
++      - ZEPHYR_P2P_PORT=4002
++      - ZEPHYR_API_PORT=8082
++      - ZEPHYR_NODE_ID=node-2
++      - LOG_LEVEL=debug
++    volumes:
++      - node2_data:/data
++    networks:
++      - zephyr_network
++
++  # ZephyrFS Node 3
++  node3:
++    build:
++      context: .
++      dockerfile: Dockerfile.test
++    depends_on:
++      coordinator:
++        condition: service_healthy
++    environment:
++      - ZEPHYR_COORDINATOR_URL=http://coordinator:8080
++      - ZEPHYR_P2P_PORT=4003
++      - ZEPHYR_API_PORT=8083
++      - ZEPHYR_NODE_ID=node-3
++      - LOG_LEVEL=debug
++    volumes:
++      - node3_data:/data
++    networks:
++      - zephyr_network
++
++volumes:
++  coordinator_data:
++  node1_data:
++  node2_data:
++  node3_data:
++
++networks:
++  zephyr_network:
++    driver: bridge

src/config.rsmodified


 impl Default for CoordinatorConfig {
     fn default() -> Self {
         Self {
+            url: "http://localhost:8080".to_string(),
             timeout: 30,
             heartbeat_interval: 60,
         }

src/coordinator/client.rsadded

++use anyhow::{Result, Context};
++use std::time::Duration;
++use tonic::transport::{Channel, Endpoint};
++use tonic::{Request, Response, Status};
++use tracing::{debug, warn};
++
++use super::types::*;
++
++/// Generated gRPC client code
++pub mod coordinator_service {
++    tonic::include_proto!("zephyrfs.coordinator");
++}
++
++use coordinator_service::{
++    coordinator_service_client::CoordinatorServiceClient,
++    RegisterNodeRequest as ProtoRegisterNodeRequest,
++    RegisterNodeResponse as ProtoRegisterNodeResponse,
++    UnregisterNodeRequest as ProtoUnregisterNodeRequest,
++    UnregisterNodeResponse as ProtoUnregisterNodeResponse,
++    GetActiveNodesRequest as ProtoGetActiveNodesRequest,
++    GetActiveNodesResponse as ProtoGetActiveNodesResponse,
++    NodeHeartbeatRequest as ProtoNodeHeartbeatRequest,
++    NodeHeartbeatResponse as ProtoNodeHeartbeatResponse,
++    RegisterFileRequest as ProtoRegisterFileRequest,
++    RegisterFileResponse as ProtoRegisterFileResponse,
++    GetFileInfoRequest as ProtoGetFileInfoRequest,
++    GetFileInfoResponse as ProtoGetFileInfoResponse,
++    UpdateChunkLocationsRequest as ProtoUpdateChunkLocationsRequest,
++    UpdateChunkLocationsResponse as ProtoUpdateChunkLocationsResponse,
++    FindChunkLocationsRequest as ProtoFindChunkLocationsRequest,
++    FindChunkLocationsResponse as ProtoFindChunkLocationsResponse,
++    GetNetworkStatusRequest as ProtoGetNetworkStatusRequest,
++    GetNetworkStatusResponse as ProtoGetNetworkStatusResponse,
++};
++
++/// Coordinator gRPC client
++#[derive(Clone)]
++pub struct CoordinatorClient {
++    client: CoordinatorServiceClient<Channel>,
++}
++
++impl CoordinatorClient {
++    /// Create a new coordinator client
++    pub async fn new(coordinator_url: &str) -> Result<Self> {
++        debug!("Connecting to coordinator at: {}", coordinator_url);
++
++        let endpoint = Endpoint::from_shared(coordinator_url.to_string())
++            .context("Invalid coordinator URL")?
++            .timeout(Duration::from_secs(10))
++            .connect_timeout(Duration::from_secs(5));
++
++        let channel = endpoint.connect().await
++            .context("Failed to connect to coordinator")?;
++
++        let client = CoordinatorServiceClient::new(channel);
++
++        debug!("Successfully connected to coordinator");
++        Ok(Self { client })
++    }
++
++    /// Register node with coordinator
++    pub async fn register_node(&self, request: RegisterNodeRequest) -> Result<RegisterNodeResponse> {
++        let proto_request = ProtoRegisterNodeRequest {
++            node_id: request.node_id,
++            addresses: request.addresses,
++            storage_capacity: request.storage_capacity,
++            capabilities: request.capabilities,
++        };
++
++        let response = self.client.clone()
++            .register_node(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        Ok(RegisterNodeResponse {
++            success: response.success,
++            message: response.message,
++            assigned_node_id: response.assigned_node_id,
++            bootstrap_peers: response.bootstrap_peers,
++        })
++    }
++
++    /// Unregister node from coordinator
++    pub async fn unregister_node(&self, request: UnregisterNodeRequest) -> Result<UnregisterNodeResponse> {
++        let proto_request = ProtoUnregisterNodeRequest {
++            node_id: request.node_id,
++            reason: request.reason,
++        };
++
++        let response = self.client.clone()
++            .unregister_node(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        Ok(UnregisterNodeResponse {
++            success: response.success,
++            message: response.message,
++        })
++    }
++
++    /// Get active nodes from coordinator
++    pub async fn get_active_nodes(&self, request: GetActiveNodesRequest) -> Result<GetActiveNodesResponse> {
++        let proto_request = ProtoGetActiveNodesRequest {
++            limit: request.limit,
++            exclude_nodes: request.exclude_nodes,
++        };
++
++        let response = self.client.clone()
++            .get_active_nodes(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        let nodes = response.nodes.into_iter()
++            .map(|node| NodeStatus {
++                node_id: node.node_id,
++                addresses: node.addresses,
++                stats: node.stats.map(|stats| NodeStats {
++                    storage_used: stats.storage_used,
++                    storage_available: stats.storage_available,
++                    chunks_stored: stats.chunks_stored,
++                    bandwidth_up: stats.bandwidth_up,
++                    bandwidth_down: stats.bandwidth_down,
++                    cpu_usage: stats.cpu_usage,
++                    memory_usage: stats.memory_usage,
++                    uptime_seconds: stats.uptime_seconds,
++                }),
++                last_heartbeat: node.last_heartbeat,
++                status: node.status,
++            })
++            .collect();
++
++        Ok(GetActiveNodesResponse {
++            nodes,
++            total_nodes: response.total_nodes,
++        })
++    }
++
++    /// Send heartbeat to coordinator
++    pub async fn node_heartbeat(&self, request: NodeHeartbeatRequest) -> Result<NodeHeartbeatResponse> {
++        let proto_stats = request.stats.map(|stats| coordinator_service::NodeStats {
++            storage_used: stats.storage_used,
++            storage_available: stats.storage_available,
++            chunks_stored: stats.chunks_stored,
++            bandwidth_up: stats.bandwidth_up,
++            bandwidth_down: stats.bandwidth_down,
++            cpu_usage: stats.cpu_usage,
++            memory_usage: stats.memory_usage,
++            uptime_seconds: stats.uptime_seconds,
++        });
++
++        let proto_request = ProtoNodeHeartbeatRequest {
++            node_id: request.node_id,
++            stats: proto_stats,
++        };
++
++        let response = self.client.clone()
++            .node_heartbeat(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        Ok(NodeHeartbeatResponse {
++            success: response.success,
++            message: response.message,
++            tasks: response.tasks,
++        })
++    }
++
++    /// Register file with coordinator
++    pub async fn register_file(&self, request: RegisterFileRequest) -> Result<RegisterFileResponse> {
++        let proto_chunks = request.chunks.into_iter()
++            .map(|chunk| coordinator_service::ChunkMetadata {
++                chunk_id: chunk.chunk_id,
++                hash: chunk.hash,
++                size: chunk.size,
++                index: chunk.index,
++            })
++            .collect();
++
++        let proto_request = ProtoRegisterFileRequest {
++            file_id: request.file_id,
++            file_name: request.file_name,
++            file_size: request.file_size,
++            file_hash: request.file_hash,
++            chunks: proto_chunks,
++            owner_node_id: request.owner_node_id,
++        };
++
++        let response = self.client.clone()
++            .register_file(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        let chunk_placements = response.chunk_placements.into_iter()
++            .map(|placement| ChunkPlacement {
++                chunk_id: placement.chunk_id,
++                target_nodes: placement.target_nodes,
++                replication_factor: placement.replication_factor,
++            })
++            .collect();
++
++        Ok(RegisterFileResponse {
++            success: response.success,
++            message: response.message,
++            chunk_placements,
++        })
++    }
++
++    /// Get file info from coordinator
++    pub async fn get_file_info(&self, request: GetFileInfoRequest) -> Result<GetFileInfoResponse> {
++        let proto_request = ProtoGetFileInfoRequest {
++            file_id: request.file_id,
++        };
++
++        let response = self.client.clone()
++            .get_file_info(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        let file_info = response.file_info.map(|info| FileRecord {
++            file_id: info.file_id,
++            file_name: info.file_name,
++            file_size: info.file_size,
++            file_hash: info.file_hash,
++            chunks: info.chunks.into_iter()
++                .map(|chunk| ChunkRecord {
++                    chunk_id: chunk.chunk_id,
++                    hash: chunk.hash,
++                    size: chunk.size,
++                    index: chunk.index,
++                    stored_at_nodes: chunk.stored_at_nodes,
++                    replication_count: chunk.replication_count,
++                })
++                .collect(),
++            owner_node_id: info.owner_node_id,
++            created_at: info.created_at,
++            last_accessed: info.last_accessed,
++        });
++
++        Ok(GetFileInfoResponse {
++            success: response.success,
++            message: response.message,
++            file_info,
++        })
++    }
++
++    /// Update chunk locations
++    pub async fn update_chunk_locations(&self, request: UpdateChunkLocationsRequest) -> Result<UpdateChunkLocationsResponse> {
++        let proto_request = ProtoUpdateChunkLocationsRequest {
++            chunk_id: request.chunk_id,
++            node_ids: request.node_ids,
++            operation: request.operation,
++        };
++
++        let response = self.client.clone()
++            .update_chunk_locations(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        Ok(UpdateChunkLocationsResponse {
++            success: response.success,
++            message: response.message,
++        })
++    }
++
++    /// Find chunk locations
++    pub async fn find_chunk_locations(&self, request: FindChunkLocationsRequest) -> Result<FindChunkLocationsResponse> {
++        let proto_request = ProtoFindChunkLocationsRequest {
++            chunk_id: request.chunk_id,
++            preferred_count: request.preferred_count,
++        };
++
++        let response = self.client.clone()
++            .find_chunk_locations(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        Ok(FindChunkLocationsResponse {
++            success: response.success,
++            message: response.message,
++            node_ids: response.node_ids,
++            node_addresses: response.node_addresses,
++        })
++    }
++
++    /// Get network status
++    pub async fn get_network_status(&self, request: GetNetworkStatusRequest) -> Result<GetNetworkStatusResponse> {
++        let proto_request = ProtoGetNetworkStatusRequest {};
++
++        let response = self.client.clone()
++            .get_network_status(Request::new(proto_request))
++            .await
++            .context("gRPC call failed")?
++            .into_inner();
++
++        let network_stats = response.network_stats.map(|stats| NetworkStats {
++            total_nodes: stats.total_nodes,
++            active_nodes: stats.active_nodes,
++            total_storage_capacity: stats.total_storage_capacity,
++            total_storage_used: stats.total_storage_used,
++            total_files: stats.total_files,
++            total_chunks: stats.total_chunks,
++            average_node_uptime: stats.average_node_uptime,
++            network_uptime_seconds: stats.network_uptime_seconds,
++        });
++
++        let active_nodes = response.active_nodes.into_iter()
++            .map(|node| NodeStatus {
++                node_id: node.node_id,
++                addresses: node.addresses,
++                stats: node.stats.map(|stats| NodeStats {
++                    storage_used: stats.storage_used,
++                    storage_available: stats.storage_available,
++                    chunks_stored: stats.chunks_stored,
++                    bandwidth_up: stats.bandwidth_up,
++                    bandwidth_down: stats.bandwidth_down,
++                    cpu_usage: stats.cpu_usage,
++                    memory_usage: stats.memory_usage,
++                    uptime_seconds: stats.uptime_seconds,
++                }),
++                last_heartbeat: node.last_heartbeat,
++                status: node.status,
++            })
++            .collect();
++
++        Ok(GetNetworkStatusResponse {
++            network_stats,
++            active_nodes,
++            timestamp: response.timestamp,
++        })
++    }
++}

src/coordinator/mod.rsadded

++use anyhow::{Result, Context};
++use std::time::{Duration, SystemTime, UNIX_EPOCH};
++use tokio::time::{interval, sleep};
++use tracing::{debug, info, warn, error};
++use uuid::Uuid;
++
++pub mod client;
++pub mod types;
++
++pub use client::CoordinatorClient;
++pub use types::*;
++
++/// Coordinator integration for node registration and coordination
++pub struct CoordinatorManager {
++    client: CoordinatorClient,
++    node_id: String,
++    coordinator_url: String,
++    heartbeat_interval: Duration,
++    registration_status: RegistrationStatus,
++}
++
++#[derive(Debug, Clone)]
++pub enum RegistrationStatus {
++    NotRegistered,
++    Registering,
++    Registered,
++    Failed(String),
++}
++
++impl CoordinatorManager {
++    /// Create a new coordinator manager
++    pub async fn new(coordinator_url: String) -> Result<Self> {
++        let client = CoordinatorClient::new(&coordinator_url).await
++            .context("Failed to create coordinator client")?;
++
++        let node_id = Uuid::new_v4().to_string();
++        let heartbeat_interval = Duration::from_secs(10);
++
++        Ok(Self {
++            client,
++            node_id,
++            coordinator_url,
++            heartbeat_interval,
++            registration_status: RegistrationStatus::NotRegistered,
++        })
++    }
++
++    /// Register this node with the coordinator
++    pub async fn register_node(
++        &mut self,
++        addresses: Vec<String>,
++        storage_capacity: u64,
++        capabilities: std::collections::HashMap<String, String>,
++    ) -> Result<RegisterNodeResponse> {
++        info!("Registering node {} with coordinator at {}", self.node_id, self.coordinator_url);
++        self.registration_status = RegistrationStatus::Registering;
++
++        let request = RegisterNodeRequest {
++            node_id: self.node_id.clone(),
++            addresses,
++            storage_capacity: storage_capacity as i64,
++            capabilities,
++        };
++
++        match self.client.register_node(request).await {
++            Ok(response) => {
++                if response.success {
++                    self.registration_status = RegistrationStatus::Registered;
++                    info!("Successfully registered with coordinator. Assigned ID: {}", response.assigned_node_id);
++
++                    // Update node ID if coordinator assigned a different one
++                    if !response.assigned_node_id.is_empty() {
++                        self.node_id = response.assigned_node_id.clone();
++                    }
++
++                    let success_response = RegisterNodeResponse {
++                        success: response.success,
++                        message: response.message,
++                        assigned_node_id: response.assigned_node_id,
++                        bootstrap_peers: response.bootstrap_peers,
++                    };
++                    Ok(success_response)
++                } else {
++                    let error_msg = format!("Registration failed: {}", response.message);
++                    self.registration_status = RegistrationStatus::Failed(error_msg.clone());
++                    warn!("{}", error_msg);
++                    Ok(response)
++                }
++            }
++            Err(e) => {
++                let error_msg = format!("Failed to register with coordinator: {}", e);
++                self.registration_status = RegistrationStatus::Failed(error_msg.clone());
++                error!("{}", error_msg);
++                Err(e)
++            }
++        }
++    }
++
++    /// Start heartbeat loop
++    pub async fn start_heartbeat(&self, stats_provider: impl Fn() -> NodeStats + Send + 'static) {
++        let client = self.client.clone();
++        let node_id = self.node_id.clone();
++        let heartbeat_interval = self.heartbeat_interval;
++
++        tokio::spawn(async move {
++            let mut interval = interval(heartbeat_interval);
++
++            loop {
++                interval.tick().await;
++
++                let stats = stats_provider();
++                let request = NodeHeartbeatRequest {
++                    node_id: node_id.clone(),
++                    stats: Some(stats),
++                };
++
++                match client.node_heartbeat(request).await {
++                    Ok(response) => {
++                        if response.success {
++                            debug!("Heartbeat sent successfully");
++                            if !response.tasks.is_empty() {
++                                debug!("Coordinator assigned {} tasks", response.tasks.len());
++                                // TODO: Handle assigned tasks
++                            }
++                        } else {
++                            warn!("Heartbeat failed: {}", response.message);
++                        }
++                    }
++                    Err(e) => {
++                        warn!("Failed to send heartbeat: {}", e);
++                        // TODO: Implement exponential backoff
++                        sleep(Duration::from_secs(5)).await;
++                    }
++                }
++            }
++        });
++    }
++
++    /// Register a file with the coordinator
++    pub async fn register_file(
++        &self,
++        file_id: String,
++        file_name: String,
++        file_size: u64,
++        file_hash: String,
++        chunks: Vec<ChunkMetadata>,
++    ) -> Result<RegisterFileResponse> {
++        debug!("Registering file {} with coordinator", file_id);
++
++        let request = RegisterFileRequest {
++            file_id,
++            file_name,
++            file_size: file_size as i64,
++            file_hash,
++            chunks,
++            owner_node_id: self.node_id.clone(),
++        };
++
++        self.client.register_file(request).await
++            .context("Failed to register file with coordinator")
++    }
++
++    /// Find chunk locations from coordinator
++    pub async fn find_chunk_locations(&self, chunk_id: String, preferred_count: i32) -> Result<FindChunkLocationsResponse> {
++        debug!("Finding locations for chunk {} from coordinator", chunk_id);
++
++        let request = FindChunkLocationsRequest {
++            chunk_id,
++            preferred_count,
++        };
++
++        self.client.find_chunk_locations(request).await
++            .context("Failed to find chunk locations from coordinator")
++    }
++
++    /// Get active nodes from coordinator
++    pub async fn get_active_nodes(&self, limit: Option<i32>, exclude_nodes: Vec<String>) -> Result<GetActiveNodesResponse> {
++        debug!("Getting active nodes from coordinator");
++
++        let request = GetActiveNodesRequest {
++            limit: limit.unwrap_or(10),
++            exclude_nodes,
++        };
++
++        self.client.get_active_nodes(request).await
++            .context("Failed to get active nodes from coordinator")
++    }
++
++    /// Get network status from coordinator
++    pub async fn get_network_status(&self) -> Result<GetNetworkStatusResponse> {
++        debug!("Getting network status from coordinator");
++
++        let request = GetNetworkStatusRequest {};
++
++        self.client.get_network_status(request).await
++            .context("Failed to get network status from coordinator")
++    }
++
++    /// Update chunk locations
++    pub async fn update_chunk_locations(
++        &self,
++        chunk_id: String,
++        node_ids: Vec<String>,
++        operation: String,
++    ) -> Result<UpdateChunkLocationsResponse> {
++        debug!("Updating chunk locations for {} (operation: {})", chunk_id, operation);
++
++        let request = UpdateChunkLocationsRequest {
++            chunk_id,
++            node_ids,
++            operation,
++        };
++
++        self.client.update_chunk_locations(request).await
++            .context("Failed to update chunk locations")
++    }
++
++    /// Unregister this node from coordinator
++    pub async fn unregister_node(&mut self, reason: Option<String>) -> Result<UnregisterNodeResponse> {
++        info!("Unregistering node {} from coordinator", self.node_id);
++
++        let request = UnregisterNodeRequest {
++            node_id: self.node_id.clone(),
++            reason: reason.unwrap_or_else(|| "Normal shutdown".to_string()),
++        };
++
++        match self.client.unregister_node(request).await {
++            Ok(response) => {
++                if response.success {
++                    self.registration_status = RegistrationStatus::NotRegistered;
++                    info!("Successfully unregistered from coordinator");
++                } else {
++                    warn!("Unregistration failed: {}", response.message);
++                }
++                Ok(response)
++            }
++            Err(e) => {
++                error!("Failed to unregister from coordinator: {}", e);
++                Err(e)
++            }
++        }
++    }
++
++    /// Get current registration status
++    pub fn get_registration_status(&self) -> &RegistrationStatus {
++        &self.registration_status
++    }
++
++    /// Get node ID
++    pub fn get_node_id(&self) -> &str {
++        &self.node_id
++    }
++}
++
++/// Convert system time to Unix timestamp
++pub fn system_time_to_unix_timestamp(time: SystemTime) -> i64 {
++    time.duration_since(UNIX_EPOCH)
++        .unwrap_or_default()
++        .as_secs() as i64
++}

src/coordinator/types.rsadded

++use std::collections::HashMap;
++
++/// Request to register a node with the coordinator
++#[derive(Debug, Clone)]
++pub struct RegisterNodeRequest {
++    pub node_id: String,
++    pub addresses: Vec<String>,
++    pub storage_capacity: i64,
++    pub capabilities: HashMap<String, String>,
++}
++
++/// Response from node registration
++#[derive(Debug, Clone)]
++pub struct RegisterNodeResponse {
++    pub success: bool,
++    pub message: String,
++    pub assigned_node_id: String,
++    pub bootstrap_peers: Vec<String>,
++}
++
++/// Request to unregister a node
++#[derive(Debug, Clone)]
++pub struct UnregisterNodeRequest {
++    pub node_id: String,
++    pub reason: String,
++}
++
++/// Response from node unregistration
++#[derive(Debug, Clone)]
++pub struct UnregisterNodeResponse {
++    pub success: bool,
++    pub message: String,
++}
++
++/// Request to get active nodes
++#[derive(Debug, Clone)]
++pub struct GetActiveNodesRequest {
++    pub limit: i32,
++    pub exclude_nodes: Vec<String>,
++}
++
++/// Response with active nodes
++#[derive(Debug, Clone)]
++pub struct GetActiveNodesResponse {
++    pub nodes: Vec<NodeStatus>,
++    pub total_nodes: i32,
++}
++
++/// Node heartbeat request
++#[derive(Debug, Clone)]
++pub struct NodeHeartbeatRequest {
++    pub node_id: String,
++    pub stats: Option<NodeStats>,
++}
++
++/// Node heartbeat response
++#[derive(Debug, Clone)]
++pub struct NodeHeartbeatResponse {
++    pub success: bool,
++    pub message: String,
++    pub tasks: Vec<String>,
++}
++
++/// Request to register a file
++#[derive(Debug, Clone)]
++pub struct RegisterFileRequest {
++    pub file_id: String,
++    pub file_name: String,
++    pub file_size: i64,
++    pub file_hash: String,
++    pub chunks: Vec<ChunkMetadata>,
++    pub owner_node_id: String,
++}
++
++/// Response from file registration
++#[derive(Debug, Clone)]
++pub struct RegisterFileResponse {
++    pub success: bool,
++    pub message: String,
++    pub chunk_placements: Vec<ChunkPlacement>,
++}
++
++/// Request to get file information
++#[derive(Debug, Clone)]
++pub struct GetFileInfoRequest {
++    pub file_id: String,
++}
++
++/// Response with file information
++#[derive(Debug, Clone)]
++pub struct GetFileInfoResponse {
++    pub success: bool,
++    pub message: String,
++    pub file_info: Option<FileRecord>,
++}
++
++/// Request to update chunk locations
++#[derive(Debug, Clone)]
++pub struct UpdateChunkLocationsRequest {
++    pub chunk_id: String,
++    pub node_ids: Vec<String>,
++    pub operation: String, // "add" or "remove"
++}
++
++/// Response from chunk location update
++#[derive(Debug, Clone)]
++pub struct UpdateChunkLocationsResponse {
++    pub success: bool,
++    pub message: String,
++}
++
++/// Request to find chunk locations
++#[derive(Debug, Clone)]
++pub struct FindChunkLocationsRequest {
++    pub chunk_id: String,
++    pub preferred_count: i32,
++}
++
++/// Response with chunk locations
++#[derive(Debug, Clone)]
++pub struct FindChunkLocationsResponse {
++    pub success: bool,
++    pub message: String,
++    pub node_ids: Vec<String>,
++    pub node_addresses: Vec<String>,
++}
++
++/// Request to get network status
++#[derive(Debug, Clone)]
++pub struct GetNetworkStatusRequest {}
++
++/// Response with network status
++#[derive(Debug, Clone)]
++pub struct GetNetworkStatusResponse {
++    pub network_stats: Option<NetworkStats>,
++    pub active_nodes: Vec<NodeStatus>,
++    pub timestamp: i64,
++}
++
++/// Node status information
++#[derive(Debug, Clone)]
++pub struct NodeStatus {
++    pub node_id: String,
++    pub addresses: Vec<String>,
++    pub stats: Option<NodeStats>,
++    pub last_heartbeat: i64,
++    pub status: String, // "active", "inactive", "maintenance"
++}
++
++/// Node statistics
++#[derive(Debug, Clone)]
++pub struct NodeStats {
++    pub storage_used: i64,
++    pub storage_available: i64,
++    pub chunks_stored: i64,
++    pub bandwidth_up: i64,
++    pub bandwidth_down: i64,
++    pub cpu_usage: f64,
++    pub memory_usage: f64,
++    pub uptime_seconds: i64,
++}
++
++/// Chunk metadata
++#[derive(Debug, Clone)]
++pub struct ChunkMetadata {
++    pub chunk_id: String,
++    pub hash: String,
++    pub size: i64,
++    pub index: i32,
++}
++
++/// Chunk placement information
++#[derive(Debug, Clone)]
++pub struct ChunkPlacement {
++    pub chunk_id: String,
++    pub target_nodes: Vec<String>,
++    pub replication_factor: i32,
++}
++
++/// File record information
++#[derive(Debug, Clone)]
++pub struct FileRecord {
++    pub file_id: String,
++    pub file_name: String,
++    pub file_size: i64,
++    pub file_hash: String,
++    pub chunks: Vec<ChunkRecord>,
++    pub owner_node_id: String,
++    pub created_at: i64,
++    pub last_accessed: i64,
++}
++
++/// Chunk record with location information
++#[derive(Debug, Clone)]
++pub struct ChunkRecord {
++    pub chunk_id: String,
++    pub hash: String,
++    pub size: i64,
++    pub index: i32,
++    pub stored_at_nodes: Vec<String>,
++    pub replication_count: i32,
++}
++
++/// Network statistics
++#[derive(Debug, Clone)]
++pub struct NetworkStats {
++    pub total_nodes: i32,
++    pub active_nodes: i32,
++    pub total_storage_capacity: i64,
++    pub total_storage_used: i64,
++    pub total_files: i64,
++    pub total_chunks: i64,
++    pub average_node_uptime: f64,
++    pub network_uptime_seconds: i64,
++}

src/lib.rsmodified


 pub mod protocol;
 pub mod node_manager;
 pub mod crypto;
+pub mod coordinator;
 
 pub use crypto::{
     ZephyrCrypto, CryptoParams, ScryptParams, AesParams, HashParams,

src/main.rsmodified


 mod protocol;
 mod node_manager;
 mod crypto;
+mod coordinator;
 
 #[cfg(test)]
 mod integration_tests;

src/node_manager.rsmodified

  use crate::config::Config;
  use crate::network::{NetworkManager, message_handler::{ZephyrMessage, NodeInfo}};
  use crate::storage::{StorageManager, StorageConfig as StorageManagerConfig};
++use crate::coordinator::{CoordinatorManager, RegistrationStatus};
  /// Integrated node manager coordinating networking and storage
  ///
      /// Storage layer manager
      pub storage_manager: Arc<StorageManager>,
++    /// Coordinator manager for network coordination
++    coordinator_manager: Option<CoordinatorManager>,
++
      /// Configuration
      config: Config,
          let network_manager = NetworkManager::new(config.clone()).await
              .context("Failed to initialize network manager")?;
++        // Initialize coordinator manager if URL is provided
++        let coordinator_manager = if !config.coordinator.url.is_empty() {
++            match CoordinatorManager::new(config.coordinator.url.clone()).await {
++                Ok(manager) => {
++                    info!("Successfully connected to coordinator at {}", config.coordinator.url);
++                    Some(manager)
++                }
++                Err(e) => {
++                    warn!("Failed to connect to coordinator at {}: {}. Running in standalone mode.", config.coordinator.url, e);
++                    None
++                }
++            }
++        } else {
++            info!("No coordinator URL configured. Running in standalone mode.");
++            None
++        };
++
          let node_stats = Arc::new(RwLock::new(NodeStats {
              chunks_served: 0,
              chunks_retrieved: 0,
          Ok(Self {
              network_manager,
              storage_manager,
++            coordinator_manager,
              config,
              message_rx,
              message_tx,
          self.network_manager.start().await
              .context("Failed to start network manager")?;
++        // Register with coordinator if available
++        if self.coordinator_manager.is_some() {
++            let node_status = self.get_node_status().await;
++            let addresses = vec![
++                format!("{}:{}", "127.0.0.1", self.config.network.p2p_port),
++                format!("{}:{}", "127.0.0.1", self.config.network.api_port),
++            ];
++
++            let mut capabilities = std::collections::HashMap::new();
++            capabilities.insert("version".to_string(), node_status.version);
++            capabilities.insert("storage".to_string(), "true".to_string());
++            capabilities.insert("encryption".to_string(), "true".to_string());
++
++            if let Some(coordinator) = self.coordinator_manager.as_mut() {
++                let response = coordinator.register_node(
++                    addresses,
++                    node_status.storage_capacity,
++                    capabilities,
++                ).await?;
++
++                if response.success {
++                    info!("Successfully registered with coordinator. Node ID: {}", coordinator.get_node_id());
++                    if !response.bootstrap_peers.is_empty() {
++                        info!("Received {} bootstrap peers from coordinator", response.bootstrap_peers.len());
++                        // TODO: Connect to bootstrap peers
++                    }
++                } else {
++                    warn!("Failed to register with coordinator: {}", response.message);
++                }
++            }
++
++            self.start_coordinator_heartbeat().await;
++        }
++
          // Start message processing loop
          self.start_message_processing().await;
+             }
+         }
++        // Register file with coordinator if available
++        if let Err(e) = self.register_file_with_coordinator(file_id, &file_hash, data.len() as u64, filename).await {
++            warn!("Failed to register file with coordinator: {}", e);
++        }
++
          // Announce file availability to peers
          if let Err(e) = self.announce_file_to_peers(file_id, &file_hash).await {
              warn!("Failed to announce file to peers: {}", e);
          // Calculate uptime
          let uptime_seconds = stats.start_time.elapsed().as_secs();
++        let node_id = if let Some(coordinator) = &self.coordinator_manager {
++            coordinator.get_node_id().to_string()
++        } else {
++            self.config.node_id.clone().unwrap_or_else(|| "local_node".to_string())
++        };
++
          NodeStatus {
--            node_id: "local_node".to_string(), // TODO: Generate proper node ID
++            node_id,
              version: env!("CARGO_PKG_VERSION").to_string(),
              uptime_seconds,
              peer_connections: stats.peer_connections,
      pub async fn shutdown(&mut self) -> Result<()> {
          info!("Shutting down ZephyrFS node");
++        // Unregister from coordinator if connected
++        if let Some(coordinator) = &mut self.coordinator_manager {
++            if let Err(e) = coordinator.unregister_node(Some("Normal shutdown".to_string())).await {
++                warn!("Failed to unregister from coordinator: {}", e);
++            }
++        }
++
          // Shutdown network manager
          self.network_manager.shutdown().await
              .context("Failed to shutdown network manager")?;
          Ok(())
+     }
++
++    /// Register this node with the coordinator
++    async fn register_with_coordinator(&self, coordinator: &mut CoordinatorManager) -> Result<()> {
++        info!("Registering node with coordinator");
++
++        let node_status = self.get_node_status().await;
++        let addresses = vec![
++            format!("{}:{}", "127.0.0.1", self.config.network.p2p_port),
++            format!("{}:{}", "127.0.0.1", self.config.network.api_port),
++        ];
++
++        let mut capabilities = std::collections::HashMap::new();
++        capabilities.insert("version".to_string(), node_status.version);
++        capabilities.insert("storage".to_string(), "true".to_string());
++        capabilities.insert("encryption".to_string(), "true".to_string());
++
++        let response = coordinator.register_node(
++            addresses,
++            node_status.storage_capacity,
++            capabilities,
++        ).await?;
++
++        if response.success {
++            info!("Successfully registered with coordinator. Node ID: {}", coordinator.get_node_id());
++            if !response.bootstrap_peers.is_empty() {
++                info!("Received {} bootstrap peers from coordinator", response.bootstrap_peers.len());
++                // TODO: Connect to bootstrap peers
++            }
++        } else {
++            warn!("Failed to register with coordinator: {}", response.message);
++        }
++
++        Ok(())
++    }
++
++    /// Start coordinator heartbeat loop
++    async fn start_coordinator_heartbeat(&self) {
++        if let Some(coordinator) = &self.coordinator_manager {
++            let node_stats = Arc::clone(&self.node_stats);
++            let storage_manager = Arc::clone(&self.storage_manager);
++
++            coordinator.start_heartbeat(move || {
++                let stats = tokio::task::block_in_place(|| {
++                    tokio::runtime::Handle::current().block_on(async {
++                        let node_stats = node_stats.read().await;
++                        let capacity_info = storage_manager.get_capacity_info().await;
++                        let uptime = node_stats.start_time.elapsed().as_secs() as i64;
++
++                        crate::coordinator::types::NodeStats {
++                            storage_used: capacity_info.used_space as i64,
++                            storage_available: capacity_info.available_space as i64,
++                            chunks_stored: capacity_info.file_count as i64, // Approximation
++                            bandwidth_up: node_stats.bytes_sent as i64,
++                            bandwidth_down: node_stats.bytes_received as i64,
++                            cpu_usage: 0.0, // TODO: Implement CPU monitoring
++                            memory_usage: 0.0, // TODO: Implement memory monitoring
++                            uptime_seconds: uptime,
++                        }
++                    })
++                });
++                stats
++            }).await;
++
++            info!("Started coordinator heartbeat");
++        }
++    }
++
++    /// Register a file with the coordinator if available
++    async fn register_file_with_coordinator(&self, file_id: &str, file_hash: &str, file_size: u64, filename: &str) -> Result<()> {
++        if let Some(coordinator) = &self.coordinator_manager {
++            // Get file chunks from storage manager
++            let chunks = match self.storage_manager.get_file_chunks(file_id).await {
++                Ok(Some(chunk_list)) => {
++                    chunk_list.into_iter().enumerate().map(|(index, chunk_id)| {
++                        crate::coordinator::types::ChunkMetadata {
++                            chunk_id: chunk_id.clone(),
++                            hash: chunk_id, // For now, use chunk_id as hash
++                            size: self.config.storage.chunk_size as i64, // Default chunk size
++                            index: index as i32,
++                        }
++                    }).collect()
++                }
++                _ => Vec::new(),
++            };
++
++            let response = coordinator.register_file(
++                file_id.to_string(),
++                filename.to_string(),
++                file_size,
++                file_hash.to_string(),
++                chunks,
++            ).await?;
++
++            if response.success {
++                debug!("Successfully registered file {} with coordinator", file_id);
++                if !response.chunk_placements.is_empty() {
++                    debug!("Coordinator provided {} chunk placement recommendations", response.chunk_placements.len());
++                    // TODO: Handle chunk placement recommendations
++                }
++            } else {
++                warn!("Failed to register file with coordinator: {}", response.message);
++            }
++        }
++
++        Ok(())
++    }
++
++    /// Find chunk locations using coordinator
++    async fn find_chunk_locations_via_coordinator(&self, chunk_id: &str) -> Result<Vec<String>> {
++        if let Some(coordinator) = &self.coordinator_manager {
++            let response = coordinator.find_chunk_locations(chunk_id.to_string(), 3).await?;
++
++            if response.success {
++                debug!("Found {} locations for chunk {} via coordinator", response.node_addresses.len(), chunk_id);
++                Ok(response.node_addresses)
++            } else {
++                debug!("Coordinator couldn't find locations for chunk {}: {}", chunk_id, response.message);
++                Ok(Vec::new())
++            }
++        } else {
++            Ok(Vec::new())
++        }
++    }
++
++    /// Get coordinator registration status
++    pub fn get_coordinator_status(&self) -> Option<&RegistrationStatus> {
++        self.coordinator_manager.as_ref().map(|c| c.get_registration_status())
++    }
+ }
  /// Comprehensive node status information

src/storage/encrypted_chunk_store.rsmodified


     #[tokio::test]
     async fn test_encrypted_chunk_store_creation() {
         let temp_dir = tempdir().unwrap();
+        let store = EncryptedChunkStore::new(temp_dir.path()).unwrap();
         let stats = store.get_encrypted_stats().await;
         
         assert_eq!(stats.total_encrypted_chunks, 0);

     #[tokio::test]
     async fn test_encrypted_chunk_deduplication() {
         let temp_dir = tempdir().unwrap();
+        let store = EncryptedChunkStore::new(temp_dir.path()).unwrap();
         
         let encrypted_data = EncryptedData {
             segment_index: 0,

src/storage/storage_manager.rsmodified

              None
+         }
+     }
++
++    /// Get chunk IDs for a file (needed for coordinator integration)
++    ///
++    /// Returns the list of chunk IDs that comprise the given file
++    pub async fn get_file_chunks(&self, file_id: &str) -> Result<Option<Vec<String>>> {
++        debug!("Getting chunk list for file: {}", file_id);
++
++        match self.metadata_store.get_file(file_id).await? {
++            Some(metadata) => {
++                Ok(Some(metadata.chunk_ids))
++            }
++            None => {
++                debug!("File not found: {}", file_id);
++                Ok(None)
++            }
++        }
++    }
+ }
  #[cfg(test)]

`@@ -132,7 +132,7 @@` impl Default for StorageConfig {
132	impl Default for CoordinatorConfig {	132	impl Default for CoordinatorConfig {
133	fn default() -> Self {	133	fn default() -> Self {
134	Self {	134	Self {
135	- url: "http://localhost:9090".to_string(),	135	+ url: "http://localhost:8080".to_string(),
136	timeout: 30,	136	timeout: 30,
137	heartbeat_interval: 60,	137	heartbeat_interval: 60,
138	}	138	}

`@@ -504,7 +504,7 @@` mod tests {
504	#[tokio::test]	504	#[tokio::test]
505	async fn test_encrypted_chunk_store_creation() {	505	async fn test_encrypted_chunk_store_creation() {
506	let temp_dir = tempdir().unwrap();	506	let temp_dir = tempdir().unwrap();
507	- let store = EncryptedChunkStore::new(temp_dir.path()).await.unwrap();	507	+ let store = EncryptedChunkStore::new(temp_dir.path()).unwrap();
508	let stats = store.get_encrypted_stats().await;	508	let stats = store.get_encrypted_stats().await;
509		509
510	assert_eq!(stats.total_encrypted_chunks, 0);	510	assert_eq!(stats.total_encrypted_chunks, 0);
`@@ -515,7 +515,7 @@` mod tests {
515	#[tokio::test]	515	#[tokio::test]
516	async fn test_encrypted_chunk_deduplication() {	516	async fn test_encrypted_chunk_deduplication() {
517	let temp_dir = tempdir().unwrap();	517	let temp_dir = tempdir().unwrap();
518	- let store = EncryptedChunkStore::new(temp_dir.path()).await.unwrap();	518	+ let store = EncryptedChunkStore::new(temp_dir.path()).unwrap();
519		519
520	let encrypted_data = EncryptedData {	520	let encrypted_data = EncryptedData {
521	segment_index: 0,	521	segment_index: 0,

`@@ -9,6 +9,7 @@` pub mod storage;
9	pub mod protocol;	9	pub mod protocol;
10	pub mod node_manager;	10	pub mod node_manager;
11	pub mod crypto;	11	pub mod crypto;
		12	+pub mod coordinator;
12		13
13	pub use crypto::{	14	pub use crypto::{
14	ZephyrCrypto, CryptoParams, ScryptParams, AesParams, HashParams,	15	ZephyrCrypto, CryptoParams, ScryptParams, AesParams, HashParams,

`@@ -9,6 +9,7 @@` mod storage;
9	mod protocol;	9	mod protocol;
10	mod node_manager;	10	mod node_manager;
11	mod crypto;	11	mod crypto;
		12	+mod coordinator;
12		13
13	#[cfg(test)]	14	#[cfg(test)]
14	mod integration_tests;	15	mod integration_tests;