ipfrs-storage 0.1.0

Storage backends and block management for IPFRS content-addressed system
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
//! Storage diagnostics and health monitoring utilities
//!
//! Provides comprehensive tools for analyzing storage performance,
//! health, and identifying potential issues.

use crate::traits::BlockStore;
use ipfrs_core::{Block, Cid, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::time::{Duration, Instant};
use sysinfo::{ProcessesToUpdate, System};

/// Comprehensive storage diagnostics report
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DiagnosticsReport {
    /// Storage backend name
    pub backend: String,
    /// Total blocks tested
    pub total_blocks: usize,
    /// Performance metrics
    pub performance: PerformanceMetrics,
    /// Health check results
    pub health: HealthMetrics,
    /// Recommendations for optimization
    pub recommendations: Vec<String>,
    /// Overall health score (0-100)
    pub health_score: u8,
}

/// Performance metrics for storage operations
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PerformanceMetrics {
    /// Average write latency
    pub avg_write_latency: Duration,
    /// Average read latency
    pub avg_read_latency: Duration,
    /// Average batch write latency
    pub avg_batch_write_latency: Duration,
    /// Average batch read latency
    pub avg_batch_read_latency: Duration,
    /// Write throughput (blocks/sec)
    pub write_throughput: f64,
    /// Read throughput (blocks/sec)
    pub read_throughput: f64,
    /// Peak memory usage (bytes)
    pub peak_memory_usage: usize,
}

/// Health metrics for storage backend
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthMetrics {
    /// Number of successful operations
    pub successful_ops: usize,
    /// Number of failed operations
    pub failed_ops: usize,
    /// Success rate (0.0 - 1.0)
    pub success_rate: f64,
    /// Data integrity check passed
    pub integrity_ok: bool,
    /// Storage is responsive
    pub responsive: bool,
}

/// Memory usage tracker for diagnostics
struct MemoryTracker {
    system: System,
    pid: sysinfo::Pid,
    peak_memory: usize,
}

impl MemoryTracker {
    /// Create a new memory tracker
    fn new() -> Self {
        let mut system = System::new();
        system.refresh_processes(ProcessesToUpdate::All, true);
        let pid = sysinfo::get_current_pid().unwrap();

        Self {
            system,
            pid,
            peak_memory: 0,
        }
    }

    /// Update peak memory usage
    fn update(&mut self) {
        self.system.refresh_processes(ProcessesToUpdate::All, true);
        if let Some(process) = self.system.process(self.pid) {
            let current_memory = process.memory() as usize;
            if current_memory > self.peak_memory {
                self.peak_memory = current_memory;
            }
        }
    }

    /// Get peak memory usage in bytes
    fn peak_memory_bytes(&self) -> usize {
        self.peak_memory
    }
}

/// Storage diagnostics runner
pub struct StorageDiagnostics<S: BlockStore> {
    store: S,
    backend_name: String,
}

impl<S: BlockStore> StorageDiagnostics<S> {
    /// Create a new diagnostics runner
    pub fn new(store: S, backend_name: String) -> Self {
        Self {
            store,
            backend_name,
        }
    }

    /// Run comprehensive diagnostics
    ///
    /// Tests include:
    /// - Write/read latency measurements
    /// - Batch operation performance
    /// - Data integrity verification
    /// - Storage responsiveness
    /// - Memory usage tracking
    pub async fn run(&mut self) -> Result<DiagnosticsReport> {
        let mut successful_ops = 0;
        let mut failed_ops = 0;

        // Initialize memory tracker
        let mut memory_tracker = MemoryTracker::new();
        memory_tracker.update();

        // Test data
        let test_blocks = self.generate_test_data()?;
        memory_tracker.update();

        // Measure write performance
        let write_start = Instant::now();
        for block in &test_blocks {
            match self.store.put(block).await {
                Ok(_) => successful_ops += 1,
                Err(_) => failed_ops += 1,
            }
        }
        let write_duration = write_start.elapsed();
        let avg_write_latency = write_duration / test_blocks.len() as u32;
        memory_tracker.update();

        // Measure read performance
        let read_start = Instant::now();
        let mut integrity_ok = true;
        for block in &test_blocks {
            match self.store.get(block.cid()).await {
                Ok(Some(retrieved)) => {
                    if retrieved.data() != block.data() {
                        integrity_ok = false;
                    }
                    successful_ops += 1;
                }
                Ok(None) => {
                    integrity_ok = false;
                    failed_ops += 1;
                }
                Err(_) => failed_ops += 1,
            }
        }
        let read_duration = read_start.elapsed();
        let avg_read_latency = read_duration / test_blocks.len() as u32;
        memory_tracker.update();

        // Measure batch write performance
        let batch_write_start = Instant::now();
        let batch_result = self.store.put_many(&test_blocks).await;
        let avg_batch_write_latency = batch_write_start.elapsed();
        if batch_result.is_ok() {
            successful_ops += test_blocks.len();
        } else {
            failed_ops += test_blocks.len();
        }
        memory_tracker.update();

        // Measure batch read performance
        let cids: Vec<Cid> = test_blocks.iter().map(|b| *b.cid()).collect();
        let batch_read_start = Instant::now();
        let _batch_read_result = self.store.get_many(&cids).await;
        let avg_batch_read_latency = batch_read_start.elapsed();
        memory_tracker.update();

        // Calculate throughput
        let write_throughput = test_blocks.len() as f64 / write_duration.as_secs_f64();
        let read_throughput = test_blocks.len() as f64 / read_duration.as_secs_f64();

        // Calculate success rate
        let total_ops = successful_ops + failed_ops;
        let success_rate = if total_ops > 0 {
            successful_ops as f64 / total_ops as f64
        } else {
            0.0
        };

        // Check responsiveness
        let responsive = avg_write_latency < Duration::from_secs(1)
            && avg_read_latency < Duration::from_millis(500);

        // Generate recommendations
        let recommendations = self.generate_recommendations(
            &avg_write_latency,
            &avg_read_latency,
            write_throughput,
            read_throughput,
            integrity_ok,
            responsive,
        );

        // Calculate health score
        let health_score = self.calculate_health_score(
            success_rate,
            integrity_ok,
            responsive,
            write_throughput,
            read_throughput,
        );

        // Get peak memory usage
        let peak_memory_usage = memory_tracker.peak_memory_bytes();

        Ok(DiagnosticsReport {
            backend: self.backend_name.clone(),
            total_blocks: test_blocks.len(),
            performance: PerformanceMetrics {
                avg_write_latency,
                avg_read_latency,
                avg_batch_write_latency,
                avg_batch_read_latency,
                write_throughput,
                read_throughput,
                peak_memory_usage,
            },
            health: HealthMetrics {
                successful_ops,
                failed_ops,
                success_rate,
                integrity_ok,
                responsive,
            },
            recommendations,
            health_score,
        })
    }

    /// Run quick health check (minimal overhead)
    pub async fn quick_health_check(&mut self) -> Result<bool> {
        // Test with a single small block
        let test_data = vec![0u8; 1024];
        let cid = crate::utils::compute_cid(&test_data);
        let block = Block::from_parts(cid, test_data.into());

        // Try write
        self.store.put(&block).await?;

        // Try read
        let retrieved = self.store.get(&cid).await?;

        // Verify
        Ok(retrieved.is_some() && retrieved.unwrap().cid() == &cid)
    }

    /// Generate test data for diagnostics
    fn generate_test_data(&self) -> Result<Vec<Block>> {
        crate::utils::generate_mixed_size_blocks(5, 3, 2)
    }

    /// Generate recommendations based on metrics
    #[allow(clippy::too_many_arguments)]
    fn generate_recommendations(
        &self,
        avg_write_latency: &Duration,
        avg_read_latency: &Duration,
        write_throughput: f64,
        read_throughput: f64,
        integrity_ok: bool,
        responsive: bool,
    ) -> Vec<String> {
        let mut recommendations = Vec::new();

        if *avg_write_latency > Duration::from_millis(100) {
            recommendations.push(
                "High write latency detected. Consider enabling write coalescing or batch operations.".to_string()
            );
        }

        if *avg_read_latency > Duration::from_millis(50) {
            recommendations.push(
                "High read latency detected. Consider enabling caching or bloom filters."
                    .to_string(),
            );
        }

        if write_throughput < 100.0 {
            recommendations.push(
                "Low write throughput. Consider using ParityDB backend or enabling compression."
                    .to_string(),
            );
        }

        if read_throughput < 200.0 {
            recommendations.push(
                "Low read throughput. Consider increasing cache size or using tiered caching."
                    .to_string(),
            );
        }

        if !integrity_ok {
            recommendations.push(
                "Data integrity issues detected! This is critical and should be investigated immediately.".to_string()
            );
        }

        if !responsive {
            recommendations.push(
                "Storage backend is not responsive. Check system resources and backend configuration.".to_string()
            );
        }

        if recommendations.is_empty() {
            recommendations.push("Storage is performing well. No issues detected.".to_string());
        }

        recommendations
    }

    /// Calculate overall health score (0-100)
    fn calculate_health_score(
        &self,
        success_rate: f64,
        integrity_ok: bool,
        responsive: bool,
        write_throughput: f64,
        read_throughput: f64,
    ) -> u8 {
        let mut score = 0u32;

        // Success rate (40 points)
        score += (success_rate * 40.0) as u32;

        // Integrity (30 points)
        if integrity_ok {
            score += 30;
        }

        // Responsiveness (15 points)
        if responsive {
            score += 15;
        }

        // Write throughput (7.5 points)
        if write_throughput >= 100.0 {
            score += 7;
        } else {
            score += (write_throughput / 100.0 * 7.0) as u32;
        }

        // Read throughput (7.5 points)
        if read_throughput >= 200.0 {
            score += 8;
        } else {
            score += (read_throughput / 200.0 * 8.0) as u32;
        }

        score.min(100) as u8
    }
}

/// Benchmark comparison between different storage backends
pub struct BenchmarkComparison {
    results: HashMap<String, DiagnosticsReport>,
}

impl BenchmarkComparison {
    /// Create a new benchmark comparison
    pub fn new() -> Self {
        Self {
            results: HashMap::new(),
        }
    }

    /// Add a benchmark result
    pub fn add_result(&mut self, name: String, report: DiagnosticsReport) {
        self.results.insert(name, report);
    }

    /// Get the fastest backend for writes
    pub fn fastest_write_backend(&self) -> Option<(&str, &DiagnosticsReport)> {
        self.results
            .iter()
            .min_by_key(|(_, r)| r.performance.avg_write_latency)
            .map(|(name, report)| (name.as_str(), report))
    }

    /// Get the fastest backend for reads
    pub fn fastest_read_backend(&self) -> Option<(&str, &DiagnosticsReport)> {
        self.results
            .iter()
            .min_by_key(|(_, r)| r.performance.avg_read_latency)
            .map(|(name, report)| (name.as_str(), report))
    }

    /// Get the healthiest backend
    pub fn healthiest_backend(&self) -> Option<(&str, &DiagnosticsReport)> {
        self.results
            .iter()
            .max_by_key(|(_, r)| r.health_score)
            .map(|(name, report)| (name.as_str(), report))
    }

    /// Generate a comparison summary
    pub fn summary(&self) -> String {
        let mut summary = String::from("=== Storage Backend Comparison ===\n\n");

        for (name, report) in &self.results {
            summary.push_str(&format!(
                "{}: Health Score = {}/100\n",
                name, report.health_score
            ));
            summary.push_str(&format!(
                "  Write Latency: {:?}, Read Latency: {:?}\n",
                report.performance.avg_write_latency, report.performance.avg_read_latency
            ));
            summary.push_str(&format!(
                "  Write Throughput: {:.2} blocks/s, Read Throughput: {:.2} blocks/s\n\n",
                report.performance.write_throughput, report.performance.read_throughput
            ));
        }

        if let Some((name, _)) = self.fastest_write_backend() {
            summary.push_str(&format!("Fastest for writes: {name}\n"));
        }

        if let Some((name, _)) = self.fastest_read_backend() {
            summary.push_str(&format!("Fastest for reads: {name}\n"));
        }

        if let Some((name, _)) = self.healthiest_backend() {
            summary.push_str(&format!("Healthiest overall: {name}\n"));
        }

        summary
    }
}

impl Default for BenchmarkComparison {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::MemoryBlockStore;

    #[tokio::test]
    async fn test_diagnostics_run() {
        let store = MemoryBlockStore::new();
        let mut diagnostics = StorageDiagnostics::new(store, "MemoryStore".to_string());

        let report = diagnostics.run().await.unwrap();
        assert_eq!(report.backend, "MemoryStore");
        assert!(report.health_score > 0);
        assert!(report.health.integrity_ok);
    }

    #[tokio::test]
    async fn test_quick_health_check() {
        let store = MemoryBlockStore::new();
        let mut diagnostics = StorageDiagnostics::new(store, "MemoryStore".to_string());

        let healthy = diagnostics.quick_health_check().await.unwrap();
        assert!(healthy);
    }

    #[tokio::test]
    async fn test_benchmark_comparison() {
        let mut comparison = BenchmarkComparison::new();

        let store1 = MemoryBlockStore::new();
        let mut diag1 = StorageDiagnostics::new(store1, "Memory1".to_string());
        let report1 = diag1.run().await.unwrap();
        comparison.add_result("Memory1".to_string(), report1);

        let store2 = MemoryBlockStore::new();
        let mut diag2 = StorageDiagnostics::new(store2, "Memory2".to_string());
        let report2 = diag2.run().await.unwrap();
        comparison.add_result("Memory2".to_string(), report2);

        assert!(comparison.fastest_write_backend().is_some());
        assert!(comparison.fastest_read_backend().is_some());
        assert!(comparison.healthiest_backend().is_some());

        let summary = comparison.summary();
        assert!(summary.contains("Storage Backend Comparison"));
    }
}