Implement Week 7 Remote Origins with Oracle fixes
- Add credentials.rs with CredentialStore, redacted Debug (session_token shows [REDACTED]) - Add nfs.rs with ESTALE retry using Fn closure, 5s health timeout - Add smb.rs with ENOTCONN retry handling, 5s health timeout - Add s3.rs/sftp.rs feature-gated stubs with security documentation - Add error variants: S3, Sftp, Timeout, Credential, NfsStaleHandle - Fix delta.rs unused imports Oracle fixes applied: - SMB retry_on_disconnect for ENOTCONN (errno 107) - session_token Debug shows [REDACTED] when Some, None otherwise - NFS/SMB health checks wrapped with tokio::time::timeout(5s) 102 tests pass, 0 warnings.
This commit is contained in:
Generated
+3
@@ -687,6 +687,7 @@ version = "0.1.0"
|
|||||||
dependencies = [
|
dependencies = [
|
||||||
"hex",
|
"hex",
|
||||||
"serde",
|
"serde",
|
||||||
|
"serde_json",
|
||||||
"tempfile",
|
"tempfile",
|
||||||
"thiserror",
|
"thiserror",
|
||||||
"tokio",
|
"tokio",
|
||||||
@@ -727,8 +728,10 @@ version = "0.1.0"
|
|||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"dashmap",
|
"dashmap",
|
||||||
|
"libc",
|
||||||
"musicfs-core",
|
"musicfs-core",
|
||||||
"tempfile",
|
"tempfile",
|
||||||
|
"thiserror",
|
||||||
"tokio",
|
"tokio",
|
||||||
"tracing",
|
"tracing",
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -59,3 +59,6 @@ clap = { version = "4", features = ["derive"] }
|
|||||||
|
|
||||||
# Testing
|
# Testing
|
||||||
tempfile = "3"
|
tempfile = "3"
|
||||||
|
|
||||||
|
# Platform-specific
|
||||||
|
libc = "0.2"
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ edition.workspace = true
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
thiserror.workspace = true
|
thiserror.workspace = true
|
||||||
serde.workspace = true
|
serde.workspace = true
|
||||||
|
serde_json.workspace = true
|
||||||
toml.workspace = true
|
toml.workspace = true
|
||||||
tokio = { workspace = true, features = ["sync"] }
|
tokio = { workspace = true, features = ["sync"] }
|
||||||
xxhash-rust.workspace = true
|
xxhash-rust.workspace = true
|
||||||
|
|||||||
@@ -0,0 +1,262 @@
|
|||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::path::PathBuf;
|
||||||
|
use thiserror::Error;
|
||||||
|
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct CredentialStore {
|
||||||
|
cache: HashMap<String, Credential>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::fmt::Debug for CredentialStore {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
f.debug_struct("CredentialStore")
|
||||||
|
.field("cache_keys", &self.cache.keys().collect::<Vec<_>>())
|
||||||
|
.finish()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Serialize, Deserialize)]
|
||||||
|
#[serde(tag = "type")]
|
||||||
|
pub enum Credential {
|
||||||
|
Basic {
|
||||||
|
username: String,
|
||||||
|
#[serde(skip_serializing)]
|
||||||
|
password: String,
|
||||||
|
},
|
||||||
|
|
||||||
|
AwsKey {
|
||||||
|
access_key_id: String,
|
||||||
|
#[serde(skip_serializing)]
|
||||||
|
secret_access_key: String,
|
||||||
|
session_token: Option<String>,
|
||||||
|
region: String,
|
||||||
|
},
|
||||||
|
|
||||||
|
SshKey {
|
||||||
|
username: String,
|
||||||
|
private_key_path: PathBuf,
|
||||||
|
#[serde(skip_serializing)]
|
||||||
|
passphrase: Option<String>,
|
||||||
|
},
|
||||||
|
|
||||||
|
EnvVar {
|
||||||
|
var_name: String,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::fmt::Debug for Credential {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
match self {
|
||||||
|
Self::Basic { username, .. } => f
|
||||||
|
.debug_struct("Basic")
|
||||||
|
.field("username", username)
|
||||||
|
.field("password", &"[REDACTED]")
|
||||||
|
.finish(),
|
||||||
|
Self::AwsKey {
|
||||||
|
access_key_id,
|
||||||
|
session_token,
|
||||||
|
region,
|
||||||
|
..
|
||||||
|
} => {
|
||||||
|
let key_preview = if access_key_id.len() > 4 {
|
||||||
|
format!("{}...", &access_key_id[..4])
|
||||||
|
} else {
|
||||||
|
"****".to_string()
|
||||||
|
};
|
||||||
|
let token_display = if session_token.is_some() {
|
||||||
|
"[REDACTED]"
|
||||||
|
} else {
|
||||||
|
"None"
|
||||||
|
};
|
||||||
|
f.debug_struct("AwsKey")
|
||||||
|
.field("access_key_id", &key_preview)
|
||||||
|
.field("secret_access_key", &"[REDACTED]")
|
||||||
|
.field("session_token", &token_display)
|
||||||
|
.field("region", region)
|
||||||
|
.finish()
|
||||||
|
}
|
||||||
|
Self::SshKey {
|
||||||
|
username,
|
||||||
|
private_key_path,
|
||||||
|
..
|
||||||
|
} => f
|
||||||
|
.debug_struct("SshKey")
|
||||||
|
.field("username", username)
|
||||||
|
.field("private_key_path", private_key_path)
|
||||||
|
.field("passphrase", &"[REDACTED]")
|
||||||
|
.finish(),
|
||||||
|
Self::EnvVar { var_name } => f
|
||||||
|
.debug_struct("EnvVar")
|
||||||
|
.field("var_name", var_name)
|
||||||
|
.finish(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl CredentialStore {
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
cache: HashMap::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn load(
|
||||||
|
&mut self,
|
||||||
|
origin_id: &str,
|
||||||
|
config: &CredentialConfig,
|
||||||
|
) -> Result<Credential, CredentialError> {
|
||||||
|
if let Some(cred) = self.cache.get(origin_id) {
|
||||||
|
return Ok(cred.clone());
|
||||||
|
}
|
||||||
|
|
||||||
|
let cred = match config {
|
||||||
|
CredentialConfig::Environment { prefix } => self.load_from_env(prefix)?,
|
||||||
|
CredentialConfig::File { path } => self.load_from_file(path)?,
|
||||||
|
CredentialConfig::Inline(cred) => cred.clone(),
|
||||||
|
};
|
||||||
|
|
||||||
|
self.cache.insert(origin_id.to_string(), cred.clone());
|
||||||
|
Ok(cred)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn load_from_env(&self, prefix: &str) -> Result<Credential, CredentialError> {
|
||||||
|
if let (Ok(key), Ok(secret)) = (
|
||||||
|
std::env::var(format!("{}_ACCESS_KEY_ID", prefix)),
|
||||||
|
std::env::var(format!("{}_SECRET_ACCESS_KEY", prefix)),
|
||||||
|
) {
|
||||||
|
return Ok(Credential::AwsKey {
|
||||||
|
access_key_id: key,
|
||||||
|
secret_access_key: secret,
|
||||||
|
session_token: std::env::var(format!("{}_SESSION_TOKEN", prefix)).ok(),
|
||||||
|
region: std::env::var(format!("{}_REGION", prefix))
|
||||||
|
.unwrap_or_else(|_| "us-east-1".to_string()),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
if let (Ok(user), Ok(pass)) = (
|
||||||
|
std::env::var(format!("{}_USERNAME", prefix)),
|
||||||
|
std::env::var(format!("{}_PASSWORD", prefix)),
|
||||||
|
) {
|
||||||
|
return Ok(Credential::Basic {
|
||||||
|
username: user,
|
||||||
|
password: pass,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
Err(CredentialError::NotFound(format!(
|
||||||
|
"No credentials found with prefix {}",
|
||||||
|
prefix
|
||||||
|
)))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn load_from_file(&self, path: &PathBuf) -> Result<Credential, CredentialError> {
|
||||||
|
let content =
|
||||||
|
std::fs::read_to_string(path).map_err(|e| CredentialError::FileRead(e.to_string()))?;
|
||||||
|
|
||||||
|
if path.extension().map(|e| e == "json").unwrap_or(false) {
|
||||||
|
serde_json::from_str(&content).map_err(|e| CredentialError::Parse(e.to_string()))
|
||||||
|
} else {
|
||||||
|
toml::from_str(&content).map_err(|e| CredentialError::Parse(e.to_string()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn clear(&mut self) {
|
||||||
|
self.cache.clear();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
|
#[serde(tag = "source")]
|
||||||
|
pub enum CredentialConfig {
|
||||||
|
Environment { prefix: String },
|
||||||
|
File { path: PathBuf },
|
||||||
|
Inline(Credential),
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Error)]
|
||||||
|
pub enum CredentialError {
|
||||||
|
#[error("Credential not found: {0}")]
|
||||||
|
NotFound(String),
|
||||||
|
|
||||||
|
#[error("Failed to read credential file: {0}")]
|
||||||
|
FileRead(String),
|
||||||
|
|
||||||
|
#[error("Failed to parse credential: {0}")]
|
||||||
|
Parse(String),
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for CredentialStore {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self::new()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_credential_debug_redacted() {
|
||||||
|
let cred = Credential::Basic {
|
||||||
|
username: "user".to_string(),
|
||||||
|
password: "secret123".to_string(),
|
||||||
|
};
|
||||||
|
|
||||||
|
let debug_output = format!("{:?}", cred);
|
||||||
|
assert!(debug_output.contains("user"));
|
||||||
|
assert!(!debug_output.contains("secret123"));
|
||||||
|
assert!(debug_output.contains("[REDACTED]"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_aws_credential_debug_redacted() {
|
||||||
|
let cred = Credential::AwsKey {
|
||||||
|
access_key_id: "AKIAIOSFODNN7EXAMPLE".to_string(),
|
||||||
|
secret_access_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".to_string(),
|
||||||
|
session_token: None,
|
||||||
|
region: "us-east-1".to_string(),
|
||||||
|
};
|
||||||
|
|
||||||
|
let debug_output = format!("{:?}", cred);
|
||||||
|
assert!(debug_output.contains("AKIA..."));
|
||||||
|
assert!(!debug_output.contains("wJalrXUtnFEMI"));
|
||||||
|
assert!(debug_output.contains("[REDACTED]"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_credential_store_debug() {
|
||||||
|
let mut store = CredentialStore::new();
|
||||||
|
store.cache.insert(
|
||||||
|
"test".to_string(),
|
||||||
|
Credential::Basic {
|
||||||
|
username: "user".to_string(),
|
||||||
|
password: "secret".to_string(),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
let debug_output = format!("{:?}", store);
|
||||||
|
assert!(debug_output.contains("test"));
|
||||||
|
assert!(!debug_output.contains("secret"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_load_from_env() {
|
||||||
|
std::env::set_var("TEST_ORIGIN_USERNAME", "testuser");
|
||||||
|
std::env::set_var("TEST_ORIGIN_PASSWORD", "testpass");
|
||||||
|
|
||||||
|
let store = CredentialStore::new();
|
||||||
|
let cred = store.load_from_env("TEST_ORIGIN").unwrap();
|
||||||
|
|
||||||
|
match cred {
|
||||||
|
Credential::Basic { username, password } => {
|
||||||
|
assert_eq!(username, "testuser");
|
||||||
|
assert_eq!(password, "testpass");
|
||||||
|
}
|
||||||
|
_ => panic!("Expected Basic credential"),
|
||||||
|
}
|
||||||
|
|
||||||
|
std::env::remove_var("TEST_ORIGIN_USERNAME");
|
||||||
|
std::env::remove_var("TEST_ORIGIN_PASSWORD");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -37,6 +37,31 @@ pub enum Error {
|
|||||||
|
|
||||||
#[error("Origin error: {0}")]
|
#[error("Origin error: {0}")]
|
||||||
Origin(String),
|
Origin(String),
|
||||||
|
|
||||||
|
#[error("S3 error: {0}")]
|
||||||
|
S3(String),
|
||||||
|
|
||||||
|
#[error("SFTP error: {0}")]
|
||||||
|
Sftp(String),
|
||||||
|
|
||||||
|
#[error("Operation timed out: {0}")]
|
||||||
|
Timeout(String),
|
||||||
|
|
||||||
|
#[error("Credential error: {0}")]
|
||||||
|
Credential(String),
|
||||||
}
|
}
|
||||||
|
|
||||||
pub type Result<T> = std::result::Result<T, Error>;
|
pub type Result<T> = std::result::Result<T, Error>;
|
||||||
|
|
||||||
|
impl Error {
|
||||||
|
pub fn is_not_found(&self) -> bool {
|
||||||
|
matches!(self, Error::FileNotFound(_))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn downcast_io(&self) -> Option<&std::io::Error> {
|
||||||
|
match self {
|
||||||
|
Error::Io(e) => Some(e),
|
||||||
|
_ => None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,10 +1,12 @@
|
|||||||
pub mod config;
|
pub mod config;
|
||||||
|
pub mod credentials;
|
||||||
pub mod error;
|
pub mod error;
|
||||||
pub mod events;
|
pub mod events;
|
||||||
pub mod resolver;
|
pub mod resolver;
|
||||||
pub mod types;
|
pub mod types;
|
||||||
|
|
||||||
pub use config::{CacheConfig, Config, ConfigError, HealthConfig, OriginConfig, OriginType};
|
pub use config::{CacheConfig, Config, ConfigError, HealthConfig, OriginConfig, OriginType};
|
||||||
|
pub use credentials::{Credential, CredentialConfig, CredentialError, CredentialStore};
|
||||||
pub use error::{Error, Result};
|
pub use error::{Error, Result};
|
||||||
pub use events::{Event, EventBus};
|
pub use events::{Event, EventBus};
|
||||||
pub use resolver::{PathResolver, PathTemplate};
|
pub use resolver::{PathResolver, PathTemplate};
|
||||||
|
|||||||
@@ -3,10 +3,17 @@ name = "musicfs-origins"
|
|||||||
version.workspace = true
|
version.workspace = true
|
||||||
edition.workspace = true
|
edition.workspace = true
|
||||||
|
|
||||||
|
[features]
|
||||||
|
default = []
|
||||||
|
s3 = []
|
||||||
|
sftp = []
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
musicfs-core = { path = "../musicfs-core" }
|
musicfs-core = { path = "../musicfs-core" }
|
||||||
async-trait.workspace = true
|
async-trait.workspace = true
|
||||||
dashmap.workspace = true
|
dashmap.workspace = true
|
||||||
|
libc.workspace = true
|
||||||
|
thiserror.workspace = true
|
||||||
tokio = { workspace = true, features = ["fs", "sync", "time"] }
|
tokio = { workspace = true, features = ["fs", "sync", "time"] }
|
||||||
tracing.workspace = true
|
tracing.workspace = true
|
||||||
|
|
||||||
|
|||||||
@@ -1,14 +1,20 @@
|
|||||||
mod failover;
|
mod failover;
|
||||||
mod health;
|
mod health;
|
||||||
mod local;
|
mod local;
|
||||||
|
mod nfs;
|
||||||
mod registry;
|
mod registry;
|
||||||
mod router;
|
mod router;
|
||||||
|
mod s3;
|
||||||
|
mod sftp;
|
||||||
|
mod smb;
|
||||||
mod traits;
|
mod traits;
|
||||||
|
|
||||||
pub use failover::{FailoverExecutor, RetryConfig};
|
pub use failover::{FailoverExecutor, RetryConfig};
|
||||||
pub use health::{HealthCheckHandle, HealthMonitor, HealthSnapshot, OriginHealthState};
|
pub use health::{HealthCheckHandle, HealthMonitor, HealthSnapshot, OriginHealthState};
|
||||||
pub use local::LocalOrigin;
|
pub use local::LocalOrigin;
|
||||||
|
pub use musicfs_core::OriginType;
|
||||||
|
pub use nfs::NfsOrigin;
|
||||||
pub use registry::OriginRegistry;
|
pub use registry::OriginRegistry;
|
||||||
pub use router::{LatencyStats, Router};
|
pub use router::{LatencyStats, Router};
|
||||||
pub use musicfs_core::OriginType;
|
pub use smb::SmbOrigin;
|
||||||
pub use traits::{Origin, WatchCallback, WatchEvent, WatchHandle};
|
pub use traits::{Origin, WatchCallback, WatchEvent, WatchHandle};
|
||||||
|
|||||||
@@ -0,0 +1,162 @@
|
|||||||
|
use crate::local::LocalOrigin;
|
||||||
|
use crate::traits::{Origin, WatchCallback, WatchHandle};
|
||||||
|
use async_trait::async_trait;
|
||||||
|
use musicfs_core::{DirEntry, FileStat, HealthStatus, OriginId, OriginType, Result};
|
||||||
|
use std::path::{Path, PathBuf};
|
||||||
|
use std::time::Duration;
|
||||||
|
use tokio::time::sleep;
|
||||||
|
use tracing::{debug, warn};
|
||||||
|
|
||||||
|
pub struct NfsOrigin {
|
||||||
|
inner: LocalOrigin,
|
||||||
|
max_retries: u32,
|
||||||
|
display_name: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl NfsOrigin {
|
||||||
|
pub fn new(id: impl Into<OriginId>, mount_point: impl Into<PathBuf>) -> Self {
|
||||||
|
let mount_point = mount_point.into();
|
||||||
|
let display_name = format!("NFS: {}", mount_point.display());
|
||||||
|
|
||||||
|
Self {
|
||||||
|
inner: LocalOrigin::new(id, &mount_point),
|
||||||
|
max_retries: 3,
|
||||||
|
display_name,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn with_max_retries(mut self, retries: u32) -> Self {
|
||||||
|
self.max_retries = retries;
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn retry_on_stale<T, F, Fut>(&self, op: F) -> Result<T>
|
||||||
|
where
|
||||||
|
F: Fn() -> Fut,
|
||||||
|
Fut: std::future::Future<Output = Result<T>>,
|
||||||
|
{
|
||||||
|
let mut delay = Duration::from_millis(100);
|
||||||
|
|
||||||
|
for attempt in 0..self.max_retries {
|
||||||
|
match op().await {
|
||||||
|
Ok(result) => return Ok(result),
|
||||||
|
Err(e) => {
|
||||||
|
if let Some(io_err) = e.downcast_io() {
|
||||||
|
#[cfg(unix)]
|
||||||
|
if io_err.raw_os_error() == Some(libc::ESTALE) {
|
||||||
|
warn!(
|
||||||
|
"NFS stale handle (attempt {}/{}), retrying after {:?}",
|
||||||
|
attempt + 1,
|
||||||
|
self.max_retries,
|
||||||
|
delay
|
||||||
|
);
|
||||||
|
sleep(delay).await;
|
||||||
|
delay *= 2;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return Err(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Err(musicfs_core::Error::NfsStaleHandle)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Origin for NfsOrigin {
|
||||||
|
fn id(&self) -> &OriginId {
|
||||||
|
self.inner.id()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn origin_type(&self) -> OriginType {
|
||||||
|
OriginType::Nfs
|
||||||
|
}
|
||||||
|
|
||||||
|
fn display_name(&self) -> &str {
|
||||||
|
&self.display_name
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn readdir(&self, path: &Path) -> Result<Vec<DirEntry>> {
|
||||||
|
self.retry_on_stale(|| self.inner.readdir(path)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn stat(&self, path: &Path) -> Result<FileStat> {
|
||||||
|
self.retry_on_stale(|| self.inner.stat(path)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read(&self, path: &Path, offset: u64, size: u32) -> Result<Vec<u8>> {
|
||||||
|
self.retry_on_stale(|| self.inner.read(path, offset, size))
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read_full(&self, path: &Path) -> Result<Vec<u8>> {
|
||||||
|
self.retry_on_stale(|| self.inner.read_full(path)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn exists(&self, path: &Path) -> Result<bool> {
|
||||||
|
self.retry_on_stale(|| self.inner.exists(path)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn health(&self) -> HealthStatus {
|
||||||
|
let health_timeout = Duration::from_secs(5);
|
||||||
|
match tokio::time::timeout(health_timeout, self.inner.stat(Path::new("/"))).await {
|
||||||
|
Ok(Ok(_)) => HealthStatus::Healthy,
|
||||||
|
Ok(Err(_)) | Err(_) => HealthStatus::Unhealthy,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn open_read(&self, path: &Path) -> Result<Box<dyn tokio::io::AsyncRead + Send + Unpin>> {
|
||||||
|
self.inner.open_read(path).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn watch(&self, path: &Path, callback: WatchCallback) -> Result<WatchHandle> {
|
||||||
|
debug!("NFS watch - inotify may be unreliable over NFS, consider polling");
|
||||||
|
self.inner.watch(path, callback).await
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use tempfile::TempDir;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_nfs_origin_basic() {
|
||||||
|
let dir = TempDir::new().unwrap();
|
||||||
|
std::fs::write(dir.path().join("test.flac"), b"audio").unwrap();
|
||||||
|
|
||||||
|
let origin = NfsOrigin::new("nfs-test", dir.path());
|
||||||
|
|
||||||
|
let entries = origin.readdir(Path::new("/")).await.unwrap();
|
||||||
|
assert_eq!(entries.len(), 1);
|
||||||
|
|
||||||
|
let data = origin.read(Path::new("/test.flac"), 0, 5).await.unwrap();
|
||||||
|
assert_eq!(&data, b"audio");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_nfs_origin_health() {
|
||||||
|
let dir = TempDir::new().unwrap();
|
||||||
|
let origin = NfsOrigin::new("nfs-test", dir.path());
|
||||||
|
|
||||||
|
assert_eq!(origin.health().await, HealthStatus::Healthy);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_nfs_origin_type() {
|
||||||
|
let dir = TempDir::new().unwrap();
|
||||||
|
let origin = NfsOrigin::new("nfs-test", dir.path());
|
||||||
|
|
||||||
|
assert_eq!(origin.origin_type(), OriginType::Nfs);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_retry_uses_fn_not_fnmut() {
|
||||||
|
fn assert_fn<F: Fn() -> Fut, Fut>(_: F) {}
|
||||||
|
|
||||||
|
let closure = || async { Ok::<_, musicfs_core::Error>(()) };
|
||||||
|
assert_fn(closure);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,51 @@
|
|||||||
|
//! S3-compatible object storage origin
|
||||||
|
//!
|
||||||
|
//! This module is feature-gated behind the `s3` feature to avoid heavy AWS SDK dependencies.
|
||||||
|
//!
|
||||||
|
//! # Oracle Security Fixes (MUST IMPLEMENT)
|
||||||
|
//!
|
||||||
|
//! 1. **Range EOF** - Clamp range to `min(requested_end, file_size)` to avoid 416 errors
|
||||||
|
//! 2. **Health check** - Use `head_bucket` not `list_objects_v2` (lighter operation)
|
||||||
|
//! 3. **Timeout handling** - Wrap all remote calls with `tokio::time::timeout(30s)`
|
||||||
|
//!
|
||||||
|
//! # Example Implementation (when feature enabled)
|
||||||
|
//!
|
||||||
|
//! ```ignore
|
||||||
|
//! async fn read(&self, path: &Path, offset: u64, size: u32) -> Result<Vec<u8>> {
|
||||||
|
//! // Oracle fix: Clamp range to file size to avoid 416 error
|
||||||
|
//! let file_size = self.stat(path).await?.size;
|
||||||
|
//! let end = std::cmp::min(offset + size as u64, file_size).saturating_sub(1);
|
||||||
|
//!
|
||||||
|
//! if offset >= file_size {
|
||||||
|
//! return Ok(Vec::new()); // EOF
|
||||||
|
//! }
|
||||||
|
//!
|
||||||
|
//! let range = format!("bytes={}-{}", offset, end);
|
||||||
|
//!
|
||||||
|
//! // Oracle fix: Add timeout to prevent hung connections
|
||||||
|
//! let resp = tokio::time::timeout(
|
||||||
|
//! Duration::from_secs(30),
|
||||||
|
//! self.client.get_object().bucket(&self.bucket).key(&key).range(range).send()
|
||||||
|
//! )
|
||||||
|
//! .await
|
||||||
|
//! .map_err(|_| Error::Timeout("S3 read timed out".into()))?
|
||||||
|
//! .map_err(|e| Error::S3(e.to_string()))?;
|
||||||
|
//!
|
||||||
|
//! // ...
|
||||||
|
//! }
|
||||||
|
//!
|
||||||
|
//! async fn health(&self) -> HealthStatus {
|
||||||
|
//! // Oracle fix: Use head_bucket instead of list_objects_v2 (lighter)
|
||||||
|
//! match self.client.head_bucket().bucket(&self.bucket).send().await {
|
||||||
|
//! Ok(_) => HealthStatus::Healthy,
|
||||||
|
//! Err(_) => HealthStatus::Unhealthy,
|
||||||
|
//! }
|
||||||
|
//! }
|
||||||
|
//! ```
|
||||||
|
|
||||||
|
#[cfg(feature = "s3")]
|
||||||
|
mod implementation {
|
||||||
|
// Full S3 implementation would go here when aws-sdk-s3 is enabled
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -0,0 +1,12 @@
|
|||||||
|
#![allow(dead_code)]
|
||||||
|
//! SFTP origin - feature-gated to avoid russh/deadpool dependencies
|
||||||
|
|
||||||
|
#[cfg(feature = "sftp")]
|
||||||
|
mod implementation {
|
||||||
|
// Full SFTP implementation with connection pooling
|
||||||
|
// Oracle fixes to implement:
|
||||||
|
// 1. Use deadpool connection pool, not Arc<Mutex<SftpSession>>
|
||||||
|
// 2. Verify SSH host keys against ~/.ssh/known_hosts
|
||||||
|
// 3. Wrap all operations with tokio::time::timeout(30s)
|
||||||
|
// 4. Cap open_read to actual file size, not u32::MAX
|
||||||
|
}
|
||||||
@@ -0,0 +1,154 @@
|
|||||||
|
use crate::local::LocalOrigin;
|
||||||
|
use crate::traits::{Origin, WatchCallback, WatchHandle};
|
||||||
|
use async_trait::async_trait;
|
||||||
|
use musicfs_core::{DirEntry, FileStat, HealthStatus, OriginId, OriginType, Result};
|
||||||
|
use std::future::Future;
|
||||||
|
use std::path::{Path, PathBuf};
|
||||||
|
use tracing::{debug, warn};
|
||||||
|
|
||||||
|
pub struct SmbOrigin {
|
||||||
|
inner: LocalOrigin,
|
||||||
|
share_path: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SmbOrigin {
|
||||||
|
pub fn from_mount(
|
||||||
|
id: impl Into<OriginId>,
|
||||||
|
mount_point: impl Into<PathBuf>,
|
||||||
|
share_path: impl Into<String>,
|
||||||
|
) -> Self {
|
||||||
|
let mount_point = mount_point.into();
|
||||||
|
let share_path = share_path.into();
|
||||||
|
|
||||||
|
Self {
|
||||||
|
inner: LocalOrigin::new(id, &mount_point),
|
||||||
|
share_path,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn is_mounted(&self) -> bool {
|
||||||
|
self.inner.exists(Path::new("/")).await.unwrap_or(false)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn retry_on_disconnect<T, F, Fut>(&self, op: F) -> Result<T>
|
||||||
|
where
|
||||||
|
F: Fn() -> Fut,
|
||||||
|
Fut: Future<Output = Result<T>>,
|
||||||
|
{
|
||||||
|
const MAX_RETRIES: u32 = 3;
|
||||||
|
|
||||||
|
for attempt in 0..MAX_RETRIES {
|
||||||
|
match op().await {
|
||||||
|
Ok(val) => return Ok(val),
|
||||||
|
Err(e) => {
|
||||||
|
if Self::is_enotconn(&e) && attempt < MAX_RETRIES - 1 {
|
||||||
|
debug!(attempt, "SMB ENOTCONN, retrying");
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
return Err(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
unreachable!()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(unix)]
|
||||||
|
fn is_enotconn(err: &musicfs_core::Error) -> bool {
|
||||||
|
if let musicfs_core::Error::Io(io_err) = err {
|
||||||
|
io_err.raw_os_error() == Some(libc::ENOTCONN)
|
||||||
|
} else {
|
||||||
|
false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(not(unix))]
|
||||||
|
fn is_enotconn(_err: &musicfs_core::Error) -> bool {
|
||||||
|
false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Origin for SmbOrigin {
|
||||||
|
fn id(&self) -> &OriginId {
|
||||||
|
self.inner.id()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn origin_type(&self) -> OriginType {
|
||||||
|
OriginType::Smb
|
||||||
|
}
|
||||||
|
|
||||||
|
fn display_name(&self) -> &str {
|
||||||
|
&self.share_path
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn readdir(&self, path: &Path) -> Result<Vec<DirEntry>> {
|
||||||
|
self.retry_on_disconnect(|| self.inner.readdir(path)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn stat(&self, path: &Path) -> Result<FileStat> {
|
||||||
|
self.retry_on_disconnect(|| self.inner.stat(path)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read(&self, path: &Path, offset: u64, size: u32) -> Result<Vec<u8>> {
|
||||||
|
self.retry_on_disconnect(|| self.inner.read(path, offset, size)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read_full(&self, path: &Path) -> Result<Vec<u8>> {
|
||||||
|
self.retry_on_disconnect(|| self.inner.read_full(path)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn exists(&self, path: &Path) -> Result<bool> {
|
||||||
|
self.retry_on_disconnect(|| self.inner.exists(path)).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn health(&self) -> HealthStatus {
|
||||||
|
let health_timeout = std::time::Duration::from_secs(5);
|
||||||
|
match tokio::time::timeout(health_timeout, self.is_mounted()).await {
|
||||||
|
Ok(true) => HealthStatus::Healthy,
|
||||||
|
Ok(false) | Err(_) => HealthStatus::Unhealthy,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn open_read(&self, path: &Path) -> Result<Box<dyn tokio::io::AsyncRead + Send + Unpin>> {
|
||||||
|
self.inner.open_read(path).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn watch(&self, path: &Path, callback: WatchCallback) -> Result<WatchHandle> {
|
||||||
|
warn!("SMB watch using inotify - may be unreliable. Consider polling for remote mounts.");
|
||||||
|
self.inner.watch(path, callback).await
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use tempfile::TempDir;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_smb_origin_basic() {
|
||||||
|
let dir = TempDir::new().unwrap();
|
||||||
|
std::fs::write(dir.path().join("test.flac"), b"audio").unwrap();
|
||||||
|
|
||||||
|
let origin = SmbOrigin::from_mount("smb-test", dir.path(), "//server/share");
|
||||||
|
|
||||||
|
let entries = origin.readdir(Path::new("/")).await.unwrap();
|
||||||
|
assert_eq!(entries.len(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_smb_origin_type() {
|
||||||
|
let dir = TempDir::new().unwrap();
|
||||||
|
let origin = SmbOrigin::from_mount("smb-test", dir.path(), "//server/share");
|
||||||
|
|
||||||
|
assert_eq!(origin.origin_type(), OriginType::Smb);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_smb_display_name() {
|
||||||
|
let dir = TempDir::new().unwrap();
|
||||||
|
let origin = SmbOrigin::from_mount("smb-test", dir.path(), "//server/music");
|
||||||
|
|
||||||
|
assert_eq!(origin.display_name(), "//server/music");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,5 +1,5 @@
|
|||||||
use crate::cdc::CdcChunker;
|
use crate::cdc::CdcChunker;
|
||||||
use musicfs_core::{ChunkHash, FileId, FileMeta, OriginId, RealPath, VirtualPath};
|
use musicfs_core::{ChunkHash, FileId, FileMeta, OriginId};
|
||||||
use musicfs_origins::Origin;
|
use musicfs_origins::Origin;
|
||||||
use std::collections::{HashMap, HashSet};
|
use std::collections::{HashMap, HashSet};
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
@@ -224,7 +224,7 @@ pub enum DeltaError {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
use musicfs_core::OriginId;
|
use musicfs_core::{OriginId, RealPath, VirtualPath};
|
||||||
use std::time::SystemTime;
|
use std::time::SystemTime;
|
||||||
|
|
||||||
fn make_file_meta(id: i64, path: &str, size: u64) -> FileMeta {
|
fn make_file_meta(id: i64, path: &str, size: u64) -> FileMeta {
|
||||||
|
|||||||
Reference in New Issue
Block a user