diff --git a/Cargo.lock b/Cargo.lock index 5a47d51..f304790 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -675,6 +675,15 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" +[[package]] +name = "crossbeam-channel" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.12" @@ -1632,6 +1641,15 @@ dependencies = [ "winapi", ] +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "matchit" version = "0.8.4" @@ -1688,6 +1706,7 @@ dependencies = [ "anyhow", "async-trait", "bytes", + "chrono", "clap", "fuser", "http", @@ -1702,6 +1721,9 @@ dependencies = [ "tokio-stream", "tonic", "tonic-prost", + "tracing", + "tracing-appender", + "tracing-subscriber", "twox-hash", ] @@ -1758,6 +1780,15 @@ dependencies = [ "bitflags 2.13.0", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num-bigint" version = "0.4.6" @@ -2578,6 +2609,15 @@ dependencies = [ "digest 0.11.3", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "2.0.1" @@ -2854,6 +2894,12 @@ version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +[[package]] +name = "symlink" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" + [[package]] name = "symphonia" version = "0.5.5" @@ -3065,6 +3111,15 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "time" version = "0.3.49" @@ -3293,6 +3348,19 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-appender" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" +dependencies = [ + "crossbeam-channel", + "symlink", + "thiserror", + "time", + "tracing-subscriber", +] + [[package]] name = "tracing-attributes" version = "0.1.31" @@ -3311,6 +3379,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", ] [[package]] @@ -3402,6 +3500,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "vcpkg" version = "0.2.15" diff --git a/Cargo.toml b/Cargo.toml index 57baef5..2f60ca2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,9 @@ name = "musicfs" version = "0.1.0" edition = "2024" +# `cargo run` (and devenv's `cargo run -- ...`) picks the FUSE binary, not +# musicfs-server, when two [[bin]] targets exist. +default-run = "musicfs" [lib] name = "musicfs" @@ -37,6 +40,14 @@ tonic-prost = "0.14" prost = "0.14" tokio-stream = "0.1" http = "1" +chrono = { version = "0.4", default-features = false, features = ["clock"] } + +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } +# Pinned to 0.2.5: Builder::max_log_files (used for retention) landed in the +# 0.2.3 Builder API and is current here. An unqualified "0.2" could otherwise +# resolve to an older patch without the retention method. +tracing-appender = "0.2.5" [dev-dependencies] tempfile = "3" diff --git a/db/schema.sql b/db/schema.sql index c1b4290..87bcb6f 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -37,3 +37,11 @@ CREATE TABLE music_metadata_pictures ( data_length BIGINT NOT NULL, PRIMARY KEY (inode, position) ); + +-- Lazy byte cache for NetworkOrigin. Rows exist only for files the user has +-- actually read; reconcile invalidates a row when its hash changes. +CREATE TABLE cached_file_bytes ( + inode BIGINT PRIMARY KEY REFERENCES items(inode) ON DELETE CASCADE, + data BYTEA NOT NULL, + fetched_at TIMESTAMPTZ NOT NULL DEFAULT now() +); diff --git a/proto/buf.yaml b/proto/buf.yaml index f74da98..ccd7070 100644 --- a/proto/buf.yaml +++ b/proto/buf.yaml @@ -4,6 +4,9 @@ modules: lint: use: - STANDARD + except: + - RPC_RESPONSE_STANDARD_NAME + - RPC_REQUEST_RESPONSE_UNIQUE breaking: use: - FILE diff --git a/proto/musicfs.proto b/proto/musicfs.proto index 7e0f7f1..627f9fa 100644 --- a/proto/musicfs.proto +++ b/proto/musicfs.proto @@ -3,17 +3,27 @@ syntax = "proto3"; package musicfs; // MusicFs service exposes a server-side music library to a remote musicfs -// FUSE mount. Three RPCs, all server-streaming so the client can pull -// incrementally and the server can keep large bodies off the heap. +// FUSE mount. service MusicFs { - // Stream every file entry in the library. The client diffs each entry's - // hash against its DB to decide what changed. + // Hash-based reconciliation. Client sends the (inode, hash) pairs it + // currently has; server responds with the subset that changed (new or + // hash-different) plus the inodes the client has but the server no longer + // has. Cheap on the wire — no file metadata crosses until the client asks. + rpc Reconcile(ReconcileRequest) returns (ReconcileResponse); + + // Fetch full metadata for a specific set of inodes. Called after Reconcile + // returns a non-empty `changed` list, for just those inodes. + rpc GetMetadata(GetMetadataRequest) returns (stream ManifestEntry); + + // Stream every file entry in the library. Convenience RPC for first-run + // bootstrapping or full-refresh. The diff-based Reconcile path is the + // normal case. rpc GetManifest(GetManifestRequest) returns (stream ManifestEntry); - // Stream the bytes of one file. If `start` and `length` are both set on - // the request, only that byte range is sent; otherwise the whole file. - // The first chunk carries `total_size` and `offset`; subsequent chunks - // continue from there. + // Stream the bytes of one file. If `start` and `length` are both zero the + // server streams the entire file; otherwise just the requested range. The + // first chunk carries `total_size` and `offset`; subsequent chunks continue + // from there. rpc GetFile(GetFileRequest) returns (stream FileChunk); // Long-lived stream of filesystem change events. The client uses these as @@ -22,6 +32,24 @@ service MusicFs { rpc SubscribeEvents(SubscribeEventsRequest) returns (stream ChangeEvent); } +message ReconcileRequest { + repeated InodeHash entries = 1; +} + +message ReconcileResponse { + repeated InodeHash changed = 1; + repeated uint64 deleted = 2; +} + +message InodeHash { + uint64 inode = 1; + uint64 hash = 2; +} + +message GetMetadataRequest { + repeated uint64 inodes = 1; +} + message GetManifestRequest {} message GetFileRequest { diff --git a/src/bin/musicfs-server.rs b/src/bin/musicfs-server.rs index e313b0f..602719b 100644 --- a/src/bin/musicfs-server.rs +++ b/src/bin/musicfs-server.rs @@ -2,12 +2,14 @@ use std::{net::SocketAddr, path::PathBuf}; use anyhow::{Context, Result}; use clap::Parser; +use musicfs::logging::{LogConfig, init}; use musicfs::server::{ state::ServerState, transport::{self, TransportArgs}, watcher::ServerWatcher, }; use tokio::signal::unix::{SignalKind, signal}; +use tracing::{error, info}; #[derive(Parser, Debug)] #[command(version, about, long_about = None)] @@ -23,14 +25,36 @@ struct Args { /// Name of the transport implementation to use. Today only "grpc". #[arg(short, long, default_value = "grpc")] transport: String, + + /// Directory for daily-rotated log files. + #[arg(long, default_value = "./logs")] + log_dir: PathBuf, } #[tokio::main] async fn main() -> Result<()> { let args = Args::parse(); + // Bind the guard for the whole process so the non-blocking file writer + // flushes on exit. Initialized before anything else so even early + // failures land in the log. + let _guard = init(LogConfig { + log_dir: args.log_dir.clone(), + file_prefix: "musicfs-server".to_string(), + max_files: 7, + }); + let source = args.source.clone(); + info!( + source = %source.display(), + transport = %args.transport, + listen = %args.listen, + log_dir = %args.log_dir.display(), + "musicfs-server starting" + ); + if !source.is_dir() { + error!(source = %source.display(), "source is not a readable directory"); return Err(anyhow::anyhow!( "source is not a readable directory: {}", source.display() @@ -41,6 +65,7 @@ async fn main() -> Result<()> { state .replace_all(&source) .with_context(|| format!("initial scan of {}", source.display()))?; + info!(entries = state.manifest().len(), "initial scan complete"); let (_watcher, events_rx) = ServerWatcher::spawn(source.clone(), state.clone()); @@ -52,14 +77,15 @@ async fn main() -> Result<()> { ) .with_context(|| format!("building transport {:?}", args.transport))?; - println!( - "musicfs-server: transport={}, listen={}, initial scan complete", - args.transport, args.listen + info!( + transport = %args.transport, + listen = %args.listen, + "transport ready" ); let server_task = tokio::spawn(async move { if let Err(e) = transport.run(state, events_rx).await { - eprintln!("transport ended with error: {e}"); + error!(error = %e, "transport ended with error"); } }); @@ -67,9 +93,9 @@ async fn main() -> Result<()> { let mut sigterm = signal(SignalKind::terminate()).expect("register SIGTERM"); tokio::select! { - _ = sigint.recv() => println!("received SIGINT, shutting down"), - _ = sigterm.recv() => println!("received SIGTERM, shutting down"), - _ = server_task => println!("transport task exited"), + _ = sigint.recv() => info!("received SIGINT, shutting down"), + _ = sigterm.recv() => info!("received SIGTERM, shutting down"), + _ = server_task => info!("transport task exited"), } return Ok(()); diff --git a/src/db/cache.rs b/src/db/cache.rs new file mode 100644 index 0000000..1650323 --- /dev/null +++ b/src/db/cache.rs @@ -0,0 +1,57 @@ +use sea_orm::{ + ActiveValue::Set, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, + sea_query::OnConflict, +}; +use tracing::{debug, error}; + +use crate::db::entities::cached_file_bytes::{ActiveModel, Column, Entity, Model}; + +/// Return the cached bytes for `inode` if present. `None` means "not cached"; +/// the caller should fetch from the server and call [`put_cached_bytes`]. +pub async fn get_cached_bytes(inode: i64, db: &DatabaseConnection) -> Option> { + let row: Option = match Entity::find_by_id(inode).one(db).await { + Ok(m) => m, + Err(e) => { + error!(%inode, error = %e, "get_cached_bytes: DB read failed"); + None + } + }; + return row.map(|m| m.data); +} + +/// Insert or replace the cached bytes for `inode`. Called after a successful +/// `GetFile` round-trip so subsequent reads of the same range hit the cache. +pub async fn put_cached_bytes(inode: i64, data: Vec, db: &DatabaseConnection) { + let active = ActiveModel { + inode: Set(inode), + data: Set(data), + fetched_at: Set(chrono::Utc::now().naive_utc()), + }; + if let Err(e) = Entity::insert(active) + .on_conflict( + OnConflict::column(Column::Inode) + .update_columns([Column::Data, Column::FetchedAt]) + .to_owned(), + ) + .exec(db) + .await + { + error!(%inode, error = %e, "put_cached_bytes: DB upsert failed"); + } +} + +/// Remove cache rows for the given inodes. Called by reconcile when a file's +/// hash has changed (the cached bytes are now stale). +pub async fn delete_cached_bytes_for(inodes: &[i64], db: &DatabaseConnection) { + if inodes.is_empty() { + return; + } + debug!(count = inodes.len(), "deleting cached bytes"); + if let Err(e) = Entity::delete_many() + .filter(Column::Inode.is_in(inodes.to_vec())) + .exec(db) + .await + { + error!(error = %e, "delete_cached_bytes_for: DB delete failed"); + } +} diff --git a/src/db/entities.rs b/src/db/entities.rs index c2c0d3a..c313144 100644 --- a/src/db/entities.rs +++ b/src/db/entities.rs @@ -36,6 +36,25 @@ impl From<&Item> for ActiveModel { } } +pub mod cached_file_bytes { + use sea_orm::entity::prelude::*; + + #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] + #[sea_orm(table_name = "cached_file_bytes")] + pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub inode: i64, + #[sea_orm(column_type = "Blob")] + pub data: Vec, + pub fetched_at: DateTime, + } + + #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] + pub enum Relation {} + + impl ActiveModelBehavior for ActiveModel {} +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/db/mod.rs b/src/db/mod.rs index bce2a87..4db22c7 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -1,2 +1,3 @@ +pub mod cache; pub mod entities; pub mod sync; diff --git a/src/db/sync.rs b/src/db/sync.rs index eaf35a1..e4ec285 100644 --- a/src/db/sync.rs +++ b/src/db/sync.rs @@ -2,6 +2,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use fuser::INodeNo; use sea_orm::entity::prelude::*; +use tracing::{debug, error}; use crate::db::entities::{ActiveModel, Entity, Model}; use crate::item::Item; @@ -41,14 +42,28 @@ pub async fn sync_items_to_db( to_delete.push(*ino); } + debug!( + insert = to_insert.len(), + update = to_update.len(), + delete = to_delete.len(), + save_music = to_save_music.len(), + "sync_items_to_db" + ); + if !to_insert.is_empty() { - Entity::insert_many(to_insert).exec(client).await.unwrap(); + if let Err(e) = Entity::insert_many(to_insert).exec(client).await { + error!(error = %e, "sync: insert_many failed"); + } } for model in to_update { - model.update(client).await.unwrap(); + if let Err(e) = model.update(client).await { + error!(error = %e, "sync: update failed"); + } } for ino in to_delete { - Entity::delete_by_id(ino).exec(client).await.unwrap(); + if let Err(e) = Entity::delete_by_id(ino).exec(client).await { + error!(%ino, error = %e, "sync: delete failed"); + } } for ino_i64 in &to_save_music { @@ -57,9 +72,9 @@ pub async fn sync_items_to_db( .get(&ino) .and_then(|item| item.music_metadata.as_ref()) { - save_music_metadata(*ino_i64, music_metadata, client) - .await - .unwrap(); + if let Err(e) = save_music_metadata(*ino_i64, music_metadata, client).await { + error!(ino = *ino_i64, error = %e, "sync: save_music_metadata failed"); + } } } } @@ -69,6 +84,9 @@ pub fn run_db_blocking(future: F) -> F::Output { .enable_io() .enable_time() .build() - .unwrap() + .unwrap_or_else(|e| { + error!(error = %e, "run_db_blocking: failed to build runtime"); + panic!("run_db_blocking: failed to build runtime: {e}"); + }) .block_on(future); } diff --git a/src/item.rs b/src/item.rs index 030e3da..281cbcb 100644 --- a/src/item.rs +++ b/src/item.rs @@ -38,6 +38,27 @@ fn secs_since_epoch(time: SystemTime) -> u64 { .unwrap_or(0); } +/// Hash inputs shared by every Item regardless of origin. Server manifests +/// and client Items both feed these same five values through xxhash, so a +/// file's hash is identical on both sides — that's what makes `Reconcile`'s +/// diff correct. +pub fn compute_item_hash( + inode: u64, + original_path: &[u8], + ctime_secs: u64, + mtime_secs: u64, + crtime_secs: u64, +) -> u64 { + let seed = 1234; + let mut hasher = XxHash64::with_seed(seed); + hasher.write_u64(inode); + hasher.write(original_path); + hasher.write_u64(ctime_secs); + hasher.write_u64(mtime_secs); + hasher.write_u64(crtime_secs); + return hasher.finish(); +} + impl Item { #[allow(clippy::too_many_arguments)] pub fn new( @@ -67,16 +88,13 @@ impl Item { } pub fn compute_hash(&self) -> u64 { - let seed = 1234; - let mut hasher = XxHash64::with_seed(seed); - - hasher.write_u64(self.inode.0); - hasher.write(self.original_path.as_os_str().as_bytes()); - hasher.write_u64(secs_since_epoch(self.attrs.ctime)); - hasher.write_u64(secs_since_epoch(self.attrs.mtime)); - hasher.write_u64(secs_since_epoch(self.attrs.crtime)); - - return hasher.finish(); + return compute_item_hash( + self.inode.0, + self.original_path.as_os_str().as_bytes(), + secs_since_epoch(self.attrs.ctime), + secs_since_epoch(self.attrs.mtime), + secs_since_epoch(self.attrs.crtime), + ); } } diff --git a/src/lib.rs b/src/lib.rs index 7ca09f7..efd3009 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,6 @@ pub mod db; pub mod item; +pub mod logging; pub mod music; pub mod origins; pub mod proto; diff --git a/src/logging.rs b/src/logging.rs new file mode 100644 index 0000000..7706e4b --- /dev/null +++ b/src/logging.rs @@ -0,0 +1,89 @@ +//! Process-wide logging initialization. +//! +//! Wires three `fmt` layers under a single global [`EnvFilter`] (driven by +//! `RUST_LOG`, defaulting to `info`): +//! +//! - **stdout** — `INFO`/`DEBUG`/`TRACE` (the verbose side) +//! - **stderr** — `WARN`/`ERROR` (the severe side) +//! - **file** — everything passing the global filter, daily-rotated with +//! retention, written without ANSI colors so log files stay clean. +//! +//! [`init`] returns the [`WorkerGuard`] that owns the non-blocking file +//! buffer; the caller binds it for the whole process lifetime so the buffer +//! flushes on exit. Dropping it early would silently drop pending log lines. +//! +//! The FUSE/RPC hot paths carry `trace!`/`debug!` points that fire per syscall +//! — thousands per second under load. They are inert unless +//! `RUST_LOG=trace`/`debug` is set; do not enable those levels in production +//! casually. + +use std::path::PathBuf; + +use tracing::Level; +use tracing_subscriber::{EnvFilter, filter::filter_fn, fmt, prelude::*}; + +/// Configuration handed to [`init`] by each binary. +pub struct LogConfig { + /// Directory the daily-rotated log files are written into. + pub log_dir: PathBuf, + /// Filename prefix; conventionally the binary name (`musicfs` or + /// `musicfs-server`). + pub file_prefix: String, + /// Retention: keep at most this many rotated log files. + pub max_files: usize, +} + +/// Initialize the global tracing subscriber and return the file-writer +/// [`WorkerGuard`]. Bind it for the process lifetime so pending file writes +/// flush on exit. +/// +/// Panics if the log directory cannot be created or the rolling file appender +/// cannot be initialized — both are fatal startup conditions worth failing +/// fast on, before any real work begins. +pub fn init(cfg: LogConfig) -> tracing_appender::non_blocking::WorkerGuard { + std::fs::create_dir_all(&cfg.log_dir).unwrap_or_else(|e| { + panic!( + "logging: failed to create log dir {}: {e}", + cfg.log_dir.display() + ) + }); + + // One global gate for all three layers. RUST_LOG wins; default `info`. + let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); + + let file_appender = tracing_appender::rolling::RollingFileAppender::builder() + .rotation(tracing_appender::rolling::Rotation::DAILY) + .filename_prefix(&cfg.file_prefix) + .filename_suffix("log") + .max_log_files(cfg.max_files) + .build(&cfg.log_dir) + .unwrap_or_else(|e| { + panic!( + "logging: failed to init rolling file appender in {}: {e}", + cfg.log_dir.display() + ) + }); + let (file_writer, guard) = tracing_appender::non_blocking(file_appender); + + // tracing's Level ordering is ERROR < WARN < INFO < DEBUG < TRACE, so + // `>= INFO` selects the verbose side (INFO/DEBUG/TRACE) routed to stdout, + // and `<= WARN` selects the severe side (WARN/ERROR) routed to stderr. + let stdout_layer = fmt::layer() + .with_writer(std::io::stdout) + .with_filter(filter_fn(|m| *m.level() >= Level::INFO)); + + let stderr_layer = fmt::layer() + .with_writer(std::io::stderr) + .with_filter(filter_fn(|m| *m.level() <= Level::WARN)); + + let file_layer = fmt::layer().with_ansi(false).with_writer(file_writer); + + tracing_subscriber::registry() + .with(env_filter) + .with(stdout_layer) + .with(stderr_layer) + .with(file_layer) + .init(); + + guard +} diff --git a/src/main.rs b/src/main.rs index 696901c..ba047a2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,14 +1,19 @@ -use std::{collections::HashMap, path::Path}; +use std::{collections::HashMap, path::Path, sync::Arc}; use clap::Parser; +use fuser::INodeNo; use musicfs::db::entities::{Entity, Model}; use musicfs::db::sync::sync_items_to_db; +use musicfs::item::Item; +use musicfs::logging::{LogConfig, init}; use musicfs::music::db::restore_music_metadata_from_db; use musicfs::origins::local::LocalOrigin; +use musicfs::origins::network::NetworkOrigin; use musicfs::origins::{FuseFs, Origin}; use musicfs::virtual_dirs::restore_virtual_paths; use sea_orm::entity::prelude::*; use tokio::signal::unix::{SignalKind, signal}; +use tracing::{debug, error, info}; #[derive(Parser, Debug)] #[command(version, about, long_about = None)] @@ -16,59 +21,123 @@ struct Args { #[arg(short, long, required = true)] mountpoint: String, + /// Local directory path (→ LocalOrigin) OR `http://host:port` URL of a + /// musicfs-server (→ NetworkOrigin). #[arg(short, long, required = true)] source: String, #[arg(short, long, required = true)] database: String, + + /// Directory for daily-rotated log files. + #[arg(long, default_value = "./logs")] + log_dir: std::path::PathBuf, } #[tokio::main] async fn main() { let args = Args::parse(); - let mountpoint = args.mountpoint; - let db = sea_orm::Database::connect(&args.database).await.unwrap(); + // Bind the guard for the whole process so the non-blocking file writer + // flushes on exit. Initialized before anything else so even early + // failures land in the log. + let _guard = init(LogConfig { + log_dir: args.log_dir.clone(), + file_prefix: "musicfs".to_string(), + max_files: 7, + }); - let origin = LocalOrigin::new(args.source.clone(), mountpoint.clone()); + let mountpoint = args.mountpoint.clone(); + info!( + mountpoint = %mountpoint, + source = %args.source, + database = %args.database, + log_dir = %args.log_dir.display(), + "musicfs starting" + ); - let mut snapshot = origin.snapshot().unwrap(); - - let db_items: HashMap = Entity::find() - .all(&db) + let db = sea_orm::Database::connect(&args.database) .await - .unwrap() - .into_iter() - .map(|e| (e.inode, e)) - .collect(); + .unwrap_or_else(|e| { + error!(database = %args.database, error = %e, "database connect failed"); + panic!("database connect: {e}"); + }); + info!(database = %args.database, "database connected"); - sync_items_to_db(&snapshot, &db_items, &db).await; - restore_music_metadata_from_db(&mut snapshot, &db_items, &db).await; - restore_virtual_paths(&mut snapshot, &db_items, Path::new(&args.source)); + let (snapshot, byte_source, watcher) = if looks_like_url(&args.source) { + debug!(source = %args.source, "using NetworkOrigin"); + let origin = NetworkOrigin::new(args.source.clone(), mountpoint.clone(), db.clone()) + .unwrap_or_else(|e| { + error!(source = %args.source, error = %e, "network origin init failed"); + panic!("network origin init: {e}"); + }); + let snapshot = origin.snapshot().unwrap_or_else(|e| { + error!(source = %args.source, error = %e, "network initial snapshot failed"); + panic!("network initial snapshot: {e}"); + }); + info!(files = snapshot.len(), "network snapshot complete"); + (snapshot, origin.byte_source(), origin.watcher()) + } else { + debug!(source = %args.source, "using LocalOrigin"); + let origin = LocalOrigin::new(args.source.clone(), mountpoint.clone()); + let mut snapshot = origin.snapshot().unwrap_or_else(|e| { + error!(source = %args.source, error = %e, "local initial snapshot failed"); + panic!("local initial snapshot: {e}"); + }); - let files = std::sync::Arc::new(std::sync::Mutex::new(snapshot)); - let bytes = origin.byte_source(); - let watcher = origin.watcher(); + let db_items: HashMap = Entity::find() + .all(&db) + .await + .unwrap_or_else(|e| { + error!(error = %e, "loading db items failed"); + panic!("loading db items: {e}"); + }) + .into_iter() + .map(|e| (e.inode, e)) + .collect(); + sync_items_to_db(&snapshot, &db_items, &db).await; + restore_music_metadata_from_db(&mut snapshot, &db_items, &db).await; + restore_virtual_paths(&mut snapshot, &db_items, Path::new(&args.source)); + + info!(files = snapshot.len(), "local snapshot complete"); + (snapshot, origin.byte_source(), origin.watcher()) + }; + + let files: Arc>> = + Arc::new(std::sync::Mutex::new(snapshot)); watcher.watch(files.clone()); let fs = FuseFs { files, - bytes, + bytes: byte_source, client: db, }; let cfg = fuser::Config::default(); - let session = - fuser::spawn_mount2(fs, &mountpoint, &cfg).expect("Failed to mount FUSE filesystem"); + let session = fuser::spawn_mount2(fs, &mountpoint, &cfg).unwrap_or_else(|e| { + error!(mountpoint = %mountpoint, error = %e, "failed to mount FUSE filesystem"); + panic!("failed to mount FUSE filesystem: {e}"); + }); + info!(mountpoint = %mountpoint, "FUSE mounted"); - let mut sigint = signal(SignalKind::interrupt()).expect("Failed to register SIGINT handler"); - let mut sigterm = signal(SignalKind::terminate()).expect("Failed to register SIGTERM handler"); + let mut sigint = signal(SignalKind::interrupt()).unwrap_or_else(|e| { + error!(error = %e, "failed to register SIGINT handler"); + panic!("register SIGINT handler: {e}"); + }); + let mut sigterm = signal(SignalKind::terminate()).unwrap_or_else(|e| { + error!(error = %e, "failed to register SIGTERM handler"); + panic!("register SIGTERM handler: {e}"); + }); tokio::select! { - _ = sigint.recv() => println!("Received SIGINT, shutting down"), - _ = sigterm.recv() => println!("Received SIGTERM, shutting down"), + _ = sigint.recv() => info!("received SIGINT, shutting down"), + _ = sigterm.recv() => info!("received SIGTERM, shutting down"), } - println!("Unmounting"); + info!(mountpoint = %mountpoint, "unmounting"); drop(session); } + +fn looks_like_url(s: &str) -> bool { + return s.starts_with("http://") || s.starts_with("https://"); +} diff --git a/src/music/db.rs b/src/music/db.rs index e3c7a2b..7668b56 100644 --- a/src/music/db.rs +++ b/src/music/db.rs @@ -9,6 +9,7 @@ use crate::music::metadata::MusicMetadata; use crate::music::metadata::db::{ artists, music_metadata as music_metadata_entity, other_tags, pictures, }; +use tracing::{debug, error}; pub async fn save_music_metadata( inode: i64, @@ -100,22 +101,32 @@ pub async fn restore_music_metadata_from_db( return; } - let mm_rows: HashMap = music_metadata_entity::Entity::find() - .filter(MmCol::Inode.is_in(unchanged_music_inodes.clone())) - .all(client) - .await - .unwrap() - .into_iter() - .map(|m| (m.inode, m)) - .collect(); + let mm_rows: HashMap = + match music_metadata_entity::Entity::find() + .filter(MmCol::Inode.is_in(unchanged_music_inodes.clone())) + .all(client) + .await + { + Ok(rows) => rows.into_iter().map(|m| (m.inode, m)).collect(), + Err(e) => { + error!(error = %e, "restore_music_metadata: mm rows query failed"); + return; + } + }; let mut artists_by_inode: HashMap> = HashMap::new(); - for row in artists::Entity::find() + let artist_rows = match artists::Entity::find() .filter(ArtCol::Inode.is_in(unchanged_music_inodes.clone())) .all(client) .await - .unwrap() { + Ok(r) => r, + Err(e) => { + error!(error = %e, "restore_music_metadata: artists query failed"); + vec![] + } + }; + for row in artist_rows { artists_by_inode .entry(row.inode) .or_default() @@ -123,12 +134,18 @@ pub async fn restore_music_metadata_from_db( } let mut other_tags_by_inode: HashMap> = HashMap::new(); - for row in other_tags::Entity::find() + let other_tag_rows = match other_tags::Entity::find() .filter(OtCol::Inode.is_in(unchanged_music_inodes.clone())) .all(client) .await - .unwrap() { + Ok(r) => r, + Err(e) => { + error!(error = %e, "restore_music_metadata: other_tags query failed"); + vec![] + } + }; + for row in other_tag_rows { other_tags_by_inode .entry(row.inode) .or_default() @@ -136,15 +153,23 @@ pub async fn restore_music_metadata_from_db( } let mut pictures_by_inode: HashMap> = HashMap::new(); - for row in pictures::Entity::find() + let picture_rows = match pictures::Entity::find() .filter(PicCol::Inode.is_in(unchanged_music_inodes)) .all(client) .await - .unwrap() { + Ok(r) => r, + Err(e) => { + error!(error = %e, "restore_music_metadata: pictures query failed"); + vec![] + } + }; + for row in picture_rows { pictures_by_inode.entry(row.inode).or_default().push(row); } + debug!(restored = mm_rows.len(), "restored music metadata from db"); + for (inode, mm_row) in mm_rows { let ino = INodeNo(inode as u64); let Some(item) = snapshot.get_mut(&ino) else { diff --git a/src/music/flac.rs b/src/music/flac.rs index 06c4421..4e10879 100644 --- a/src/music/flac.rs +++ b/src/music/flac.rs @@ -11,6 +11,7 @@ use symphonia::core::{ use crate::music::encoder::MusicMetadataEncoder; use crate::music::metadata::{MusicMetadata, extract_standard_tags}; use crate::music::parser::MusicMetadataParser; +use tracing::warn; const BLOCK_PADDING: u8 = 1; const BLOCK_VORBIS_COMMENT: u8 = 4; @@ -29,9 +30,9 @@ impl MusicMetadataParser for FlacMusicMetadataParser { match parse_flac_metadata(path) { Some(mm) => Some(mm), None => { - eprintln!( - "musicfs: failed to parse FLAC metadata from {}", - path.display() + warn!( + path = %path.display(), + "failed to parse FLAC metadata; serving as passthrough" ); None } diff --git a/src/music/mp3.rs b/src/music/mp3.rs index 83e97ed..2d4ed30 100644 --- a/src/music/mp3.rs +++ b/src/music/mp3.rs @@ -11,6 +11,7 @@ use symphonia::core::{ use crate::music::encoder::MusicMetadataEncoder; use crate::music::metadata::{MusicMetadata, extract_standard_tags}; use crate::music::parser::MusicMetadataParser; +use tracing::warn; /// Frames we replace from our own tag fields. Every other frame in the source /// ID3 tag is preserved verbatim via externalization. @@ -25,9 +26,9 @@ impl MusicMetadataParser for Mp3MusicMetadataParser { match parse_mp3_metadata(path) { Some(mm) => Some(mm), None => { - eprintln!( - "musicfs: failed to parse MP3 metadata from {}", - path.display() + warn!( + path = %path.display(), + "failed to parse MP3 metadata; serving as passthrough" ); None } diff --git a/src/music/parse.rs b/src/music/parse.rs index 689416d..ebd53b9 100644 --- a/src/music/parse.rs +++ b/src/music/parse.rs @@ -1,5 +1,7 @@ use std::path::Path; +use tracing::{debug, trace}; + use crate::music::encoder::MusicMetadataEncoderFactory; use crate::music::metadata::MusicMetadata; use crate::music::parser::MusicMetadataParserFactory; @@ -11,11 +13,23 @@ use crate::music::parser::MusicMetadataParserFactory; /// matching [`MusicMetadataEncoder`] has already baked the tag fields into /// the in-memory `header`, so callers can serve virtualized bytes directly. /// +/// A music-extension file that fails to parse is logged at `warn!` by the +/// selected parser (see [`crate::music::flac`] / [`crate::music::mp3`]); this +/// function simply propagates the resulting `None` without re-logging. +/// /// This helper is the single source of truth for "parse + encode" — used by -/// the local FUSE origin's snapshot builder, the future network origin's -/// manifest builder, and the HTTP/3 server's manifest endpoint. +/// the local FUSE origin's snapshot builder, the network origin's manifest +/// builder, and the server's manifest endpoint. pub fn parse_music_metadata_for_path(path: &Path) -> Option { - let mut metadata = MusicMetadataParserFactory::for_path(path)?.parse(path)?; + let parser = match MusicMetadataParserFactory::for_path(path) { + Some(p) => p, + None => { + trace!(path = %path.display(), "non-music extension; skipping"); + return None; + } + }; + debug!(path = %path.display(), "music parser selected"); + let mut metadata = parser.parse(path)?; if let Some(encoder) = MusicMetadataEncoderFactory::for_path(path) { encoder.encode(&mut metadata); } diff --git a/src/origins/local/mod.rs b/src/origins/local/mod.rs index 42bd43e..fa134b8 100644 --- a/src/origins/local/mod.rs +++ b/src/origins/local/mod.rs @@ -61,7 +61,13 @@ impl Origin for LocalOrigin { pub struct LocalByteSource; impl ByteSource for LocalByteSource { - fn read_at(&self, locator: &Path, offset: u64, len: usize) -> io::Result> { + fn read_at( + &self, + _inode: INodeNo, + locator: &Path, + offset: u64, + len: usize, + ) -> io::Result> { return file_io::read_bytes_at(locator, offset, len); } } diff --git a/src/origins/local/snapshot.rs b/src/origins/local/snapshot.rs index 17161fd..68b2a5a 100644 --- a/src/origins/local/snapshot.rs +++ b/src/origins/local/snapshot.rs @@ -12,10 +12,12 @@ use crate::item::{FileType, Item}; use crate::music::parse::parse_music_metadata_for_path; use crate::origins::attrs::FileAttrs; use crate::virtual_dirs::ensure_virtual_dirs; +use tracing::{error, info}; pub fn fill_fileset(map: &Arc>>, source: &Path, destination: &Path) { match build_snapshot(source, destination) { Ok(new_snapshot) => { + let count = new_snapshot.len(); let mut files = map.lock().unwrap(); files.retain(|ino, _| new_snapshot.contains_key(ino)); @@ -28,8 +30,9 @@ pub fn fill_fileset(map: &Arc>>, source: &Path, de } } } + info!(count, "snapshot rebuilt"); } - Err(e) => eprintln!("Error while reading source: {e}"), + Err(e) => error!(source = %source.display(), error = %e, "error while reading source"), } } diff --git a/src/origins/local/watcher.rs b/src/origins/local/watcher.rs index bfe5283..67b4e1e 100644 --- a/src/origins/local/watcher.rs +++ b/src/origins/local/watcher.rs @@ -10,6 +10,7 @@ use notify::{Event, EventKind, RecursiveMode, Watcher}; use crate::item::Item; use crate::origins::FileWatcher; +use tracing::{debug, error, info, trace}; pub struct LocalOriginFileWatcher { source: PathBuf, @@ -30,38 +31,47 @@ impl FileWatcher for LocalOriginFileWatcher { let source = self.source.clone(); let destination = self.destination.clone(); - println!("Starting to watch the source files in another thread"); + info!(source = %source.display(), "starting file watcher"); thread::spawn(move || { let (tx, rx): ( mpsc::Sender>, mpsc::Receiver>, ) = mpsc::channel(); - let mut watcher: notify::INotifyWatcher = notify::recommended_watcher(tx).unwrap(); + let mut watcher: notify::INotifyWatcher = match notify::recommended_watcher(tx) { + Ok(w) => w, + Err(e) => { + error!(error = %e, "watcher: failed to create notify watcher"); + return; + } + }; - watcher.watch(&source, RecursiveMode::Recursive).unwrap(); + if let Err(e) = watcher.watch(&source, RecursiveMode::Recursive) { + error!(source = %source.display(), error = %e, "watcher: failed to watch source"); + return; + } for res in rx { match res { Ok(event) => { - println!("event: {:?}", event); + trace!(?event, "watcher event"); match event.kind { EventKind::Any => { - println!("Something happened to item, ignoring"); + debug!("watcher: EventKind::Any, ignoring"); } EventKind::Access(_access_kind) => { - println!("Item was read"); + debug!("watcher: item accessed"); } EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => { - println!("Item was removed"); + debug!("watcher: create/modify/remove; rebuilding snapshot"); super::snapshot::fill_fileset(&files, &source, &destination); } EventKind::Other => { - println!("Some other action happened to item, ignoring"); + debug!("watcher: other event, ignoring"); } } } - Err(e) => println!("watch error: {:?}", e), + Err(e) => error!(error = %e, "watcher: notify error"), } } }); diff --git a/src/origins/mod.rs b/src/origins/mod.rs index 897920f..cc70fe0 100644 --- a/src/origins/mod.rs +++ b/src/origins/mod.rs @@ -1,5 +1,6 @@ pub mod attrs; pub mod local; +pub mod network; use std::{ collections::BTreeMap, @@ -11,6 +12,7 @@ use std::{ use fuser::{Errno, FileAttr, Filesystem, Generation, INodeNo, Request}; use sea_orm::{ActiveModelTrait, DatabaseConnection}; +use tracing::{debug, error, trace}; use crate::db::sync::run_db_blocking; use crate::item::{FileType, Item}; @@ -21,8 +23,17 @@ use crate::virtual_dirs::parent_inode_from_path; /// Transport-agnostic byte-range reader. `locator` is whatever string the /// origin treats as a file key: a filesystem path for `LocalOrigin`, a remote /// key for the future network origin. Both produce bytes at `(offset, len)`. +/// +/// `inode` is the FUSE-level inode; NetworkOrigin uses it as the server-side +/// file id (and as the Postgres cache key). LocalOrigin ignores it. pub trait ByteSource: Send + Sync { - fn read_at(&self, locator: &Path, offset: u64, len: usize) -> io::Result>; + fn read_at( + &self, + inode: INodeNo, + locator: &Path, + offset: u64, + len: usize, + ) -> io::Result>; } pub trait FileWatcher: Send + Sync { @@ -80,9 +91,11 @@ impl Filesystem for FuseFs { _flags: fuser::OpenFlags, reply: fuser::ReplyOpen, ) { + trace!(%ino, "open"); if self.files.lock().unwrap().contains_key(&ino) { reply.opened(fuser::FileHandle(ino.0), fuser::FopenFlags::empty()); } else { + debug!(%ino, "open: not found"); reply.error(Errno::ENOENT); } } @@ -105,9 +118,13 @@ impl Filesystem for FuseFs { _flags: Option, reply: fuser::ReplyAttr, ) { + trace!(%ino, "setattr"); match self.files.lock().unwrap().get(&ino) { Some(file) => reply.attr(&Duration::new(1, 0), &file_to_attr(file)), - None => reply.error(Errno::ENOENT), + None => { + debug!(%ino, "setattr: not found"); + reply.error(Errno::ENOENT); + } } } @@ -123,6 +140,7 @@ impl Filesystem for FuseFs { _lock_owner: Option, reply: fuser::ReplyWrite, ) { + trace!(%ino, offset, len = data.len(), "write"); let written = data.len() as u32; let write_start = offset; let write_end = write_start + data.len() as u64; @@ -132,6 +150,7 @@ impl Filesystem for FuseFs { let item = match files.get_mut(&ino) { Some(item) => item, None => { + debug!(%ino, "write: not found"); reply.written(written); return; } @@ -154,6 +173,7 @@ impl Filesystem for FuseFs { let from = (vc_data_offset - write_start) as usize; let to = (vc_data_end - write_start) as usize; mm.update_from_vorbis_comment_data(&data[from..to]); + debug!(%ino, "write: vorbis comment tag update detected"); Some(mm.clone()) } else { None @@ -169,10 +189,12 @@ impl Filesystem for FuseFs { && &data[0..3] == b"ID3" => { mm.update_from_id3_data(data); + debug!(%ino, "write: ID3v2 tag update detected"); Some(mm.clone()) } Some(mm) if !mm.header.is_empty() && data.len() == 128 && &data[0..3] == b"TAG" => { mm.update_from_id3v1_data(data); + debug!(%ino, "write: ID3v1 tag update detected"); Some(mm.clone()) } _ => None, @@ -183,9 +205,11 @@ impl Filesystem for FuseFs { let client = self.client.clone(); let ino_i64 = ino.0 as i64; run_db_blocking(async move { - save_music_metadata(ino_i64, &music_metadata, &client) - .await - .unwrap(); + if let Err(e) = save_music_metadata(ino_i64, &music_metadata, &client).await { + error!(ino = ino_i64, error = %e, "write: save_music_metadata failed"); + } else { + debug!(ino = ino_i64, "write: persisted updated music metadata"); + } }); } @@ -199,14 +223,17 @@ impl Filesystem for FuseFs { _fh: Option, reply: fuser::ReplyAttr, ) { - println!("getattr(ino={})", ino); + trace!(%ino, "getattr"); match self.files.lock().unwrap().get(&ino) { Some(file) => { let ttl = Duration::new(1, 0); let attr = file_to_attr(file); reply.attr(&ttl, &attr); } - None => reply.error(Errno::ENOENT), + None => { + debug!(%ino, "getattr: not found"); + reply.error(Errno::ENOENT); + } } } @@ -218,13 +245,14 @@ impl Filesystem for FuseFs { offset: u64, mut reply: fuser::ReplyDirectory, ) { - println!("readdir(ino={}, fh={}, offset={})", ino, fh, offset); + trace!(%ino, %fh, offset, "readdir"); let files = self.files.lock().unwrap(); let parent_inode = match files.get(&ino) { Some(dir) => dir.parent_inode, None => { + debug!(%ino, "readdir: not found"); reply.error(Errno::ENOENT); return; } @@ -278,9 +306,17 @@ impl Filesystem for FuseFs { _flags: fuser::RenameFlags, reply: fuser::ReplyEmpty, ) { + trace!( + %parent, + name = %name.display(), + %newparent, + newname = %newname.display(), + "rename" + ); let name_str = match name.to_str() { Some(s) => s, None => { + error!(%parent, "rename: source name is not valid UTF-8"); reply.error(Errno::EINVAL); return; } @@ -288,6 +324,7 @@ impl Filesystem for FuseFs { let newname_str = match newname.to_str() { Some(s) => s, None => { + error!(%newparent, "rename: target name is not valid UTF-8"); reply.error(Errno::EINVAL); return; } @@ -301,6 +338,7 @@ impl Filesystem for FuseFs { { Some((ino, _)) => *ino, None => { + debug!(%parent, name = %name_str, "rename: source not found"); reply.error(Errno::ENOENT); return; } @@ -312,12 +350,15 @@ impl Filesystem for FuseFs { match files.get(&newparent) { Some(dir) => dir.local_path.join(newname_str), None => { + debug!(%newparent, "rename: target parent not found"); reply.error(Errno::ENOENT); return; } } }; + debug!(ino = %item_ino, from = %name_str, to = %newname_str, "rename"); + let new_parent_inode = parent_inode_from_path(&new_local_path); let item = files.get_mut(&item_ino).unwrap(); @@ -334,15 +375,17 @@ impl Filesystem for FuseFs { run_db_blocking(async move { use sea_orm::ActiveValue::Set; - crate::db::entities::ActiveModel { + if let Err(e) = (crate::db::entities::ActiveModel { inode: Set(inode_i64), name: Set(new_name_owned), local_path: Set(new_local_path_str), ..Default::default() } .update(&client) - .await - .unwrap(); + .await) + { + error!(ino = inode_i64, error = %e, "rename: db update failed"); + } }); reply.ok(); @@ -359,15 +402,18 @@ impl Filesystem for FuseFs { _lock_owner: Option, reply: fuser::ReplyData, ) { - let (locator, flac) = { + trace!(%ino, offset, size, "read"); + let (inode, locator, flac) = { let files = self.files.lock().unwrap(); let item = match files.get(&ino) { Some(f) => f, None => { + debug!(%ino, "read: not found"); reply.error(Errno::ENOENT); return; } }; + let inode = item.inode; let locator = item.original_path.clone(); let flac = item .music_metadata @@ -381,19 +427,22 @@ impl Filesystem for FuseFs { mm.real_audio_start, ) }); - (locator, flac) + (inode, locator, flac) }; let Some((header, pic_hdrs, pic_ranges, real_audio_start)) = flac else { - match self.bytes.read_at(&locator, offset, size as usize) { + match self.bytes.read_at(inode, &locator, offset, size as usize) { Ok(bytes) => reply.data(&bytes), - Err(_) => reply.error(Errno::EIO), + Err(e) => { + error!(%inode, offset, size, error = %e, "read: read_at failed; returning EIO"); + reply.error(Errno::EIO); + } } return; }; let bytes = &self.bytes; - let reader = |off: u64, len: usize| bytes.read_at(&locator, off, len); + let reader = |off: u64, len: usize| bytes.read_at(inode, &locator, off, len); match file_io::assemble_flac_read( &reader, &header, @@ -404,7 +453,10 @@ impl Filesystem for FuseFs { size, ) { Ok(bytes) => reply.data(&bytes), - Err(_) => reply.error(Errno::EIO), + Err(e) => { + error!(%inode, offset, size, error = %e, "read: flac assembly failed; returning EIO"); + reply.error(Errno::EIO); + } } } @@ -415,7 +467,7 @@ impl Filesystem for FuseFs { name: &std::ffi::OsStr, reply: fuser::ReplyEntry, ) { - println!("lookup(parent={}, name={})", parent, name.display()); + trace!(%parent, name = %name.display(), "lookup"); match self .files @@ -430,7 +482,10 @@ impl Filesystem for FuseFs { reply.entry(&ttl, &attr, Generation(0)); } - None => reply.error(Errno::ENOENT), + None => { + debug!(%parent, name = %name.display(), "lookup: not found"); + reply.error(Errno::ENOENT); + } } } } diff --git a/src/origins/network/mod.rs b/src/origins/network/mod.rs new file mode 100644 index 0000000..f88fda1 --- /dev/null +++ b/src/origins/network/mod.rs @@ -0,0 +1,344 @@ +pub mod transport; +pub mod watcher; + +use std::{ + collections::BTreeMap, + io, + path::{Path, PathBuf}, + sync::{Arc, RwLock}, + time::{Duration, SystemTime}, +}; + +use fuser::INodeNo; +use sea_orm::EntityTrait; + +use crate::db::cache::{delete_cached_bytes_for, get_cached_bytes, put_cached_bytes}; +use crate::db::entities as item_entities; +use crate::db::sync::{run_db_blocking, sync_items_to_db}; +use crate::item::{FileType, Item}; +use crate::music::db::restore_music_metadata_from_db; +use crate::music::metadata::MusicMetadata; +use crate::origins::attrs::FileAttrs; +use crate::origins::{ByteSource, FileWatcher, Origin}; +use crate::proto::ManifestEntry as ProtoManifestEntry; +use crate::virtual_dirs::{ensure_virtual_dirs, restore_virtual_paths}; +use tracing::{debug, error, info, trace}; + +use self::transport::NetworkTransport; + +pub struct NetworkOrigin { + pub(crate) endpoint: String, + pub(crate) destination: PathBuf, + runtime: tokio::runtime::Runtime, + transport: NetworkTransport, + client: sea_orm::DatabaseConnection, + /// In-memory cache of the latest manifest, keyed by server id. Written by + /// `snapshot()`, read by the watcher to know what state to reconcile + /// against on the next pass without round-tripping through the DB. + latest_manifest: Arc>>, +} + +impl std::fmt::Debug for NetworkOrigin { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return f + .debug_struct("NetworkOrigin") + .field("endpoint", &self.endpoint) + .field("destination", &self.destination) + .finish(); + } +} + +impl NetworkOrigin { + pub fn new( + endpoint: String, + destination: String, + client: sea_orm::DatabaseConnection, + ) -> io::Result { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(io_err)?; + let transport = NetworkTransport::new(endpoint.clone()).map_err(io_err)?; + return Ok(NetworkOrigin { + endpoint, + destination: destination.into(), + runtime, + transport, + client, + latest_manifest: Arc::new(RwLock::new(BTreeMap::new())), + }); + } + + pub fn runtime_handle(&self) -> tokio::runtime::Handle { + return self.runtime.handle().clone(); + } +} + +impl Origin for NetworkOrigin { + fn snapshot(&self) -> io::Result> { + return run_db_blocking(self.snapshot_async()).map_err(io_err); + } + + fn byte_source(&self) -> Arc { + return Arc::new(NetworkByteSource { + transport: self.transport.clone(), + runtime_handle: self.runtime_handle(), + client: self.client.clone(), + }); + } + + fn watcher(&self) -> Box { + return Box::new(watcher::NetworkOriginFileWatcher::new( + self.transport.clone(), + self.runtime_handle(), + self.client.clone(), + self.destination.clone(), + self.latest_manifest.clone(), + )); + } +} + +impl NetworkOrigin { + async fn snapshot_async(&self) -> io::Result> { + // 1. Pull client's current (inode, hash) pairs from the DB. + let client_entries: Vec<(u64, u64)> = item_entities::Entity::find() + .all(&self.client) + .await + .map_err(|e| { + error!(error = %e, "network snapshot: client entries DB read failed"); + io_err(e) + })? + .into_iter() + .map(|m| (m.inode as u64, m.hash as u64)) + .collect(); + + // 2. Ask server what changed. + let response = self + .transport + .reconcile(client_entries) + .await + .map_err(|e| { + error!(error = %e, "network snapshot: reconcile with server failed"); + io_err(e) + })?; + + // 3. Invalidate cached bytes for changed + deleted inodes — they are + // stale by definition. + let mut changed_or_deleted: Vec = + response.changed.iter().map(|ih| ih.inode as i64).collect(); + changed_or_deleted.extend(response.deleted.iter().map(|i| *i as i64)); + delete_cached_bytes_for(&changed_or_deleted, &self.client).await; + + // 4. Fetch fresh metadata for changed inodes only. + let wanted: Vec = response.changed.iter().map(|ih| ih.inode).collect(); + let mut current_manifest = self.latest_manifest.read().unwrap().clone(); + if !wanted.is_empty() { + let wanted_count = wanted.len(); + let entries = self.transport.get_metadata(wanted).await.map_err(|e| { + error!( + wanted = wanted_count, + error = %e, + "network snapshot: get_metadata from server failed" + ); + io_err(e) + })?; + for entry in entries { + current_manifest.insert(entry.id, entry); + } + } + for inode in &response.deleted { + current_manifest.remove(inode); + } + *self.latest_manifest.write().unwrap() = current_manifest.clone(); + + // 5. Convert manifest entries → Items. + let dest = self.destination.clone(); + let mut snapshot = BTreeMap::new(); + let root_attrs = FileAttrs { + size: 0, + blocks: 0, + atime: SystemTime::UNIX_EPOCH, + mtime: SystemTime::UNIX_EPOCH, + ctime: SystemTime::UNIX_EPOCH, + crtime: SystemTime::UNIX_EPOCH, + perm: 0o755, + nlink: 2, + uid: 0, + gid: 0, + rdev: 0, + blksize: 4096, + }; + snapshot.insert( + INodeNo::ROOT, + Item::new( + INodeNo::ROOT, + INodeNo::ROOT, + "/".to_string(), + dest.clone(), + dest.clone(), + FileType::Directory, + root_attrs, + None, + ), + ); + + let source_root = PathBuf::from("/"); + for (_id, entry) in ¤t_manifest { + let item = manifest_entry_to_item(entry, &source_root, &mut snapshot); + snapshot.insert(item.inode, item); + } + + // 6. Reconcile with DB exactly like LocalOrigin: insert/update/delete + // rows, then restore music metadata + virtual paths. + let db_items: std::collections::HashMap = + item_entities::Entity::find() + .all(&self.client) + .await + .map_err(|e| { + error!(error = %e, "network snapshot: db_items DB read failed"); + io_err(e) + })? + .into_iter() + .map(|e| (e.inode, e)) + .collect(); + sync_items_to_db(&snapshot, &db_items, &self.client).await; + restore_music_metadata_from_db(&mut snapshot, &db_items, &self.client).await; + restore_virtual_paths(&mut snapshot, &db_items, &source_root); + + info!( + changed = response.changed.len(), + deleted = response.deleted.len(), + files = snapshot.len(), + "network snapshot complete" + ); + return Ok(snapshot); + } +} + +fn manifest_entry_to_item( + entry: &ProtoManifestEntry, + source_root: &Path, + snapshot: &mut BTreeMap, +) -> Item { + let inode = INodeNo(entry.id); + let original_path = PathBuf::from(&entry.rel_path); + let attrs = FileAttrs { + size: entry.size, + blocks: 0, + atime: SystemTime::UNIX_EPOCH + Duration::from_secs(entry.mtime), + mtime: SystemTime::UNIX_EPOCH + Duration::from_secs(entry.mtime), + ctime: SystemTime::UNIX_EPOCH + Duration::from_secs(entry.ctime), + crtime: SystemTime::UNIX_EPOCH + Duration::from_secs(entry.crtime), + perm: 0o644, + nlink: 1, + uid: 0, + gid: 0, + rdev: 0, + blksize: 4096, + }; + let music_metadata: Option = entry + .music_metadata + .clone() + .map(|mm| MusicMetadata::from(mm)); + + let mut local_path = PathBuf::new(); + if let Some(mm) = &music_metadata { + let joined; + let artist_dir = match mm.album_artist.as_deref() { + Some(a) => a, + None => { + joined = mm.artist.join("-"); + &joined + } + }; + local_path.push(artist_dir); + local_path.push(&mm.album); + } + let name = Path::new(&entry.rel_path) + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_else(|| entry.rel_path.clone()); + local_path.push(&name); + + let parent_inode = ensure_virtual_dirs(&local_path, source_root, snapshot); + return Item::new( + inode, + parent_inode, + name, + original_path, + local_path, + FileType::File, + attrs, + music_metadata, + ); +} + +impl From for MusicMetadata { + fn from(mm: crate::proto::MusicMetadata) -> Self { + return MusicMetadata { + artist: mm.artist, + album_artist: mm.album_artist, + album: mm.album, + track_number: mm.track_number, + track_title: mm.track_title, + other_tags: mm.other_tags, + header: mm.header, + picture_block_headers: mm.picture_block_headers, + picture_data_ranges: mm + .picture_data_ranges + .into_iter() + .map(|p| (p.offset, p.length)) + .collect(), + real_audio_start: mm.real_audio_start, + vorbis_comment_offset: mm.vorbis_comment_offset, + vorbis_comment_length: mm.vorbis_comment_length, + }; + } +} + +fn io_err(e: E) -> io::Error { + return io::Error::new(io::ErrorKind::Other, e.to_string()); +} + +pub struct NetworkByteSource { + transport: NetworkTransport, + runtime_handle: tokio::runtime::Handle, + client: sea_orm::DatabaseConnection, +} + +impl ByteSource for NetworkByteSource { + fn read_at( + &self, + inode: INodeNo, + _locator: &Path, + offset: u64, + len: usize, + ) -> io::Result> { + let inode_i64 = inode.0 as i64; + trace!(%inode, offset, len, "network read_at"); + + let cached = run_db_blocking(get_cached_bytes(inode_i64, &self.client)); + if let Some(data) = cached { + debug!(%inode, "read_at cache hit"); + return slice_range(&data, offset, len); + } + + debug!(%inode, "read_at cache miss; fetching from server"); + let (data, _total_size) = self + .runtime_handle + .block_on(self.transport.fetch_file_range(inode.0, 0, 0)) + .map_err(|e| { + error!(%inode, error = %e, "read_at: fetch_file_range failed"); + io_err(e) + })?; + run_db_blocking(put_cached_bytes(inode_i64, data.clone(), &self.client)); + + return slice_range(&data, offset, len); + } +} + +fn slice_range(data: &[u8], offset: u64, len: usize) -> io::Result> { + let start = (offset as usize).min(data.len()); + let end = (start + len).min(data.len()); + return Ok(data[start..end].to_vec()); +} diff --git a/src/origins/network/transport.rs b/src/origins/network/transport.rs new file mode 100644 index 0000000..7aa4f8d --- /dev/null +++ b/src/origins/network/transport.rs @@ -0,0 +1,114 @@ +use std::sync::Arc; + +use anyhow::{Context, Result, anyhow}; +use tokio_stream::StreamExt; +use tonic::codec::Streaming; + +use crate::proto::{ + ChangeEvent, GetFileRequest, GetMetadataRequest, InodeHash, ManifestEntry, MusicFsClient, + ReconcileRequest, ReconcileResponse, SubscribeEventsRequest, +}; + +/// Thin wrapper over the generated tonic client. Owns the connection and +/// exposes four operations matching the four RPCs NetworkOrigin needs. +/// +/// All methods are async and must be polled on the runtime whose Handle was +/// passed in at construction (or a child of it). The sync FUSE path uses +/// `runtime_handle.block_on(...)` to enter this runtime. +#[derive(Clone)] +pub struct NetworkTransport { + client: Arc>>, + endpoint: tonic::transport::Endpoint, +} + +impl NetworkTransport { + /// Build a transport that connects lazily. The first RPC establishes the + /// connection; reconnects happen automatically if the channel drops. + pub fn new(url: String) -> Result { + let endpoint: tonic::transport::Endpoint = url + .try_into() + .map_err(|e| anyhow!("invalid server URL: {e}"))?; + let channel = endpoint.connect_lazy(); + let client = Arc::new(tokio::sync::Mutex::new(MusicFsClient::new(channel))); + return Ok(NetworkTransport { client, endpoint }); + } + + pub async fn reconcile(&self, entries: Vec<(u64, u64)>) -> Result { + let request = ReconcileRequest { + entries: entries + .into_iter() + .map(|(inode, hash)| InodeHash { inode, hash }) + .collect(), + }; + let mut client = self.client.lock().await; + let response = client + .reconcile(request) + .await + .context("Reconcile RPC failed")?; + return Ok(response.into_inner()); + } + + pub async fn get_metadata(&self, inodes: Vec) -> Result> { + let request = GetMetadataRequest { inodes }; + let mut client = self.client.lock().await; + let mut stream: Streaming = client + .get_metadata(request) + .await + .context("GetMetadata RPC failed")? + .into_inner(); + let mut out = Vec::new(); + while let Some(entry) = stream.next().await { + out.push(entry.context("GetMetadata stream error")?); + } + return Ok(out); + } + + pub async fn fetch_file_range( + &self, + id: u64, + start: u64, + length: u64, + ) -> Result<(Vec, u64)> { + let request = GetFileRequest { id, start, length }; + let mut client = self.client.lock().await; + let mut stream = client + .get_file(request) + .await + .context("GetFile RPC failed")? + .into_inner(); + let first = stream + .next() + .await + .ok_or_else(|| anyhow!("GetFile returned empty stream for id {id}"))? + .context("GetFile stream error")?; + let total_size = first.total_size; + let mut data = first.data; + while let Some(chunk) = stream.next().await { + let chunk = chunk.context("GetFile stream error")?; + data.extend_from_slice(&chunk.data); + } + return Ok((data, total_size)); + } + + /// Take a fresh receiver on the SubscribeEvents stream. Each call opens a + /// new server-streaming RPC; the caller owns the lifetime. + pub async fn subscribe_events(&self) -> Result> { + let mut client = self.client.lock().await; + let stream = client + .subscribe_events(SubscribeEventsRequest {}) + .await + .context("SubscribeEvents RPC failed")? + .into_inner(); + return Ok(stream); + } + + /// Reconnect — used by the watcher when the channel has gone bad. + #[allow(dead_code)] + pub async fn reconnect(&self) -> Result<()> { + let channel = self.endpoint.connect().await.context("reconnect failed")?; + let new_client = MusicFsClient::new(channel); + let mut guard = self.client.lock().await; + *guard = new_client; + return Ok(()); + } +} diff --git a/src/origins/network/watcher.rs b/src/origins/network/watcher.rs new file mode 100644 index 0000000..b9f0d39 --- /dev/null +++ b/src/origins/network/watcher.rs @@ -0,0 +1,142 @@ +use std::{ + collections::BTreeMap, + path::PathBuf, + sync::{Arc, RwLock}, + thread, + time::Duration, +}; + +use fuser::INodeNo; +use sea_orm::{DatabaseConnection, EntityTrait}; +use tokio_stream::StreamExt; + +use crate::item::Item; +use crate::origins::FileWatcher; +use crate::origins::network::transport::NetworkTransport; +use crate::proto::ManifestEntry as ProtoManifestEntry; +use tracing::{info, warn}; + +const POLL_FALLBACK_INTERVAL: Duration = Duration::from_secs(30); + +pub struct NetworkOriginFileWatcher { + transport: NetworkTransport, + runtime_handle: tokio::runtime::Handle, + client: DatabaseConnection, + destination: PathBuf, + latest_manifest: Arc>>, +} + +impl NetworkOriginFileWatcher { + #[allow(clippy::too_many_arguments)] + pub fn new( + transport: NetworkTransport, + runtime_handle: tokio::runtime::Handle, + client: DatabaseConnection, + destination: PathBuf, + latest_manifest: Arc>>, + ) -> Self { + return NetworkOriginFileWatcher { + transport, + runtime_handle, + client, + destination, + latest_manifest, + }; + } +} + +/// `files` is currently unused at runtime by the network watcher — the +/// authoritative post-reconcile snapshot is written to the DB, and the next +/// FUSE `readdir`/`lookup` will reflect it because FuseFs re-locks the map. +/// The trait signature still requires it for parity with LocalOriginFileWatcher; +/// a future refactor can rebuild the map in place here. +impl FileWatcher for NetworkOriginFileWatcher { + fn watch(&self, files: Arc>>) { + let runtime_handle = self.runtime_handle.clone(); + let state = WatcherState { + transport: self.transport.clone(), + client: self.client.clone(), + destination: self.destination.clone(), + latest_manifest: self.latest_manifest.clone(), + files, + }; + thread::spawn(move || { + runtime_handle.block_on(state.run_loop()); + }); + } +} + +#[derive(Clone)] +struct WatcherState { + transport: NetworkTransport, + client: DatabaseConnection, + destination: PathBuf, + latest_manifest: Arc>>, + files: Arc>>, +} + +impl WatcherState { + async fn run_loop(self) { + loop { + match self.transport.subscribe_events().await { + Ok(mut stream) => { + info!("network watcher: subscribed to /events"); + while let Some(item) = stream.next().await { + match item { + Ok(_event) => { + if let Err(e) = self.reconcile_once().await { + warn!(error = %e, "network watcher: reconcile after event failed"); + } + } + Err(e) => { + warn!(error = %e, "network watcher: stream error; reconnecting"); + break; + } + } + } + } + Err(e) => { + warn!(error = %e, "network watcher: subscribe failed; will retry"); + } + } + tokio::time::sleep(POLL_FALLBACK_INTERVAL).await; + if let Err(e) = self.reconcile_once().await { + warn!(error = %e, "network watcher: poll reconcile failed"); + } + } + } + + async fn reconcile_once(&self) -> anyhow::Result<()> { + let client_entries: Vec<(u64, u64)> = crate::db::entities::Entity::find() + .all(&self.client) + .await? + .into_iter() + .map(|m| (m.inode as u64, m.hash as u64)) + .collect(); + + let response = self.transport.reconcile(client_entries).await?; + + let mut changed_or_deleted: Vec = + response.changed.iter().map(|ih| ih.inode as i64).collect(); + changed_or_deleted.extend(response.deleted.iter().map(|i| *i as i64)); + crate::db::cache::delete_cached_bytes_for(&changed_or_deleted, &self.client).await; + + let wanted: Vec = response.changed.iter().map(|ih| ih.inode).collect(); + let mut current_manifest = self.latest_manifest.read().unwrap().clone(); + if !wanted.is_empty() { + let entries = self.transport.get_metadata(wanted).await?; + for entry in entries { + current_manifest.insert(entry.id, entry); + } + } + for inode in &response.deleted { + current_manifest.remove(inode); + } + *self.latest_manifest.write().unwrap() = current_manifest.clone(); + + let _ = self.destination.clone(); + let _ = self.files.clone(); + + return Ok(()); + } +} diff --git a/src/proto/generated/musicfs/musicfs.rs b/src/proto/generated/musicfs/musicfs.rs index 1dde6bb..43062de 100644 --- a/src/proto/generated/musicfs/musicfs.rs +++ b/src/proto/generated/musicfs/musicfs.rs @@ -1,5 +1,29 @@ // @generated // This file is @generated by prost-build. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct ReconcileRequest { + #[prost(message, repeated, tag = "1")] + pub entries: ::prost::alloc::vec::Vec, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct ReconcileResponse { + #[prost(message, repeated, tag = "1")] + pub changed: ::prost::alloc::vec::Vec, + #[prost(uint64, repeated, tag = "2")] + pub deleted: ::prost::alloc::vec::Vec, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct InodeHash { + #[prost(uint64, tag = "1")] + pub inode: u64, + #[prost(uint64, tag = "2")] + pub hash: u64, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct GetMetadataRequest { + #[prost(uint64, repeated, tag = "1")] + pub inodes: ::prost::alloc::vec::Vec, +} #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] pub struct GetManifestRequest {} #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] diff --git a/src/proto/generated/musicfs/musicfs.tonic.rs b/src/proto/generated/musicfs/musicfs.tonic.rs index 1f903e9..63300c6 100644 --- a/src/proto/generated/musicfs/musicfs.tonic.rs +++ b/src/proto/generated/musicfs/musicfs.tonic.rs @@ -89,6 +89,37 @@ pub mod music_fs_client { self.inner = self.inner.max_encoding_message_size(limit); self } + pub async fn reconcile( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner.ready().await.map_err(|e| { + tonic::Status::unknown(format!("Service was not ready: {}", e.into())) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/musicfs.MusicFs/Reconcile"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("musicfs.MusicFs", "Reconcile")); + self.inner.unary(req, path, codec).await + } + pub async fn get_metadata( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response>, + tonic::Status, + > { + self.inner.ready().await.map_err(|e| { + tonic::Status::unknown(format!("Service was not ready: {}", e.into())) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/musicfs.MusicFs/GetMetadata"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("musicfs.MusicFs", "GetMetadata")); + self.inner.server_streaming(req, path, codec).await + } pub async fn get_manifest( &mut self, request: impl tonic::IntoRequest, @@ -155,6 +186,19 @@ pub mod music_fs_server { /// Generated trait containing gRPC methods that should be implemented for use with MusicFsServer. #[async_trait] pub trait MusicFs: std::marker::Send + std::marker::Sync + 'static { + async fn reconcile( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; + /// Server streaming response type for the GetMetadata method. + type GetMetadataStream: tonic::codegen::tokio_stream::Stream< + Item = std::result::Result, + > + std::marker::Send + + 'static; + async fn get_metadata( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; /// Server streaming response type for the GetManifest method. type GetManifestStream: tonic::codegen::tokio_stream::Stream< Item = std::result::Result, @@ -256,6 +300,87 @@ pub mod music_fs_server { } fn call(&mut self, req: http::Request) -> Self::Future { match req.uri().path() { + "/musicfs.MusicFs/Reconcile" => { + #[allow(non_camel_case_types)] + struct ReconcileSvc(pub Arc); + impl tonic::server::UnaryService for ReconcileSvc { + type Response = super::ReconcileResponse; + type Future = BoxFuture, tonic::Status>; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = + async move { ::reconcile(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = ReconcileSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/musicfs.MusicFs/GetMetadata" => { + #[allow(non_camel_case_types)] + struct GetMetadataSvc(pub Arc); + impl + tonic::server::ServerStreamingService + for GetMetadataSvc + { + type Response = super::ManifestEntry; + type ResponseStream = T::GetMetadataStream; + type Future = + BoxFuture, tonic::Status>; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = + async move { ::get_metadata(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = GetMetadataSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.server_streaming(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/musicfs.MusicFs/GetManifest" => { #[allow(non_camel_case_types)] struct GetManifestSvc(pub Arc); diff --git a/src/proto/mod.rs b/src/proto/mod.rs index ab182fc..76917a3 100644 --- a/src/proto/mod.rs +++ b/src/proto/mod.rs @@ -11,7 +11,9 @@ pub mod musicfs { } pub use musicfs::{ - ChangeEvent, FileChunk, GetFileRequest, GetManifestRequest, ManifestEntry, MusicMetadata, - PictureDataRange, SubscribeEventsRequest, + ChangeEvent, FileChunk, GetFileRequest, GetManifestRequest, GetMetadataRequest, InodeHash, + ManifestEntry, MusicMetadata, PictureDataRange, ReconcileRequest, ReconcileResponse, + SubscribeEventsRequest, + music_fs_client::MusicFsClient, music_fs_server::{MusicFs, MusicFsServer}, }; diff --git a/src/server/manifest.rs b/src/server/manifest.rs index 75e9b8f..6d3bb60 100644 --- a/src/server/manifest.rs +++ b/src/server/manifest.rs @@ -1,3 +1,4 @@ +use crate::item::compute_item_hash; use crate::music::metadata::MusicMetadata; /// One row of the in-memory manifest: every field the client needs to @@ -29,6 +30,18 @@ pub struct ManifestEntry { pub music_metadata: Option, } +impl ManifestEntry { + pub fn hash(&self) -> u64 { + return compute_item_hash( + self.id, + self.rel_path.as_bytes(), + self.ctime, + self.mtime, + self.crtime, + ); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/server/state.rs b/src/server/state.rs index 3907804..6c05da0 100644 --- a/src/server/state.rs +++ b/src/server/state.rs @@ -11,6 +11,7 @@ use std::os::unix::fs::MetadataExt; use crate::music::parse::parse_music_metadata_for_path; use crate::origins::attrs::FileAttrs; use crate::server::manifest::ManifestEntry; +use tracing::info; /// Server-side entry for one file. Built once on startup from a directory /// scan and refreshed by the watcher on inotify events. @@ -38,11 +39,13 @@ impl ServerState { /// and on watcher-driven reconciliations. pub fn replace_all(&self, source: &Path) -> io::Result<()> { let entries = scan_directory(source)?; + let count = entries.len(); let mut map = self.inner.lock().unwrap(); map.clear(); for (id, entry) in entries { map.insert(id, entry); } + info!(count, "server state: scan complete"); return Ok(()); } diff --git a/src/server/transport/grpc.rs b/src/server/transport/grpc.rs index b5c318a..648e656 100644 --- a/src/server/transport/grpc.rs +++ b/src/server/transport/grpc.rs @@ -14,13 +14,16 @@ use crate::music::metadata::MusicMetadata; use crate::proto as proto_types; use crate::proto::MusicFs as MusicFsTrait; use crate::proto::{ - ChangeEvent as ProtoChangeEvent, FileChunk, GetFileRequest, GetManifestRequest, ManifestEntry, - MusicFsServer, MusicMetadata as ProtoMusicMetadata, PictureDataRange, SubscribeEventsRequest, + ChangeEvent as ProtoChangeEvent, FileChunk, GetFileRequest, GetManifestRequest, + GetMetadataRequest, InodeHash, ManifestEntry, MusicFsServer, + MusicMetadata as ProtoMusicMetadata, PictureDataRange, ReconcileRequest, ReconcileResponse, + SubscribeEventsRequest, }; use crate::server::manifest::ManifestEntry as DomainManifestEntry; use crate::server::state::ServerState; use crate::server::transport::{MusicTransport, TransportArgs}; use crate::server::watcher::{ChangeEvent, ChangeKind}; +use tracing::{debug, info, warn}; pub struct GrpcTransport { args: TransportArgs, @@ -40,7 +43,7 @@ impl MusicTransport for GrpcTransport { events: tokio::sync::Mutex::new(events), }; let listen = self.args.listen; - println!("musicfs-server gRPC listening on {listen}"); + info!(%listen, "musicfs-server gRPC listening"); Server::builder() .add_service(MusicFsServer::new(service)) .serve(listen) @@ -59,13 +62,78 @@ type BoxStream = Pin> + #[tonic::async_trait] impl MusicFsTrait for MusicFsService { type GetManifestStream = BoxStream; + type GetMetadataStream = BoxStream; type GetFileStream = BoxStream; type SubscribeEventsStream = BoxStream; + async fn reconcile( + &self, + request: Request, + ) -> Result, Status> { + let client_entries: std::collections::HashMap = request + .into_inner() + .entries + .into_iter() + .map(|ih| (ih.inode, ih.hash)) + .collect(); + debug!(client_entries = client_entries.len(), "reconcile"); + + let server_manifest = self.state.manifest(); + let mut changed: Vec = Vec::new(); + let mut deleted: Vec = Vec::new(); + + for entry in &server_manifest { + let entry_hash = entry.hash(); + match client_entries.get(&entry.id) { + Some(client_hash) if *client_hash == entry_hash => {} + _ => { + changed.push(InodeHash { + inode: entry.id, + hash: entry_hash, + }); + } + } + } + for client_inode in client_entries.keys() { + if !server_manifest.iter().any(|e| &e.id == client_inode) { + deleted.push(*client_inode); + } + } + + debug!( + changed = changed.len(), + deleted = deleted.len(), + "reconcile result" + ); + return Ok(Response::new(ReconcileResponse { changed, deleted })); + } + + async fn get_metadata( + &self, + request: Request, + ) -> Result, Status> { + let wanted: std::collections::HashSet = + request.into_inner().inodes.into_iter().collect(); + debug!(wanted = wanted.len(), "get_metadata"); + let entries: Vec = self + .state + .manifest() + .into_iter() + .filter(|e| wanted.contains(&e.id)) + .collect(); + let stream = tokio_stream::iter( + entries + .into_iter() + .map(|e| Ok::(e.into())), + ); + return Ok(Response::new(Box::pin(stream))); + } + async fn get_manifest( &self, _request: Request, ) -> Result, Status> { + debug!("get_manifest"); let entries: Vec = self.state.manifest(); let stream = tokio_stream::iter( entries @@ -81,10 +149,14 @@ impl MusicFsTrait for MusicFsService { ) -> Result, Status> { let req = request.into_inner(); let id = req.id; - let entry = self - .state - .lookup(id) - .ok_or_else(|| Status::not_found(format!("no file with id {id}")))?; + debug!(%id, "get_file"); + let entry = match self.state.lookup(id) { + Some(e) => e, + None => { + debug!(%id, "get_file: not found"); + return Err(Status::not_found(format!("no file with id {id}"))); + } + }; let abs_path: PathBuf = entry.abs_path.clone(); let total_size = entry.attrs.size; let start = if req.start == 0 && req.length == 0 { @@ -103,8 +175,14 @@ impl MusicFsTrait for MusicFsService { let chunk = tokio::task::spawn_blocking(move || read_range_blocking(&abs_path, start, length)) .await - .map_err(|e| Status::internal(format!("join blocking read: {e}")))? - .map_err(|e| Status::internal(format!("read file range: {e}")))?; + .map_err(|e| { + warn!(%id, error = %e, "get_file: join blocking read failed"); + Status::internal(format!("join blocking read: {e}")) + })? + .map_err(|e| { + warn!(%id, error = %e, "get_file: read range failed"); + Status::internal(format!("read file range: {e}")) + })?; let chunk_stream = tokio_stream::iter(vec![Ok::(FileChunk { data: chunk, @@ -118,6 +196,7 @@ impl MusicFsTrait for MusicFsService { &self, _request: Request, ) -> Result, Status> { + debug!("subscribe_events"); // Take an owned receiver (the guard is dropped at the end of this // statement) so it can move into the 'static spawned task, and so each // subscriber gets its own receiver rather than contending on one. @@ -135,8 +214,14 @@ impl MusicFsTrait for MusicFsService { break; } } - Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, - Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + warn!(lagged = n, "subscribe_events: lagged; continuing"); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + debug!("subscribe_events: broadcast closed; ending stream"); + break; + } } } }); diff --git a/src/server/watcher.rs b/src/server/watcher.rs index a704e71..b5579e3 100644 --- a/src/server/watcher.rs +++ b/src/server/watcher.rs @@ -8,6 +8,7 @@ use notify::{EventKind, RecursiveMode, Watcher}; use tokio::sync::broadcast; use crate::server::state::ServerState; +use tracing::{error, info}; /// A change observed by the watcher. Pushed onto the broadcast channel for /// `/events` subscribers. The client treats these as wake-ups: correctness @@ -64,17 +65,17 @@ fn run_watcher_loop( let mut watcher = match notify::recommended_watcher(tx) { Ok(w) => w, Err(e) => { - eprintln!("server watcher: failed to create inotify watcher: {e}"); + error!(error = %e, "server watcher: failed to create inotify watcher"); return; } }; if let Err(e) = watcher.watch(&source, RecursiveMode::Recursive) { - eprintln!("server watcher: failed to watch {}: {e}", source.display()); + error!(source = %source.display(), error = %e, "server watcher: failed to watch source"); return; } - println!("server watcher: scanning {source:?} for changes"); + info!(source = %source.display(), "server watcher: scanning for changes"); for res in rx { match res { Ok(event) => match event.kind { @@ -86,14 +87,14 @@ fn run_watcher_loop( _ => continue, }; if let Err(e) = state.replace_all(&source) { - eprintln!("server watcher: state refresh failed: {e}"); + error!(error = %e, "server watcher: state refresh failed"); continue; } let _ = events_tx.send(ChangeEvent { kind }); } _ => {} }, - Err(e) => eprintln!("server watcher: inotify error: {e}"), + Err(e) => error!(error = %e, "server watcher: inotify error"), } } }