zephyrfs-node Public

Watch 0 Fork 0 Star 0
Rust · 33750 bytes Raw Blame History
1 //! Bandwidth-Optimized Recovery Algorithms
2 //!
3 //! Efficient algorithms for data recovery that minimize bandwidth usage
4 //! while maximizing recovery speed and reliability
5
6 use anyhow::Result;
7 use serde::{Deserialize, Serialize};
8 use std::collections::{HashMap, BTreeMap, HashSet};
9 use chrono::{DateTime, Utc, Duration};
10
11 use crate::economics::earnings_calculator::GeographicRegion;
12 use super::reed_solomon::{EncodedChunk, ReconstructionRequest};
13
14 /// Bandwidth-optimized recovery manager
15 #[derive(Debug, Clone)]
16 pub struct RecoveryOptimizer {
17 /// Network topology information
18 pub network_topology: NetworkTopology,
19 /// Bandwidth optimization strategies
20 pub optimization_strategies: Vec<OptimizationStrategy>,
21 /// Performance metrics
22 pub performance_metrics: RecoveryMetrics,
23 /// Recovery algorithms configuration
24 pub algorithms: AlgorithmConfig,
25 }
26
27 #[derive(Debug, Clone, Serialize, Deserialize)]
28 pub struct NetworkTopology {
29 /// Node connectivity information
30 pub nodes: HashMap<String, NodeInfo>,
31 /// Connection bandwidth matrix
32 pub bandwidth_matrix: HashMap<(String, String), BandwidthInfo>,
33 /// Regional connectivity
34 pub regional_links: HashMap<GeographicRegion, RegionalConnectivity>,
35 /// Network congestion levels
36 pub congestion_levels: HashMap<String, CongestionInfo>,
37 }
38
39 #[derive(Debug, Clone, Serialize, Deserialize)]
40 pub struct NodeInfo {
41 pub node_id: String,
42 pub region: GeographicRegion,
43 pub available_bandwidth_mbps: f64,
44 pub latency_profile: HashMap<String, f64>,
45 pub connection_quality: ConnectionQuality,
46 pub load_factor: f64,
47 pub active_transfers: u32,
48 }
49
50 #[derive(Debug, Clone, Serialize, Deserialize)]
51 pub struct BandwidthInfo {
52 pub theoretical_max_mbps: f64,
53 pub current_available_mbps: f64,
54 pub average_utilization: f64,
55 pub latency_ms: f64,
56 pub reliability_score: f64,
57 pub cost_per_gb: f64,
58 }
59
60 #[derive(Debug, Clone, Serialize, Deserialize)]
61 pub struct RegionalConnectivity {
62 pub region: GeographicRegion,
63 pub total_capacity_gbps: f64,
64 pub utilized_capacity_gbps: f64,
65 pub inter_region_links: HashMap<GeographicRegion, f64>,
66 pub backbone_quality: BackboneQuality,
67 }
68
69 #[derive(Debug, Clone, Serialize, Deserialize)]
70 pub struct CongestionInfo {
71 pub node_id: String,
72 pub current_load: f64,
73 pub predicted_load: f64,
74 pub congestion_trend: CongestionTrend,
75 pub time_to_clear: Option<Duration>,
76 }
77
78 #[derive(Debug, Clone, Serialize, Deserialize)]
79 pub enum CongestionTrend {
80 Increasing,
81 Stable,
82 Decreasing,
83 }
84
85 #[derive(Debug, Clone, Serialize, Deserialize)]
86 pub enum ConnectionQuality {
87 Excellent, // Fiber, low latency
88 Good, // Fast broadband
89 Average, // Standard broadband
90 Poor, // Slow/unreliable
91 }
92
93 #[derive(Debug, Clone, Serialize, Deserialize)]
94 pub enum BackboneQuality {
95 Tier1, // Top-tier internet backbone
96 Tier2, // Regional provider
97 Tier3, // Local provider
98 Satellite, // Satellite connectivity
99 }
100
101 #[derive(Debug, Clone, Serialize, Deserialize)]
102 pub enum OptimizationStrategy {
103 ParallelRecovery, // Download chunks in parallel
104 ProgressiveRecovery, // Start with most critical chunks
105 LocalityOptimized, // Prefer nearby nodes
106 LoadBalanced, // Balance load across nodes
107 CostOptimized, // Minimize transfer costs
108 LatencyOptimized, // Minimize total time
109 AdaptiveBandwidth, // Adjust based on available bandwidth
110 }
111
112 #[derive(Debug, Clone, Serialize, Deserialize)]
113 pub struct RecoveryMetrics {
114 pub total_recoveries: u64,
115 pub successful_recoveries: u64,
116 pub average_recovery_time_seconds: f64,
117 pub average_bandwidth_efficiency: f64,
118 pub total_bytes_recovered: u64,
119 pub cost_savings_percent: f64,
120 pub last_updated: DateTime<Utc>,
121 }
122
123 #[derive(Debug, Clone, Serialize, Deserialize)]
124 pub struct AlgorithmConfig {
125 pub max_parallel_streams: u32,
126 pub chunk_prefetch_count: u32,
127 pub adaptive_bandwidth_threshold: f64,
128 pub load_balancing_factor: f64,
129 pub locality_preference_weight: f64,
130 pub congestion_avoidance_enabled: bool,
131 }
132
133 #[derive(Debug, Clone, Serialize, Deserialize)]
134 pub struct RecoveryPlan {
135 pub plan_id: String,
136 pub target_chunks: Vec<String>,
137 pub recovery_steps: Vec<RecoveryStep>,
138 pub estimated_time_seconds: f64,
139 pub estimated_bandwidth_usage_mb: f64,
140 pub estimated_cost: f64,
141 pub optimization_strategy: OptimizationStrategy,
142 pub fallback_plans: Vec<FallbackPlan>,
143 }
144
145 #[derive(Debug, Clone, Serialize, Deserialize)]
146 pub struct RecoveryStep {
147 pub step_id: String,
148 pub step_type: RecoveryStepType,
149 pub source_nodes: Vec<String>,
150 pub target_chunks: Vec<String>,
151 pub estimated_duration_seconds: f64,
152 pub bandwidth_requirement_mbps: f64,
153 pub priority: RecoveryPriority,
154 pub dependencies: Vec<String>, // Step IDs this depends on
155 }
156
157 #[derive(Debug, Clone, Serialize, Deserialize)]
158 pub enum RecoveryStepType {
159 DirectTransfer, // Direct chunk download
160 ParallelTransfer, // Multiple chunks in parallel
161 ErasureReconstruct, // Reed-Solomon reconstruction
162 VerifyIntegrity, // Verify recovered data
163 Prefetch, // Preemptive chunk fetching
164 }
165
166 #[derive(Debug, Clone, Serialize, Deserialize)]
167 pub enum RecoveryPriority {
168 Critical, // Must complete first
169 High, // Important for performance
170 Normal, // Standard priority
171 Background, // Can be delayed
172 }
173
174 #[derive(Debug, Clone, Serialize, Deserialize)]
175 pub struct FallbackPlan {
176 pub fallback_id: String,
177 pub trigger_conditions: Vec<String>,
178 pub alternative_steps: Vec<RecoveryStep>,
179 pub performance_impact: f64,
180 }
181
182 impl Default for AlgorithmConfig {
183 fn default() -> Self {
184 Self {
185 max_parallel_streams: 8,
186 chunk_prefetch_count: 2,
187 adaptive_bandwidth_threshold: 0.8,
188 load_balancing_factor: 0.3,
189 locality_preference_weight: 0.6,
190 congestion_avoidance_enabled: true,
191 }
192 }
193 }
194
195 impl RecoveryOptimizer {
196 /// Create new recovery optimizer
197 pub fn new() -> Self {
198 Self {
199 network_topology: NetworkTopology {
200 nodes: HashMap::new(),
201 bandwidth_matrix: HashMap::new(),
202 regional_links: HashMap::new(),
203 congestion_levels: HashMap::new(),
204 },
205 optimization_strategies: vec![
206 OptimizationStrategy::AdaptiveBandwidth,
207 OptimizationStrategy::LoadBalanced,
208 OptimizationStrategy::LocalityOptimized,
209 ],
210 performance_metrics: RecoveryMetrics {
211 total_recoveries: 0,
212 successful_recoveries: 0,
213 average_recovery_time_seconds: 0.0,
214 average_bandwidth_efficiency: 0.0,
215 total_bytes_recovered: 0,
216 cost_savings_percent: 0.0,
217 last_updated: Utc::now(),
218 },
219 algorithms: AlgorithmConfig::default(),
220 }
221 }
222
223 /// Create optimized recovery plan
224 pub fn create_recovery_plan(
225 &self,
226 missing_chunks: &[String],
227 available_chunks: &HashMap<String, Vec<NodeLocation>>,
228 recovery_requirements: RecoveryRequirements,
229 ) -> Result<RecoveryPlan> {
230
231 // Analyze available sources
232 let source_analysis = self.analyze_chunk_sources(available_chunks)?;
233
234 // Select optimal strategy based on requirements
235 let strategy = self.select_optimization_strategy(&recovery_requirements, &source_analysis)?;
236
237 // Generate recovery steps
238 let recovery_steps = self.generate_recovery_steps(
239 missing_chunks,
240 available_chunks,
241 &strategy,
242 &recovery_requirements,
243 )?;
244
245 // Calculate estimates
246 let (estimated_time, estimated_bandwidth, estimated_cost) =
247 self.calculate_recovery_estimates(&recovery_steps)?;
248
249 // Generate fallback plans
250 let fallback_plans = self.generate_fallback_plans(
251 missing_chunks,
252 available_chunks,
253 &recovery_requirements,
254 )?;
255
256 Ok(RecoveryPlan {
257 plan_id: format!("recovery_plan_{}", Utc::now().timestamp()),
258 target_chunks: missing_chunks.to_vec(),
259 recovery_steps,
260 estimated_time_seconds: estimated_time,
261 estimated_bandwidth_usage_mb: estimated_bandwidth,
262 estimated_cost,
263 optimization_strategy: strategy,
264 fallback_plans,
265 })
266 }
267
268 /// Analyze available chunk sources for optimization
269 fn analyze_chunk_sources(
270 &self,
271 available_chunks: &HashMap<String, Vec<NodeLocation>>,
272 ) -> Result<SourceAnalysis> {
273 let mut analysis = SourceAnalysis {
274 total_sources: 0,
275 sources_by_region: HashMap::new(),
276 bandwidth_distribution: BandwidthDistribution::default(),
277 load_distribution: LoadDistribution::default(),
278 };
279
280 for (chunk_id, locations) in available_chunks {
281 analysis.total_sources += locations.len();
282
283 for location in locations {
284 // Analyze by region
285 *analysis.sources_by_region.entry(location.region.clone()).or_insert(0) += 1;
286
287 // Analyze bandwidth availability
288 if let Some(node_info) = self.network_topology.nodes.get(&location.node_id) {
289 analysis.bandwidth_distribution.update(node_info.available_bandwidth_mbps);
290 analysis.load_distribution.update(node_info.load_factor);
291 }
292 }
293 }
294
295 Ok(analysis)
296 }
297
298 /// Select optimal recovery strategy
299 fn select_optimization_strategy(
300 &self,
301 requirements: &RecoveryRequirements,
302 analysis: &SourceAnalysis,
303 ) -> Result<OptimizationStrategy> {
304
305 // Priority-based selection
306 if requirements.time_critical {
307 if analysis.bandwidth_distribution.high_bandwidth_sources > 3 {
308 return Ok(OptimizationStrategy::ParallelRecovery);
309 } else {
310 return Ok(OptimizationStrategy::LatencyOptimized);
311 }
312 }
313
314 if requirements.cost_sensitive {
315 return Ok(OptimizationStrategy::CostOptimized);
316 }
317
318 if analysis.sources_by_region.len() > 1 {
319 return Ok(OptimizationStrategy::LocalityOptimized);
320 }
321
322 // Default to adaptive bandwidth
323 Ok(OptimizationStrategy::AdaptiveBandwidth)
324 }
325
326 /// Generate optimized recovery steps
327 fn generate_recovery_steps(
328 &self,
329 missing_chunks: &[String],
330 available_chunks: &HashMap<String, Vec<NodeLocation>>,
331 strategy: &OptimizationStrategy,
332 requirements: &RecoveryRequirements,
333 ) -> Result<Vec<RecoveryStep>> {
334 match strategy {
335 OptimizationStrategy::ParallelRecovery => {
336 self.generate_parallel_recovery_steps(missing_chunks, available_chunks, requirements)
337 },
338 OptimizationStrategy::ProgressiveRecovery => {
339 self.generate_progressive_recovery_steps(missing_chunks, available_chunks, requirements)
340 },
341 OptimizationStrategy::LocalityOptimized => {
342 self.generate_locality_optimized_steps(missing_chunks, available_chunks, requirements)
343 },
344 OptimizationStrategy::LoadBalanced => {
345 self.generate_load_balanced_steps(missing_chunks, available_chunks, requirements)
346 },
347 OptimizationStrategy::AdaptiveBandwidth => {
348 self.generate_adaptive_bandwidth_steps(missing_chunks, available_chunks, requirements)
349 },
350 _ => {
351 // Default implementation
352 self.generate_basic_recovery_steps(missing_chunks, available_chunks, requirements)
353 }
354 }
355 }
356
357 /// Generate parallel recovery steps
358 fn generate_parallel_recovery_steps(
359 &self,
360 missing_chunks: &[String],
361 available_chunks: &HashMap<String, Vec<NodeLocation>>,
362 _requirements: &RecoveryRequirements,
363 ) -> Result<Vec<RecoveryStep>> {
364 let mut steps = Vec::new();
365 let max_parallel = self.algorithms.max_parallel_streams as usize;
366
367 // Group chunks into parallel batches
368 for (batch_idx, chunk_batch) in missing_chunks.chunks(max_parallel).enumerate() {
369 let mut source_nodes = Vec::new();
370 let mut chunk_list = Vec::new();
371
372 for chunk_id in chunk_batch {
373 if let Some(locations) = available_chunks.get(chunk_id) {
374 // Select best source for this chunk
375 let best_source = self.select_best_source(locations)?;
376 source_nodes.push(best_source.node_id.clone());
377 chunk_list.push(chunk_id.clone());
378 }
379 }
380
381 if !chunk_list.is_empty() {
382 steps.push(RecoveryStep {
383 step_id: format!("parallel_batch_{}", batch_idx),
384 step_type: RecoveryStepType::ParallelTransfer,
385 source_nodes,
386 target_chunks: chunk_list,
387 estimated_duration_seconds: 30.0, // Estimate based on parallel efficiency
388 bandwidth_requirement_mbps: 100.0 * chunk_batch.len() as f64,
389 priority: RecoveryPriority::High,
390 dependencies: Vec::new(),
391 });
392 }
393 }
394
395 // Add verification step
396 steps.push(RecoveryStep {
397 step_id: "verify_parallel_recovery".to_string(),
398 step_type: RecoveryStepType::VerifyIntegrity,
399 source_nodes: Vec::new(),
400 target_chunks: missing_chunks.to_vec(),
401 estimated_duration_seconds: 5.0,
402 bandwidth_requirement_mbps: 0.0,
403 priority: RecoveryPriority::Critical,
404 dependencies: steps.iter().map(|s| s.step_id.clone()).collect(),
405 });
406
407 Ok(steps)
408 }
409
410 /// Generate locality-optimized recovery steps
411 fn generate_locality_optimized_steps(
412 &self,
413 missing_chunks: &[String],
414 available_chunks: &HashMap<String, Vec<NodeLocation>>,
415 requirements: &RecoveryRequirements,
416 ) -> Result<Vec<RecoveryStep>> {
417 // Group chunks by optimal source region
418 let mut chunks_by_region: HashMap<GeographicRegion, Vec<String>> = HashMap::new();
419
420 for chunk_id in missing_chunks {
421 if let Some(locations) = available_chunks.get(chunk_id) {
422 let optimal_region = self.find_optimal_source_region(locations, requirements)?;
423 chunks_by_region.entry(optimal_region)
424 .or_insert_with(Vec::new)
425 .push(chunk_id.clone());
426 }
427 }
428
429 // Create steps for each region
430 let mut steps = Vec::new();
431 for (region, chunks) in chunks_by_region {
432 let region_nodes = self.get_region_nodes(&region);
433
434 steps.push(RecoveryStep {
435 step_id: format!("locality_{:?}", region),
436 step_type: RecoveryStepType::DirectTransfer,
437 source_nodes: region_nodes.into_iter().take(3).collect(), // Top 3 nodes
438 target_chunks: chunks,
439 estimated_duration_seconds: 45.0,
440 bandwidth_requirement_mbps: 50.0,
441 priority: RecoveryPriority::Normal,
442 dependencies: Vec::new(),
443 });
444 }
445
446 Ok(steps)
447 }
448
449 /// Generate adaptive bandwidth recovery steps
450 fn generate_adaptive_bandwidth_steps(
451 &self,
452 missing_chunks: &[String],
453 available_chunks: &HashMap<String, Vec<NodeLocation>>,
454 _requirements: &RecoveryRequirements,
455 ) -> Result<Vec<RecoveryStep>> {
456 let mut steps = Vec::new();
457
458 // Analyze current network conditions
459 let network_capacity = self.calculate_available_network_capacity()?;
460
461 // Adaptive chunking based on available bandwidth
462 let optimal_batch_size = self.calculate_optimal_batch_size(network_capacity);
463
464 for (batch_idx, chunk_batch) in missing_chunks.chunks(optimal_batch_size).enumerate() {
465 let bandwidth_per_chunk = network_capacity / chunk_batch.len() as f64;
466
467 let mut batch_sources = Vec::new();
468 for chunk_id in chunk_batch {
469 if let Some(locations) = available_chunks.get(chunk_id) {
470 let source = self.select_bandwidth_optimal_source(locations, bandwidth_per_chunk)?;
471 batch_sources.push(source.node_id.clone());
472 }
473 }
474
475 steps.push(RecoveryStep {
476 step_id: format!("adaptive_batch_{}", batch_idx),
477 step_type: RecoveryStepType::ParallelTransfer,
478 source_nodes: batch_sources,
479 target_chunks: chunk_batch.to_vec(),
480 estimated_duration_seconds: 60.0 / (network_capacity / 100.0), // Scale with capacity
481 bandwidth_requirement_mbps: network_capacity * 0.8, // Use 80% of capacity
482 priority: RecoveryPriority::Normal,
483 dependencies: Vec::new(),
484 });
485 }
486
487 Ok(steps)
488 }
489
490 /// Generate basic recovery steps (fallback)
491 fn generate_basic_recovery_steps(
492 &self,
493 missing_chunks: &[String],
494 available_chunks: &HashMap<String, Vec<NodeLocation>>,
495 _requirements: &RecoveryRequirements,
496 ) -> Result<Vec<RecoveryStep>> {
497 let mut steps = Vec::new();
498
499 for (idx, chunk_id) in missing_chunks.iter().enumerate() {
500 if let Some(locations) = available_chunks.get(chunk_id) {
501 let source = self.select_best_source(locations)?;
502
503 steps.push(RecoveryStep {
504 step_id: format!("basic_recovery_{}", idx),
505 step_type: RecoveryStepType::DirectTransfer,
506 source_nodes: vec![source.node_id.clone()],
507 target_chunks: vec![chunk_id.clone()],
508 estimated_duration_seconds: 30.0,
509 bandwidth_requirement_mbps: 25.0,
510 priority: RecoveryPriority::Normal,
511 dependencies: Vec::new(),
512 });
513 }
514 }
515
516 Ok(steps)
517 }
518
519 /// Select best source node from available locations
520 fn select_best_source<'a>(&self, locations: &'a [NodeLocation]) -> Result<&'a NodeLocation> {
521 let mut best_location = &locations[0];
522 let mut best_score = 0.0;
523
524 for location in locations {
525 let score = self.calculate_source_score(location)?;
526 if score > best_score {
527 best_score = score;
528 best_location = location;
529 }
530 }
531
532 Ok(best_location)
533 }
534
535 /// Calculate source quality score
536 fn calculate_source_score(&self, location: &NodeLocation) -> Result<f64> {
537 let node_info = self.network_topology.nodes.get(&location.node_id)
538 .ok_or_else(|| anyhow::anyhow!("Node info not found"))?;
539
540 let bandwidth_score = (node_info.available_bandwidth_mbps / 100.0).min(1.0);
541 let load_score = 1.0 - node_info.load_factor;
542 let quality_score = match node_info.connection_quality {
543 ConnectionQuality::Excellent => 1.0,
544 ConnectionQuality::Good => 0.8,
545 ConnectionQuality::Average => 0.6,
546 ConnectionQuality::Poor => 0.3,
547 };
548
549 Ok(bandwidth_score * 0.4 + load_score * 0.3 + quality_score * 0.3)
550 }
551
552 /// Calculate available network capacity
553 fn calculate_available_network_capacity(&self) -> Result<f64> {
554 let total_capacity: f64 = self.network_topology.nodes
555 .values()
556 .map(|node| node.available_bandwidth_mbps)
557 .sum();
558
559 let avg_utilization: f64 = self.network_topology.nodes
560 .values()
561 .map(|node| node.load_factor)
562 .sum::<f64>() / self.network_topology.nodes.len() as f64;
563
564 Ok(total_capacity * (1.0 - avg_utilization))
565 }
566
567 /// Calculate optimal batch size for current network conditions
568 fn calculate_optimal_batch_size(&self, available_bandwidth: f64) -> usize {
569 // Adaptive batch sizing based on bandwidth
570 if available_bandwidth > 500.0 {
571 8 // High bandwidth - large batches
572 } else if available_bandwidth > 200.0 {
573 4 // Medium bandwidth
574 } else if available_bandwidth > 50.0 {
575 2 // Low bandwidth
576 } else {
577 1 // Very low bandwidth - sequential
578 }
579 }
580
581 /// Select bandwidth-optimal source
582 fn select_bandwidth_optimal_source<'a>(
583 &self,
584 locations: &'a [NodeLocation],
585 required_bandwidth: f64,
586 ) -> Result<&'a NodeLocation> {
587 let mut best_location = &locations[0];
588 let mut best_bandwidth = 0.0;
589
590 for location in locations {
591 if let Some(node_info) = self.network_topology.nodes.get(&location.node_id) {
592 if node_info.available_bandwidth_mbps >= required_bandwidth &&
593 node_info.available_bandwidth_mbps > best_bandwidth {
594 best_bandwidth = node_info.available_bandwidth_mbps;
595 best_location = location;
596 }
597 }
598 }
599
600 Ok(best_location)
601 }
602
603 /// Find optimal source region for chunk recovery
604 fn find_optimal_source_region(
605 &self,
606 locations: &[NodeLocation],
607 requirements: &RecoveryRequirements,
608 ) -> Result<GeographicRegion> {
609 // If client region is specified, prefer that
610 if let Some(client_region) = &requirements.client_region {
611 if locations.iter().any(|loc| loc.region == *client_region) {
612 return Ok(client_region.clone());
613 }
614 }
615
616 // Otherwise, select region with best connectivity
617 let mut region_scores: HashMap<GeographicRegion, f64> = HashMap::new();
618
619 for location in locations {
620 let score = region_scores.entry(location.region.clone()).or_insert(0.0);
621 *score += self.calculate_source_score(location)?;
622 }
623
624 let best_region = region_scores
625 .into_iter()
626 .max_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal))
627 .map(|(region, _)| region)
628 .unwrap_or(GeographicRegion::NorthAmerica);
629
630 Ok(best_region)
631 }
632
633 /// Get nodes in a specific region
634 fn get_region_nodes(&self, region: &GeographicRegion) -> Vec<String> {
635 self.network_topology.nodes
636 .values()
637 .filter(|node| node.region == *region)
638 .map(|node| node.node_id.clone())
639 .collect()
640 }
641
642 /// Generate fallback plans
643 fn generate_fallback_plans(
644 &self,
645 missing_chunks: &[String],
646 available_chunks: &HashMap<String, Vec<NodeLocation>>,
647 requirements: &RecoveryRequirements,
648 ) -> Result<Vec<FallbackPlan>> {
649 let mut fallback_plans = Vec::new();
650
651 // Fallback 1: Sequential recovery if parallel fails
652 fallback_plans.push(FallbackPlan {
653 fallback_id: "sequential_fallback".to_string(),
654 trigger_conditions: vec!["parallel_transfer_failed".to_string()],
655 alternative_steps: self.generate_basic_recovery_steps(
656 missing_chunks,
657 available_chunks,
658 requirements,
659 )?,
660 performance_impact: 0.5, // 50% slower
661 });
662
663 // Fallback 2: High-latency sources if primary sources fail
664 fallback_plans.push(FallbackPlan {
665 fallback_id: "high_latency_fallback".to_string(),
666 trigger_conditions: vec!["primary_sources_unavailable".to_string()],
667 alternative_steps: self.generate_high_latency_recovery_steps(
668 missing_chunks,
669 available_chunks,
670 )?,
671 performance_impact: 1.5, // 150% slower
672 });
673
674 Ok(fallback_plans)
675 }
676
677 /// Generate high-latency recovery steps
678 fn generate_high_latency_recovery_steps(
679 &self,
680 missing_chunks: &[String],
681 available_chunks: &HashMap<String, Vec<NodeLocation>>,
682 ) -> Result<Vec<RecoveryStep>> {
683 // Implementation would select slower but more reliable sources
684 self.generate_basic_recovery_steps(missing_chunks, available_chunks, &RecoveryRequirements::default())
685 }
686
687 /// Calculate recovery estimates
688 fn calculate_recovery_estimates(
689 &self,
690 steps: &[RecoveryStep],
691 ) -> Result<(f64, f64, f64)> {
692 let total_time = steps.iter()
693 .map(|step| step.estimated_duration_seconds)
694 .sum::<f64>();
695
696 let total_bandwidth = steps.iter()
697 .map(|step| step.bandwidth_requirement_mbps * step.estimated_duration_seconds / 8.0) // Convert to MB
698 .sum::<f64>();
699
700 let total_cost = total_bandwidth * 0.02; // Estimate $0.02 per GB
701
702 Ok((total_time, total_bandwidth, total_cost))
703 }
704
705 /// Execute recovery plan
706 pub async fn execute_recovery_plan(&mut self, plan: &RecoveryPlan) -> Result<RecoveryExecutionResult> {
707 let start_time = crate::SerializableInstant::now();
708 let mut executed_steps = Vec::new();
709 let mut total_bytes_recovered = 0u64;
710
711 // Execute steps according to dependencies
712 for step in &plan.recovery_steps {
713 let step_result = self.execute_recovery_step(step).await?;
714 total_bytes_recovered += step_result.bytes_transferred;
715 executed_steps.push(step_result);
716 }
717
718 let execution_time = start_time.elapsed().as_secs_f64();
719
720 // Update metrics
721 self.performance_metrics.total_recoveries += 1;
722 self.performance_metrics.successful_recoveries += 1;
723 self.performance_metrics.total_bytes_recovered += total_bytes_recovered;
724
725 // Update average recovery time
726 let total_successful = self.performance_metrics.successful_recoveries as f64;
727 self.performance_metrics.average_recovery_time_seconds =
728 (self.performance_metrics.average_recovery_time_seconds * (total_successful - 1.0) + execution_time) / total_successful;
729
730 Ok(RecoveryExecutionResult {
731 plan_id: plan.plan_id.clone(),
732 success: true,
733 execution_time_seconds: execution_time,
734 bytes_recovered: total_bytes_recovered,
735 bandwidth_efficiency: self.calculate_bandwidth_efficiency(plan, execution_time),
736 executed_steps,
737 error_message: None,
738 })
739 }
740
741 /// Execute single recovery step
742 async fn execute_recovery_step(&self, step: &RecoveryStep) -> Result<StepExecutionResult> {
743 // Simulate step execution
744 let chunk_size = 1024 * 1024; // 1MB per chunk
745 let bytes_per_chunk = chunk_size;
746 let total_bytes = step.target_chunks.len() as u64 * bytes_per_chunk;
747
748 // Simulate transfer time
749 tokio::time::sleep(tokio::time::Duration::from_millis(
750 (step.estimated_duration_seconds * 100.0) as u64
751 )).await;
752
753 Ok(StepExecutionResult {
754 step_id: step.step_id.clone(),
755 success: true,
756 bytes_transferred: total_bytes,
757 actual_duration_seconds: step.estimated_duration_seconds,
758 bandwidth_used_mbps: step.bandwidth_requirement_mbps,
759 error_message: None,
760 })
761 }
762
763 /// Calculate bandwidth efficiency
764 fn calculate_bandwidth_efficiency(&self, plan: &RecoveryPlan, actual_time: f64) -> f64 {
765 let theoretical_optimal = plan.estimated_bandwidth_usage_mb / plan.estimated_time_seconds;
766 let actual_efficiency = plan.estimated_bandwidth_usage_mb / actual_time;
767
768 (actual_efficiency / theoretical_optimal).min(1.0)
769 }
770 }
771
772 // Supporting types and implementations
773
774 #[derive(Debug, Clone, Serialize, Deserialize)]
775 pub struct NodeLocation {
776 pub node_id: String,
777 pub region: GeographicRegion,
778 pub availability_score: f64,
779 }
780
781 #[derive(Debug, Clone, Serialize, Deserialize)]
782 pub struct RecoveryRequirements {
783 pub time_critical: bool,
784 pub cost_sensitive: bool,
785 pub client_region: Option<GeographicRegion>,
786 pub max_bandwidth_mbps: Option<f64>,
787 pub preferred_quality: ConnectionQuality,
788 }
789
790 impl Default for RecoveryRequirements {
791 fn default() -> Self {
792 Self {
793 time_critical: false,
794 cost_sensitive: false,
795 client_region: None,
796 max_bandwidth_mbps: None,
797 preferred_quality: ConnectionQuality::Good,
798 }
799 }
800 }
801
802 #[derive(Debug, Default)]
803 struct SourceAnalysis {
804 pub total_sources: usize,
805 pub sources_by_region: HashMap<GeographicRegion, usize>,
806 pub bandwidth_distribution: BandwidthDistribution,
807 pub load_distribution: LoadDistribution,
808 }
809
810 #[derive(Debug, Default)]
811 struct BandwidthDistribution {
812 pub high_bandwidth_sources: usize, // >100 Mbps
813 pub medium_bandwidth_sources: usize, // 50-100 Mbps
814 pub low_bandwidth_sources: usize, // <50 Mbps
815 }
816
817 impl BandwidthDistribution {
818 fn update(&mut self, bandwidth: f64) {
819 if bandwidth > 100.0 {
820 self.high_bandwidth_sources += 1;
821 } else if bandwidth > 50.0 {
822 self.medium_bandwidth_sources += 1;
823 } else {
824 self.low_bandwidth_sources += 1;
825 }
826 }
827 }
828
829 #[derive(Debug, Default)]
830 struct LoadDistribution {
831 pub low_load_sources: usize, // <30% load
832 pub medium_load_sources: usize, // 30-70% load
833 pub high_load_sources: usize, // >70% load
834 }
835
836 impl LoadDistribution {
837 fn update(&mut self, load: f64) {
838 if load < 0.3 {
839 self.low_load_sources += 1;
840 } else if load < 0.7 {
841 self.medium_load_sources += 1;
842 } else {
843 self.high_load_sources += 1;
844 }
845 }
846 }
847
848 #[derive(Debug, Clone, Serialize, Deserialize)]
849 pub struct RecoveryExecutionResult {
850 pub plan_id: String,
851 pub success: bool,
852 pub execution_time_seconds: f64,
853 pub bytes_recovered: u64,
854 pub bandwidth_efficiency: f64,
855 pub executed_steps: Vec<StepExecutionResult>,
856 pub error_message: Option<String>,
857 }
858
859 #[derive(Debug, Clone, Serialize, Deserialize)]
860 pub struct StepExecutionResult {
861 pub step_id: String,
862 pub success: bool,
863 pub bytes_transferred: u64,
864 pub actual_duration_seconds: f64,
865 pub bandwidth_used_mbps: f64,
866 pub error_message: Option<String>,
867 }
868
869 // Placeholder implementations for missing methods
870 impl RecoveryOptimizer {
871 fn generate_progressive_recovery_steps(
872 &self,
873 missing_chunks: &[String],
874 available_chunks: &HashMap<String, Vec<NodeLocation>>,
875 requirements: &RecoveryRequirements,
876 ) -> Result<Vec<RecoveryStep>> {
877 // Implementation would prioritize chunks by importance
878 self.generate_basic_recovery_steps(missing_chunks, available_chunks, requirements)
879 }
880
881 fn generate_load_balanced_steps(
882 &self,
883 missing_chunks: &[String],
884 available_chunks: &HashMap<String, Vec<NodeLocation>>,
885 requirements: &RecoveryRequirements,
886 ) -> Result<Vec<RecoveryStep>> {
887 // Implementation would distribute load evenly across nodes
888 self.generate_basic_recovery_steps(missing_chunks, available_chunks, requirements)
889 }
890 }
891
892 #[cfg(test)]
893 mod tests {
894 use super::*;
895
896 #[test]
897 fn test_recovery_optimizer_creation() {
898 let optimizer = RecoveryOptimizer::new();
899 assert!(!optimizer.optimization_strategies.is_empty());
900 assert_eq!(optimizer.algorithms.max_parallel_streams, 8);
901 }
902
903 #[test]
904 fn test_bandwidth_distribution() {
905 let mut dist = BandwidthDistribution::default();
906
907 dist.update(150.0); // High
908 dist.update(75.0); // Medium
909 dist.update(25.0); // Low
910
911 assert_eq!(dist.high_bandwidth_sources, 1);
912 assert_eq!(dist.medium_bandwidth_sources, 1);
913 assert_eq!(dist.low_bandwidth_sources, 1);
914 }
915
916 #[test]
917 fn test_optimal_batch_size_calculation() {
918 let optimizer = RecoveryOptimizer::new();
919
920 assert_eq!(optimizer.calculate_optimal_batch_size(600.0), 8);
921 assert_eq!(optimizer.calculate_optimal_batch_size(300.0), 4);
922 assert_eq!(optimizer.calculate_optimal_batch_size(100.0), 2);
923 assert_eq!(optimizer.calculate_optimal_batch_size(30.0), 1);
924 }
925
926 #[tokio::test]
927 async fn test_recovery_plan_creation() {
928 let optimizer = RecoveryOptimizer::new();
929 let missing_chunks = vec!["chunk1".to_string(), "chunk2".to_string()];
930 let mut available_chunks = HashMap::new();
931
932 available_chunks.insert("chunk1".to_string(), vec![
933 NodeLocation {
934 node_id: "node1".to_string(),
935 region: GeographicRegion::NorthAmerica,
936 availability_score: 0.9,
937 }
938 ]);
939
940 let requirements = RecoveryRequirements::default();
941
942 let plan = optimizer.create_recovery_plan(
943 &missing_chunks,
944 &available_chunks,
945 requirements,
946 ).unwrap();
947
948 assert!(!plan.plan_id.is_empty());
949 assert!(!plan.recovery_steps.is_empty());
950 assert!(plan.estimated_time_seconds > 0.0);
951 }
952 }