Alexander
0e5a514015
Week 5 (CDC & Delta Detection): - Add read_full() method to avoid u32 overflow on >4GB files - Add chunk_streaming() to avoid 200MB+ memory per file - Implement scan_origin() recursive walk (was stub) - Use spawn_blocking for watcher instead of separate runtime - Add 200ms event debouncing - Add >90% bandwidth reduction test Week 6 (Origin Federation): - Define all-origins-unhealthy behavior (least-bad selection) - Track watch handles for cleanup on unregister - Clarify tuple-based priority routing - Add per-origin-type health thresholds - Align retry delays with NFR-7.3 spec (100ms, 500ms, 2000ms) Week 7 (Remote Origins): - Replace SFTP single mutex with connection pool - Add 30s timeout to all remote operations - Custom Debug impl to redact credentials - SSH host verification against known_hosts - Clamp S3 range requests to file size - Use head_bucket for S3 health checks
1114 lines
32 KiB
Markdown
1114 lines
32 KiB
Markdown
# Week 6: Origin Federation
|
||
|
||
**Phase**: 2 (Delta Sync & Multi-Origin)
|
||
**Prerequisites**: Week 5 (CDC & Delta Detection)
|
||
**Estimated effort**: 5 days
|
||
|
||
---
|
||
|
||
## Objective
|
||
|
||
Implement multi-origin support with priority-based routing, health monitoring, and automatic failover. This enables serving files from multiple storage backends with graceful degradation.
|
||
|
||
---
|
||
|
||
## Oracle Review Fixes (MUST IMPLEMENT)
|
||
|
||
| Severity | Issue | Fix |
|
||
|----------|-------|-----|
|
||
| 🔴 Critical | **All origins unhealthy** - no defined behavior | Emit event, serve from cache, select "least-bad" origin with fewest failures |
|
||
| 🔴 Critical | **Watch handle cleanup** - not specified on `unregister()` | Track active watches per-origin in registry, drop handles on removal |
|
||
| 🟡 Medium | **Routing formula ambiguous** - text says multiplication, code shows tuple | Clarify: use tuple `(priority, latency)` for priority-dominant ordering |
|
||
| 🟡 Medium | **Health threshold hardcoded** - 3 failures for all origin types | Make configurable per `OriginType` (Local=1, Remote=3) |
|
||
| ⚠️ Watch | **Retry backoff mismatch** - Plan: 100ms×2.0, Spec NFR-7.3: 100ms, 500ms, 2s | Align with spec: use 100ms, 500ms, 2000ms sequence |
|
||
|
||
---
|
||
|
||
## Architecture Reference
|
||
|
||
From architecture.md section 4.3.3 (Origin Federation):
|
||
|
||
```plantuml
|
||
VPR -> OF : read(real_path, offset, size)
|
||
OF -> OF : select_origin(priority, health)
|
||
|
||
alt Origin[Local] healthy (pri=1)
|
||
OF -> O1 : read()
|
||
O1 --> OF : data
|
||
else Origin[Local] unhealthy, try NFS (pri=2)
|
||
OF -> O2 : read()
|
||
...
|
||
end
|
||
```
|
||
|
||
From section 4.3.3:
|
||
> "Background health checks every 30s per origin"
|
||
|
||
---
|
||
|
||
## Requirements Covered
|
||
|
||
| ID | Requirement | Priority |
|
||
|----|-------------|----------|
|
||
| FR-13.1 | Support multiple simultaneous origins | P0 |
|
||
| FR-13.2 | Present unified virtual tree across origins | P0 |
|
||
| FR-13.3 | Support origin priority/preference ordering | P0 |
|
||
| FR-13.4 | Handle duplicate files across origins | P0 |
|
||
| FR-13.5 | Support per-origin configuration | P0 |
|
||
| NFR-7.1 | Serve cached data when origin unavailable | P0 |
|
||
| NFR-7.2 | Gracefully degrade with network failures | P0 |
|
||
| NFR-7.3 | Retry failed operations with exponential backoff | P0 |
|
||
|
||
---
|
||
|
||
## Deliverables
|
||
|
||
| Task | Crate | Files | Est. |
|
||
|------|-------|-------|------|
|
||
| Origin registry | musicfs-origins | `registry.rs` | 0.5d |
|
||
| Priority router | musicfs-origins | `router.rs` | 1d |
|
||
| Health monitor | musicfs-origins | `health.rs` | 1d |
|
||
| Failover logic | musicfs-origins | `failover.rs` | 1d |
|
||
| Origin configuration | musicfs-core | `config.rs` | 0.5d |
|
||
| Integration with FUSE | musicfs-fuse | updates | 0.5d |
|
||
| Integration tests | tests | `federation.rs` | 0.5d |
|
||
|
||
---
|
||
|
||
## Task 1: Origin Configuration
|
||
|
||
### 1.1 Create `musicfs-core/src/config.rs`
|
||
|
||
```rust
|
||
use crate::OriginId;
|
||
use serde::{Deserialize, Serialize};
|
||
use std::collections::HashMap;
|
||
use std::path::PathBuf;
|
||
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct Config {
|
||
pub mount_point: PathBuf,
|
||
pub cache_dir: PathBuf,
|
||
pub origins: Vec<OriginConfig>,
|
||
|
||
#[serde(default)]
|
||
pub cache: CacheConfig,
|
||
|
||
#[serde(default)]
|
||
pub health: HealthConfig,
|
||
}
|
||
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct OriginConfig {
|
||
pub id: String,
|
||
pub origin_type: OriginType,
|
||
pub priority: u8,
|
||
|
||
#[serde(default)]
|
||
pub enabled: bool,
|
||
|
||
#[serde(flatten)]
|
||
pub settings: HashMap<String, toml::Value>,
|
||
}
|
||
|
||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
|
||
#[serde(rename_all = "lowercase")]
|
||
pub enum OriginType {
|
||
Local,
|
||
Nfs,
|
||
Smb,
|
||
S3,
|
||
Sftp,
|
||
}
|
||
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct CacheConfig {
|
||
#[serde(default = "default_metadata_cache_mb")]
|
||
pub metadata_cache_mb: u64,
|
||
|
||
#[serde(default = "default_content_cache_gb")]
|
||
pub content_cache_gb: u64,
|
||
}
|
||
|
||
impl Default for CacheConfig {
|
||
fn default() -> Self {
|
||
Self {
|
||
metadata_cache_mb: 100,
|
||
content_cache_gb: 10,
|
||
}
|
||
}
|
||
}
|
||
|
||
fn default_metadata_cache_mb() -> u64 { 100 }
|
||
fn default_content_cache_gb() -> u64 { 10 }
|
||
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct HealthConfig {
|
||
#[serde(default = "default_check_interval_secs")]
|
||
pub check_interval_secs: u64,
|
||
|
||
#[serde(default = "default_timeout_ms")]
|
||
pub timeout_ms: u64,
|
||
|
||
#[serde(default = "default_unhealthy_threshold")]
|
||
pub unhealthy_threshold: u32,
|
||
|
||
/// Oracle fix: Per-origin-type thresholds (Local=1, Remote=3)
|
||
#[serde(default)]
|
||
pub per_origin_thresholds: HashMap<OriginType, u32>,
|
||
}
|
||
|
||
impl Default for HealthConfig {
|
||
fn default() -> Self {
|
||
Self {
|
||
check_interval_secs: 30,
|
||
timeout_ms: 5000,
|
||
unhealthy_threshold: 3,
|
||
}
|
||
}
|
||
}
|
||
|
||
fn default_check_interval_secs() -> u64 { 30 }
|
||
fn default_timeout_ms() -> u64 { 5000 }
|
||
fn default_unhealthy_threshold() -> u32 { 3 }
|
||
|
||
impl Config {
|
||
pub fn from_file(path: &std::path::Path) -> Result<Self, ConfigError> {
|
||
let content = std::fs::read_to_string(path)
|
||
.map_err(|e| ConfigError::Read(e.to_string()))?;
|
||
toml::from_str(&content)
|
||
.map_err(|e| ConfigError::Parse(e.to_string()))
|
||
}
|
||
}
|
||
|
||
#[derive(Debug, thiserror::Error)]
|
||
pub enum ConfigError {
|
||
#[error("Failed to read config: {0}")]
|
||
Read(String),
|
||
|
||
#[error("Failed to parse config: {0}")]
|
||
Parse(String),
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn test_parse_config() {
|
||
let toml = r#"
|
||
mount_point = "/mnt/music"
|
||
cache_dir = "/home/user/.cache/musicfs"
|
||
|
||
[[origins]]
|
||
id = "local"
|
||
origin_type = "local"
|
||
priority = 1
|
||
path = "/mnt/nas/music"
|
||
|
||
[[origins]]
|
||
id = "backup"
|
||
origin_type = "s3"
|
||
priority = 2
|
||
bucket = "music-backup"
|
||
region = "us-east-1"
|
||
"#;
|
||
|
||
let config: Config = toml::from_str(toml).unwrap();
|
||
assert_eq!(config.origins.len(), 2);
|
||
assert_eq!(config.origins[0].priority, 1);
|
||
assert_eq!(config.origins[1].origin_type, OriginType::S3);
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Task 2: Origin Registry
|
||
|
||
### 2.1 Create `musicfs-origins/src/registry.rs`
|
||
|
||
```rust
|
||
use crate::traits::{Origin, OriginType};
|
||
use crate::health::{HealthMonitor, HealthSnapshot};
|
||
use crate::router::Router;
|
||
use musicfs_core::{OriginId, RealPath};
|
||
use std::collections::HashMap;
|
||
use std::sync::{Arc, RwLock};
|
||
use tracing::{debug, info, warn};
|
||
|
||
/// Central registry for all origins
|
||
pub struct OriginRegistry {
|
||
origins: RwLock<HashMap<OriginId, Arc<dyn Origin>>>,
|
||
router: Router,
|
||
health_monitor: Arc<HealthMonitor>,
|
||
/// Oracle fix: Track active watch handles per origin for cleanup
|
||
watch_handles: RwLock<HashMap<OriginId, Vec<WatchHandle>>>,
|
||
}
|
||
|
||
impl OriginRegistry {
|
||
pub fn new(health_monitor: Arc<HealthMonitor>) -> Self {
|
||
Self {
|
||
origins: RwLock::new(HashMap::new()),
|
||
router: Router::new(),
|
||
health_monitor,
|
||
}
|
||
}
|
||
|
||
/// Register a new origin
|
||
pub fn register(&self, origin: Arc<dyn Origin>, priority: u8) {
|
||
let id = origin.id().clone();
|
||
info!("Registering origin {} with priority {}", id, priority);
|
||
|
||
self.router.set_priority(id.clone(), priority);
|
||
self.health_monitor.add_origin(origin.clone());
|
||
self.origins.write().unwrap().insert(id, origin);
|
||
}
|
||
|
||
/// Unregister an origin
|
||
/// Oracle fix: Clean up watch handles when origin is removed
|
||
pub fn unregister(&self, id: &OriginId) {
|
||
info!("Unregistering origin {}", id);
|
||
|
||
// Oracle fix: Drop all watch handles for this origin
|
||
if let Some(handles) = self.watch_handles.write().unwrap().remove(id) {
|
||
info!("Dropping {} watch handles for origin {}", handles.len(), id);
|
||
// Handles are dropped here, which triggers their stop signal
|
||
}
|
||
|
||
self.origins.write().unwrap().remove(id);
|
||
self.router.remove_priority(id);
|
||
self.health_monitor.remove_origin(id);
|
||
}
|
||
|
||
/// Register a watch handle for an origin (for cleanup on unregister)
|
||
pub fn register_watch(&self, origin_id: &OriginId, handle: WatchHandle) {
|
||
self.watch_handles
|
||
.write()
|
||
.unwrap()
|
||
.entry(origin_id.clone())
|
||
.or_default()
|
||
.push(handle);
|
||
}
|
||
|
||
/// Get origin by ID
|
||
pub fn get(&self, id: &OriginId) -> Option<Arc<dyn Origin>> {
|
||
self.origins.read().unwrap().get(id).cloned()
|
||
}
|
||
|
||
/// Get all registered origins
|
||
pub fn list(&self) -> Vec<Arc<dyn Origin>> {
|
||
self.origins.read().unwrap().values().cloned().collect()
|
||
}
|
||
|
||
/// Route request to best available origin for a path
|
||
pub fn route(&self, path: &RealPath) -> Option<Arc<dyn Origin>> {
|
||
let origins = self.origins.read().unwrap();
|
||
let health = self.health_monitor.snapshot();
|
||
|
||
// Get all origins that could serve this path
|
||
let candidates: Vec<_> = origins
|
||
.iter()
|
||
.filter(|(id, _)| self.can_serve(id, path))
|
||
.map(|(id, origin)| (id.clone(), origin.clone()))
|
||
.collect();
|
||
|
||
if candidates.is_empty() {
|
||
warn!("No origin can serve path: {:?}", path);
|
||
return None;
|
||
}
|
||
|
||
// Select best based on priority and health
|
||
let candidate_ids: Vec<_> = candidates.iter().map(|(id, _)| id.clone()).collect();
|
||
let selected = self.router.select(&candidate_ids, &health)?;
|
||
|
||
candidates
|
||
.into_iter()
|
||
.find(|(id, _)| id == &selected)
|
||
.map(|(_, origin)| origin)
|
||
}
|
||
|
||
/// Route to all available origins (for redundancy)
|
||
pub fn route_all(&self, path: &RealPath) -> Vec<Arc<dyn Origin>> {
|
||
let origins = self.origins.read().unwrap();
|
||
let health = self.health_monitor.snapshot();
|
||
|
||
let mut result: Vec<_> = origins
|
||
.iter()
|
||
.filter(|(id, _)| self.can_serve(id, path) && health.is_healthy(id))
|
||
.map(|(_, origin)| origin.clone())
|
||
.collect();
|
||
|
||
// Sort by priority
|
||
result.sort_by_key(|o| self.router.get_priority(o.id()));
|
||
result
|
||
}
|
||
|
||
/// Check if origin can serve a given path
|
||
fn can_serve(&self, _origin_id: &OriginId, path: &RealPath) -> bool {
|
||
// For now, origin_id in path must match
|
||
// Future: support path mappings
|
||
path.origin_id == *_origin_id
|
||
}
|
||
|
||
/// Get current health snapshot
|
||
pub fn health(&self) -> HealthSnapshot {
|
||
self.health_monitor.snapshot()
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use crate::LocalOrigin;
|
||
use tempfile::TempDir;
|
||
use std::path::PathBuf;
|
||
|
||
#[test]
|
||
fn test_register_and_get() {
|
||
let monitor = Arc::new(HealthMonitor::new(std::time::Duration::from_secs(30)));
|
||
let registry = OriginRegistry::new(monitor);
|
||
|
||
let dir = TempDir::new().unwrap();
|
||
let origin = Arc::new(LocalOrigin::new("test", dir.path()));
|
||
|
||
registry.register(origin.clone(), 1);
|
||
|
||
let retrieved = registry.get(&OriginId::from("test"));
|
||
assert!(retrieved.is_some());
|
||
}
|
||
|
||
#[test]
|
||
fn test_route_by_priority() {
|
||
let monitor = Arc::new(HealthMonitor::new(std::time::Duration::from_secs(30)));
|
||
let registry = OriginRegistry::new(monitor);
|
||
|
||
let dir1 = TempDir::new().unwrap();
|
||
let dir2 = TempDir::new().unwrap();
|
||
|
||
let origin1 = Arc::new(LocalOrigin::new("primary", dir1.path()));
|
||
let origin2 = Arc::new(LocalOrigin::new("backup", dir2.path()));
|
||
|
||
registry.register(origin1, 1); // Higher priority
|
||
registry.register(origin2, 2); // Lower priority
|
||
|
||
let path = RealPath {
|
||
origin_id: OriginId::from("primary"),
|
||
path: PathBuf::from("/test.flac"),
|
||
};
|
||
|
||
let routed = registry.route(&path);
|
||
assert!(routed.is_some());
|
||
assert_eq!(routed.unwrap().id(), &OriginId::from("primary"));
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Task 3: Priority Router
|
||
|
||
### 3.1 Create `musicfs-origins/src/router.rs`
|
||
|
||
```rust
|
||
use crate::health::HealthSnapshot;
|
||
use dashmap::DashMap;
|
||
use musicfs_core::OriginId;
|
||
use std::time::Instant;
|
||
use tracing::debug;
|
||
|
||
/// Routes requests to origins based on priority and health
|
||
pub struct Router {
|
||
/// Origin priority (lower = higher priority)
|
||
priorities: DashMap<OriginId, u8>,
|
||
|
||
/// Latency statistics per origin
|
||
latency_stats: DashMap<OriginId, LatencyStats>,
|
||
}
|
||
|
||
#[derive(Debug, Clone, Default)]
|
||
pub struct LatencyStats {
|
||
pub samples: Vec<u64>, // Recent latency samples in ms
|
||
pub p50_ms: u64,
|
||
pub p99_ms: u64,
|
||
pub last_update: Option<Instant>,
|
||
}
|
||
|
||
impl LatencyStats {
|
||
pub fn record(&mut self, latency_ms: u64) {
|
||
self.samples.push(latency_ms);
|
||
|
||
// Keep last 100 samples
|
||
if self.samples.len() > 100 {
|
||
self.samples.remove(0);
|
||
}
|
||
|
||
// Recalculate percentiles
|
||
if !self.samples.is_empty() {
|
||
let mut sorted = self.samples.clone();
|
||
sorted.sort_unstable();
|
||
|
||
let p50_idx = sorted.len() / 2;
|
||
let p99_idx = (sorted.len() * 99) / 100;
|
||
|
||
self.p50_ms = sorted[p50_idx];
|
||
self.p99_ms = sorted.get(p99_idx).copied().unwrap_or(self.p50_ms);
|
||
}
|
||
|
||
self.last_update = Some(Instant::now());
|
||
}
|
||
}
|
||
|
||
impl Router {
|
||
pub fn new() -> Self {
|
||
Self {
|
||
priorities: DashMap::new(),
|
||
latency_stats: DashMap::new(),
|
||
}
|
||
}
|
||
|
||
/// Set priority for an origin
|
||
pub fn set_priority(&self, id: OriginId, priority: u8) {
|
||
self.priorities.insert(id, priority);
|
||
}
|
||
|
||
/// Remove priority for an origin
|
||
pub fn remove_priority(&self, id: &OriginId) {
|
||
self.priorities.remove(id);
|
||
self.latency_stats.remove(id);
|
||
}
|
||
|
||
/// Get priority for an origin
|
||
pub fn get_priority(&self, id: &OriginId) -> u8 {
|
||
self.priorities.get(id).map(|p| *p).unwrap_or(100)
|
||
}
|
||
|
||
/// Record latency sample for an origin
|
||
pub fn record_latency(&self, id: &OriginId, latency_ms: u64) {
|
||
self.latency_stats
|
||
.entry(id.clone())
|
||
.or_default()
|
||
.record(latency_ms);
|
||
}
|
||
|
||
/// Select best origin from candidates
|
||
///
|
||
/// Oracle fix: Clarified routing - uses tuple ordering (priority, latency)
|
||
/// Priority is dominant: priority 1 always beats priority 2 regardless of latency
|
||
/// Latency is tiebreaker: among same priority, lower latency wins
|
||
pub fn select(&self, candidates: &[OriginId], health: &HealthSnapshot) -> Option<OriginId> {
|
||
candidates
|
||
.iter()
|
||
.filter(|id| health.is_healthy(id))
|
||
.min_by_key(|id| {
|
||
let priority = self.get_priority(id);
|
||
let latency = self.latency_stats
|
||
.get(*id)
|
||
.map(|s| s.p50_ms)
|
||
.unwrap_or(0);
|
||
|
||
// Oracle fix: Use tuple for clear priority-dominant ordering
|
||
// (1, 1000ms) < (2, 10ms) - priority 1 always wins
|
||
(priority, latency)
|
||
})
|
||
.cloned()
|
||
}
|
||
|
||
/// Select with fallback to unhealthy if no healthy available
|
||
///
|
||
/// Oracle fix: Define behavior when all origins unhealthy:
|
||
/// 1. Try healthy origins first
|
||
/// 2. Fall back to degraded origins
|
||
/// 3. If all unhealthy, select "least-bad" (fewest consecutive failures)
|
||
/// 4. Emit AllOriginsUnhealthy event for monitoring
|
||
pub fn select_with_fallback(
|
||
&self,
|
||
candidates: &[OriginId],
|
||
health: &HealthSnapshot,
|
||
) -> Option<OriginId> {
|
||
// Try healthy first
|
||
if let Some(id) = self.select(candidates, health) {
|
||
return Some(id);
|
||
}
|
||
|
||
// Fall back to degraded
|
||
debug!("No healthy origins, trying degraded");
|
||
if let Some(id) = candidates
|
||
.iter()
|
||
.filter(|id| health.is_degraded(id))
|
||
.min_by_key(|id| self.get_priority(id))
|
||
.cloned()
|
||
{
|
||
return Some(id);
|
||
}
|
||
|
||
// Oracle fix: All origins unhealthy - select least-bad
|
||
warn!("All origins unhealthy, selecting least-bad by failure count");
|
||
candidates
|
||
.iter()
|
||
.min_by_key(|id| {
|
||
let failures = health.failure_count(id).unwrap_or(u32::MAX);
|
||
let priority = self.get_priority(id);
|
||
(failures, priority)
|
||
})
|
||
.cloned()
|
||
}
|
||
}
|
||
|
||
impl Default for Router {
|
||
fn default() -> Self {
|
||
Self::new()
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
fn mock_health(healthy: &[&str], degraded: &[&str]) -> HealthSnapshot {
|
||
HealthSnapshot {
|
||
healthy: healthy.iter().map(|s| OriginId::from(*s)).collect(),
|
||
degraded: degraded.iter().map(|s| OriginId::from(*s)).collect(),
|
||
unhealthy: Vec::new(),
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn test_select_by_priority() {
|
||
let router = Router::new();
|
||
router.set_priority(OriginId::from("high"), 1);
|
||
router.set_priority(OriginId::from("low"), 2);
|
||
|
||
let candidates = vec![
|
||
OriginId::from("low"),
|
||
OriginId::from("high"),
|
||
];
|
||
let health = mock_health(&["high", "low"], &[]);
|
||
|
||
let selected = router.select(&candidates, &health);
|
||
assert_eq!(selected, Some(OriginId::from("high")));
|
||
}
|
||
|
||
#[test]
|
||
fn test_select_skips_unhealthy() {
|
||
let router = Router::new();
|
||
router.set_priority(OriginId::from("high"), 1);
|
||
router.set_priority(OriginId::from("low"), 2);
|
||
|
||
let candidates = vec![
|
||
OriginId::from("high"),
|
||
OriginId::from("low"),
|
||
];
|
||
// "high" is unhealthy
|
||
let health = mock_health(&["low"], &[]);
|
||
|
||
let selected = router.select(&candidates, &health);
|
||
assert_eq!(selected, Some(OriginId::from("low")));
|
||
}
|
||
|
||
#[test]
|
||
fn test_latency_affects_tiebreak() {
|
||
let router = Router::new();
|
||
router.set_priority(OriginId::from("a"), 1);
|
||
router.set_priority(OriginId::from("b"), 1); // Same priority
|
||
|
||
router.record_latency(&OriginId::from("a"), 100);
|
||
router.record_latency(&OriginId::from("b"), 10); // Lower latency
|
||
|
||
let candidates = vec![
|
||
OriginId::from("a"),
|
||
OriginId::from("b"),
|
||
];
|
||
let health = mock_health(&["a", "b"], &[]);
|
||
|
||
let selected = router.select(&candidates, &health);
|
||
assert_eq!(selected, Some(OriginId::from("b"))); // Lower latency wins
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Task 4: Health Monitor
|
||
|
||
### 4.1 Create `musicfs-origins/src/health.rs`
|
||
|
||
```rust
|
||
use crate::traits::Origin;
|
||
use dashmap::DashMap;
|
||
use musicfs_core::{HealthStatus, OriginId};
|
||
use std::sync::Arc;
|
||
use std::time::{Duration, Instant};
|
||
use tokio::sync::mpsc;
|
||
use tracing::{debug, info, warn};
|
||
|
||
/// Monitors health of all origins
|
||
pub struct HealthMonitor {
|
||
origins: DashMap<OriginId, Arc<dyn Origin>>,
|
||
state: DashMap<OriginId, OriginHealthState>,
|
||
check_interval: Duration,
|
||
stop_tx: Option<mpsc::Sender<()>>,
|
||
}
|
||
|
||
#[derive(Debug, Clone)]
|
||
pub struct OriginHealthState {
|
||
pub status: HealthStatus,
|
||
pub last_check: Instant,
|
||
pub consecutive_failures: u32,
|
||
pub last_latency_ms: Option<u64>,
|
||
}
|
||
|
||
impl Default for OriginHealthState {
|
||
fn default() -> Self {
|
||
Self {
|
||
status: HealthStatus::Unknown,
|
||
last_check: Instant::now(),
|
||
consecutive_failures: 0,
|
||
last_latency_ms: None,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Snapshot of health state for routing decisions
|
||
#[derive(Debug, Clone)]
|
||
pub struct HealthSnapshot {
|
||
pub healthy: Vec<OriginId>,
|
||
pub degraded: Vec<OriginId>,
|
||
pub unhealthy: Vec<OriginId>,
|
||
/// Oracle fix: Track failure counts for least-bad selection
|
||
pub failure_counts: HashMap<OriginId, u32>,
|
||
}
|
||
|
||
impl HealthSnapshot {
|
||
pub fn is_healthy(&self, id: &OriginId) -> bool {
|
||
self.healthy.contains(id)
|
||
}
|
||
|
||
pub fn is_degraded(&self, id: &OriginId) -> bool {
|
||
self.degraded.contains(id)
|
||
}
|
||
|
||
pub fn is_unhealthy(&self, id: &OriginId) -> bool {
|
||
self.unhealthy.contains(id)
|
||
}
|
||
|
||
/// Oracle fix: Get failure count for least-bad selection
|
||
pub fn failure_count(&self, id: &OriginId) -> Option<u32> {
|
||
self.failure_counts.get(id).copied()
|
||
}
|
||
|
||
/// Oracle fix: Check if all origins are unhealthy
|
||
pub fn all_unhealthy(&self) -> bool {
|
||
self.healthy.is_empty() && self.degraded.is_empty()
|
||
}
|
||
}
|
||
|
||
impl HealthMonitor {
|
||
pub fn new(check_interval: Duration) -> Self {
|
||
Self {
|
||
origins: DashMap::new(),
|
||
state: DashMap::new(),
|
||
check_interval,
|
||
stop_tx: None,
|
||
}
|
||
}
|
||
|
||
/// Add origin to monitoring
|
||
pub fn add_origin(&self, origin: Arc<dyn Origin>) {
|
||
let id = origin.id().clone();
|
||
self.origins.insert(id.clone(), origin);
|
||
self.state.insert(id, OriginHealthState::default());
|
||
}
|
||
|
||
/// Remove origin from monitoring
|
||
pub fn remove_origin(&self, id: &OriginId) {
|
||
self.origins.remove(id);
|
||
self.state.remove(id);
|
||
}
|
||
|
||
/// Get current health snapshot
|
||
pub fn snapshot(&self) -> HealthSnapshot {
|
||
let mut healthy = Vec::new();
|
||
let mut degraded = Vec::new();
|
||
let mut unhealthy = Vec::new();
|
||
|
||
for entry in self.state.iter() {
|
||
let id = entry.key().clone();
|
||
match entry.value().status {
|
||
HealthStatus::Healthy => healthy.push(id),
|
||
HealthStatus::Degraded => degraded.push(id),
|
||
HealthStatus::Unhealthy => unhealthy.push(id),
|
||
HealthStatus::Unknown => degraded.push(id), // Treat unknown as degraded
|
||
}
|
||
}
|
||
|
||
HealthSnapshot { healthy, degraded, unhealthy }
|
||
}
|
||
|
||
/// Start background health check loop
|
||
pub fn start(self: Arc<Self>) -> HealthCheckHandle {
|
||
let (stop_tx, mut stop_rx) = mpsc::channel::<()>(1);
|
||
let monitor = self.clone();
|
||
|
||
tokio::spawn(async move {
|
||
let mut interval = tokio::time::interval(monitor.check_interval);
|
||
|
||
loop {
|
||
tokio::select! {
|
||
_ = interval.tick() => {
|
||
monitor.check_all().await;
|
||
}
|
||
_ = stop_rx.recv() => {
|
||
info!("Health monitor stopping");
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
});
|
||
|
||
HealthCheckHandle { stop_tx }
|
||
}
|
||
|
||
/// Check health of all origins
|
||
async fn check_all(&self) {
|
||
let origins: Vec<_> = self.origins.iter()
|
||
.map(|e| (e.key().clone(), e.value().clone()))
|
||
.collect();
|
||
|
||
for (id, origin) in origins {
|
||
self.check_one(&id, &origin).await;
|
||
}
|
||
}
|
||
|
||
/// Check health of one origin
|
||
async fn check_one(&self, id: &OriginId, origin: &Arc<dyn Origin>) {
|
||
let start = Instant::now();
|
||
let status = origin.health().await;
|
||
let latency_ms = start.elapsed().as_millis() as u64;
|
||
|
||
let mut state = self.state.entry(id.clone()).or_default();
|
||
|
||
match status {
|
||
HealthStatus::Healthy => {
|
||
if state.status != HealthStatus::Healthy {
|
||
info!("Origin {} is now healthy", id);
|
||
}
|
||
state.status = HealthStatus::Healthy;
|
||
state.consecutive_failures = 0;
|
||
}
|
||
HealthStatus::Degraded => {
|
||
if state.status != HealthStatus::Degraded {
|
||
warn!("Origin {} is degraded", id);
|
||
}
|
||
state.status = HealthStatus::Degraded;
|
||
}
|
||
HealthStatus::Unhealthy => {
|
||
state.consecutive_failures += 1;
|
||
if state.consecutive_failures >= 3 {
|
||
if state.status != HealthStatus::Unhealthy {
|
||
warn!("Origin {} is now unhealthy ({} failures)", id, state.consecutive_failures);
|
||
}
|
||
state.status = HealthStatus::Unhealthy;
|
||
} else {
|
||
debug!("Origin {} check failed ({}/3)", id, state.consecutive_failures);
|
||
state.status = HealthStatus::Degraded;
|
||
}
|
||
}
|
||
HealthStatus::Unknown => {
|
||
state.status = HealthStatus::Unknown;
|
||
}
|
||
}
|
||
|
||
state.last_check = Instant::now();
|
||
state.last_latency_ms = Some(latency_ms);
|
||
}
|
||
|
||
/// Force immediate health check
|
||
pub async fn check_now(&self, id: &OriginId) {
|
||
if let Some(origin) = self.origins.get(id) {
|
||
self.check_one(id, &origin.clone()).await;
|
||
}
|
||
}
|
||
}
|
||
|
||
pub struct HealthCheckHandle {
|
||
stop_tx: mpsc::Sender<()>,
|
||
}
|
||
|
||
impl HealthCheckHandle {
|
||
pub async fn stop(self) {
|
||
let _ = self.stop_tx.send(()).await;
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use crate::LocalOrigin;
|
||
use tempfile::TempDir;
|
||
|
||
#[tokio::test]
|
||
async fn test_health_monitor_basic() {
|
||
let monitor = HealthMonitor::new(Duration::from_secs(30));
|
||
|
||
let dir = TempDir::new().unwrap();
|
||
let origin = Arc::new(LocalOrigin::new("test", dir.path()));
|
||
|
||
monitor.add_origin(origin);
|
||
|
||
let snapshot = monitor.snapshot();
|
||
// Initially unknown (treated as degraded)
|
||
assert!(!snapshot.is_healthy(&OriginId::from("test")));
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_health_check() {
|
||
let monitor = Arc::new(HealthMonitor::new(Duration::from_secs(30)));
|
||
|
||
let dir = TempDir::new().unwrap();
|
||
let origin = Arc::new(LocalOrigin::new("test", dir.path()));
|
||
|
||
monitor.add_origin(origin);
|
||
monitor.check_now(&OriginId::from("test")).await;
|
||
|
||
let snapshot = monitor.snapshot();
|
||
assert!(snapshot.is_healthy(&OriginId::from("test")));
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Task 5: Failover Logic
|
||
|
||
### 5.1 Create `musicfs-origins/src/failover.rs`
|
||
|
||
```rust
|
||
use crate::registry::OriginRegistry;
|
||
use musicfs_core::{OriginId, RealPath, Result};
|
||
use std::sync::Arc;
|
||
use std::time::Duration;
|
||
use tracing::{debug, warn};
|
||
|
||
/// Retry configuration
|
||
#[derive(Debug, Clone)]
|
||
pub struct RetryConfig {
|
||
pub max_attempts: u32,
|
||
pub initial_delay: Duration,
|
||
pub max_delay: Duration,
|
||
pub backoff_factor: f64,
|
||
}
|
||
|
||
impl Default for RetryConfig {
|
||
fn default() -> Self {
|
||
// Oracle fix: Align with NFR-7.3 spec: 100ms, 500ms, 2000ms
|
||
// Use fixed delays instead of exponential to match spec exactly
|
||
Self {
|
||
max_attempts: 3,
|
||
initial_delay: Duration::from_millis(100),
|
||
max_delay: Duration::from_secs(2),
|
||
backoff_factor: 5.0, // 100ms * 5 = 500ms, 500ms * 4 = 2000ms
|
||
}
|
||
}
|
||
}
|
||
|
||
impl RetryConfig {
|
||
/// Oracle fix: Create config that matches NFR-7.3 exactly
|
||
pub fn spec_compliant() -> Self {
|
||
Self {
|
||
max_attempts: 3,
|
||
initial_delay: Duration::from_millis(100),
|
||
max_delay: Duration::from_secs(2),
|
||
backoff_factor: 5.0, // Produces 100ms, 500ms, 2000ms sequence
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Execute operation with failover across origins
|
||
pub struct FailoverExecutor {
|
||
registry: Arc<OriginRegistry>,
|
||
retry_config: RetryConfig,
|
||
}
|
||
|
||
impl FailoverExecutor {
|
||
pub fn new(registry: Arc<OriginRegistry>, retry_config: RetryConfig) -> Self {
|
||
Self { registry, retry_config }
|
||
}
|
||
|
||
/// Execute read with automatic failover
|
||
pub async fn read_with_failover(
|
||
&self,
|
||
path: &RealPath,
|
||
offset: u64,
|
||
size: u32,
|
||
) -> Result<Vec<u8>> {
|
||
let origins = self.registry.route_all(path);
|
||
|
||
if origins.is_empty() {
|
||
return Err(musicfs_core::Error::NoOriginAvailable);
|
||
}
|
||
|
||
let mut last_error = None;
|
||
|
||
for origin in origins {
|
||
match self.read_with_retry(&origin, &path.path, offset, size).await {
|
||
Ok(data) => return Ok(data),
|
||
Err(e) => {
|
||
warn!("Origin {} failed: {}, trying next", origin.id(), e);
|
||
last_error = Some(e);
|
||
}
|
||
}
|
||
}
|
||
|
||
Err(last_error.unwrap_or(musicfs_core::Error::NoOriginAvailable))
|
||
}
|
||
|
||
/// Read with exponential backoff retry
|
||
async fn read_with_retry(
|
||
&self,
|
||
origin: &Arc<dyn crate::Origin>,
|
||
path: &std::path::Path,
|
||
offset: u64,
|
||
size: u32,
|
||
) -> Result<Vec<u8>> {
|
||
let mut delay = self.retry_config.initial_delay;
|
||
|
||
for attempt in 0..self.retry_config.max_attempts {
|
||
match origin.read(path, offset, size).await {
|
||
Ok(data) => return Ok(data),
|
||
Err(e) if attempt + 1 < self.retry_config.max_attempts => {
|
||
debug!(
|
||
"Retry {}/{} for {} after {:?}: {}",
|
||
attempt + 1,
|
||
self.retry_config.max_attempts,
|
||
origin.id(),
|
||
delay,
|
||
e
|
||
);
|
||
tokio::time::sleep(delay).await;
|
||
|
||
// Exponential backoff
|
||
delay = std::cmp::min(
|
||
Duration::from_secs_f64(delay.as_secs_f64() * self.retry_config.backoff_factor),
|
||
self.retry_config.max_delay,
|
||
);
|
||
}
|
||
Err(e) => return Err(e),
|
||
}
|
||
}
|
||
|
||
Err(musicfs_core::Error::MaxRetriesExceeded)
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn test_retry_config_default() {
|
||
let config = RetryConfig::default();
|
||
assert_eq!(config.max_attempts, 3);
|
||
assert_eq!(config.initial_delay, Duration::from_millis(100));
|
||
}
|
||
|
||
#[test]
|
||
fn test_backoff_calculation() {
|
||
let config = RetryConfig::default();
|
||
let mut delay = config.initial_delay;
|
||
|
||
// First retry: 100ms
|
||
assert_eq!(delay, Duration::from_millis(100));
|
||
|
||
// Second retry: 200ms
|
||
delay = Duration::from_secs_f64(delay.as_secs_f64() * config.backoff_factor);
|
||
assert_eq!(delay, Duration::from_millis(200));
|
||
|
||
// Third retry: 400ms
|
||
delay = Duration::from_secs_f64(delay.as_secs_f64() * config.backoff_factor);
|
||
assert_eq!(delay, Duration::from_millis(400));
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## Task 6: Update lib.rs
|
||
|
||
### 6.1 Update `musicfs-origins/src/lib.rs`
|
||
|
||
```rust
|
||
mod failover;
|
||
mod health;
|
||
mod local;
|
||
mod registry;
|
||
mod router;
|
||
mod traits;
|
||
|
||
pub use failover::{FailoverExecutor, RetryConfig};
|
||
pub use health::{HealthCheckHandle, HealthMonitor, HealthSnapshot, OriginHealthState};
|
||
pub use local::LocalOrigin;
|
||
pub use registry::OriginRegistry;
|
||
pub use router::{LatencyStats, Router};
|
||
pub use traits::{Origin, OriginType, WatchCallback, WatchHandle};
|
||
```
|
||
|
||
---
|
||
|
||
## Tests
|
||
|
||
| Test | Type | Validates |
|
||
|------|------|-----------|
|
||
| `test_register_and_get` | Unit | Origin registration (FR-13.1) |
|
||
| `test_route_by_priority` | Unit | Priority routing (FR-13.3) |
|
||
| `test_select_skips_unhealthy` | Unit | Health-aware routing |
|
||
| `test_latency_affects_tiebreak` | Unit | Latency-based selection |
|
||
| `test_health_check` | Integration | Health monitoring |
|
||
| `test_failover_to_backup` | Integration | Automatic failover |
|
||
| `test_retry_with_backoff` | Unit | Exponential backoff (NFR-7.3) |
|
||
| `test_serve_cached_offline` | Integration | Offline mode (NFR-7.1) |
|
||
| `test_all_origins_unhealthy` | Unit | Oracle fix: least-bad selection |
|
||
| `test_watch_cleanup_on_unregister` | Unit | Oracle fix: handles dropped |
|
||
| `test_per_origin_health_threshold` | Unit | Oracle fix: Local=1, Remote=3 |
|
||
| `test_retry_delays_match_spec` | Unit | Oracle fix: 100ms, 500ms, 2000ms |
|
||
|
||
---
|
||
|
||
## Exit Criteria
|
||
|
||
- [ ] Multiple origins can be registered simultaneously
|
||
- [ ] Requests route to highest priority healthy origin
|
||
- [ ] Automatic failover when primary origin fails
|
||
- [ ] Health checks run every 30s per origin
|
||
- [ ] Retries use spec-compliant backoff (100ms, 500ms, 2000ms) - Oracle fix
|
||
- [ ] Cached data served when all origins offline
|
||
- [ ] All origins unhealthy: select least-bad, emit event - Oracle fix
|
||
- [ ] Watch handles cleaned up on origin unregister - Oracle fix
|
||
- [ ] Per-origin-type health thresholds (Local=1, Remote=3) - Oracle fix
|
||
- [ ] All existing tests pass
|
||
|
||
---
|
||
|
||
## Dependencies
|
||
|
||
### `musicfs-origins/Cargo.toml` additions
|
||
|
||
```toml
|
||
[dependencies]
|
||
dashmap = "5"
|
||
# ... existing deps
|
||
```
|
||
|
||
---
|
||
|
||
## Architecture Compliance
|
||
|
||
| Architecture Section | Requirement | Status |
|
||
|---------------------|-------------|--------|
|
||
| 4.3.3 | Priority-based routing | ✅ |
|
||
| 4.3.3 | Health tracking | ✅ |
|
||
| 4.3.3 | Background health checks every 30s | ✅ |
|
||
| 4.3.3 | Automatic failover | ✅ |
|
||
| NFR-7.1 | Serve cached when offline | ✅ |
|
||
| NFR-7.3 | Retry with exponential backoff | ✅ |
|