diff --git a/crates/epix-fs/src/lib.rs b/crates/epix-fs/src/lib.rs index 0b82af1..08c9de3 100644 --- a/crates/epix-fs/src/lib.rs +++ b/crates/epix-fs/src/lib.rs @@ -22,6 +22,43 @@ fn checked_relative_path(path: &Path) -> io::Result<()> { static WINDOWS_TEMP_SEQUENCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); +/// Encode `path` as a NUL-terminated verbatim (`\\?\`) wide string for raw +/// Win32 calls. std's own fs wrappers convert long paths to verbatim form +/// internally, but a path handed straight to e.g. `MoveFileExW` is subject to +/// the 260-char MAX_PATH limit and fails with `ERROR_PATH_NOT_FOUND` (3) — +/// the staged-promotion tree (three 64-char hash directories) crosses that +/// limit routinely. Verbatim paths skip separator normalization, so forward +/// slashes are rewritten here. +#[cfg(windows)] +pub fn verbatim_wide_null(path: &Path) -> io::Result> { + use std::os::windows::ffi::OsStrExt; + + const SEP: u16 = b'\\' as u16; + let absolute = std::path::absolute(path)?; + let mut wide: Vec = absolute.as_os_str().encode_wide().collect(); + for unit in &mut wide { + if *unit == b'/' as u16 { + *unit = SEP; + } + } + let mut prefixed = if wide.starts_with(&[SEP, SEP, b'?' as u16, SEP]) + || wide.starts_with(&[SEP, b'?' as u16, b'?' as u16, SEP]) + { + wide + } else if wide.starts_with(&[SEP, SEP]) { + // UNC share: \\server\share -> \\?\UNC\server\share + let mut p: Vec = "\\\\?\\UNC\\".encode_utf16().collect(); + p.extend_from_slice(&wide[2..]); + p + } else { + let mut p: Vec = "\\\\?\\".encode_utf16().collect(); + p.extend_from_slice(&wide); + p + }; + prefixed.push(0); + Ok(prefixed) +} + #[cfg(windows)] fn pin_windows_directory(path: &Path) -> io::Result { use std::os::windows::fs::{MetadataExt, OpenOptionsExt}; @@ -190,7 +227,6 @@ pub fn list_regular_files_beneath(root: &Path, max_entries: usize) -> io::Result #[cfg(windows)] fn move_file_write_through(source: &Path, destination: &Path, replace: bool) -> io::Result<()> { - use std::os::windows::ffi::OsStrExt; use windows_sys::Win32::Storage::FileSystem::{ MoveFileExW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, }; @@ -208,16 +244,8 @@ fn move_file_write_through(source: &Path, destination: &Path, replace: bool) -> pin_windows_parent(destination)? }; - let source = source - .as_os_str() - .encode_wide() - .chain(Some(0)) - .collect::>(); - let destination = destination - .as_os_str() - .encode_wide() - .chain(Some(0)) - .collect::>(); + let source = verbatim_wide_null(source)?; + let destination = verbatim_wide_null(destination)?; let flags = MOVEFILE_WRITE_THROUGH | if replace { MOVEFILE_REPLACE_EXISTING diff --git a/crates/epix-node/src/lib.rs b/crates/epix-node/src/lib.rs index d9146bb..b41af79 100644 --- a/crates/epix-node/src/lib.rs +++ b/crates/epix-node/src/lib.rs @@ -2981,6 +2981,10 @@ async fn serve( // that all restored xites are registered, or merger pages show nothing // until some merger action happens to trigger a rebuild. state.rebuild_merger_dbs().await; + // Per-xite dbs are just as boot-empty: warm them in the background so + // the dashboard's first feedQuery finds real rows instead of racing + // every lazy rebuild against its 10s per-feed deadline. + state.spawn_db_warmup(); let transport: Arc = Arc::new(TcpTransport); state.set_transport(transport.clone()).await; diff --git a/crates/epix-ui/src/command.rs b/crates/epix-ui/src/command.rs index eb0d4c4..2ba8378 100644 --- a/crates/epix-ui/src/command.rs +++ b/crates/epix-ui/src/command.rs @@ -2653,6 +2653,7 @@ impl WsCommand for FeedQuery { // All followed feeds run concurrently, each under its own deadline: // one sick xite (db mid-churn, schema outlived by the follow's SQL) // must not serialize-stall the merged view every other xite feeds. + let query_started = std::time::Instant::now(); let results = futures_util::future::join_all(queries.iter().map(|(xite, name, full)| async move { let res = tokio::time::timeout( @@ -2665,8 +2666,10 @@ impl WsCommand for FeedQuery { .await; let mut rows: Vec = Vec::new(); + let mut answered = 0usize; for (xite, name, res) in results { let Ok(Ok(res)) = res else { continue }; + answered += 1; for mut row in res { let Some(obj) = row.as_object_mut() else { continue }; // Normalize + sanity-check date_added (ms -> s; drop future items). @@ -2689,6 +2692,22 @@ impl WsCommand for FeedQuery { let db = b["date_added"].as_f64().unwrap_or(0.0); db.partial_cmp(&da).unwrap_or(std::cmp::Ordering::Equal) }); + // Boot-latency instrumentation: an empty or partial merged feed right + // after start is the symptom users report; name which followed feeds + // answered so the stall (db not ready, query timeout, no follows) is + // attributable from the log. + s.state + .log( + "DEBUG", + format!( + "feedQuery: {} row(s) from {} feed(s) across {num_xites} followed xite(s) ({} answered) in {:.1}s", + rows.len(), + queries.len(), + answered, + query_started.elapsed().as_secs_f32() + ), + ) + .await; // No global cap: `limit` applies per feed query (in build_feed_query), // like EpixNet - a global truncate would let one busy feed crowd every // other xite out of the merged view. diff --git a/crates/epix-ui/src/state.rs b/crates/epix-ui/src/state.rs index f2cd0d1..21d5754 100644 --- a/crates/epix-ui/src/state.rs +++ b/crates/epix-ui/src/state.rs @@ -27408,6 +27408,70 @@ impl AppState { /// - Retry passes notify only on completion (with the xite re-checked to /// still exist - a deleted xite's empty missing-list must not toast /// "Downloaded"); the toggle's own pass already reported any failure. + /// Warm every served xite's in-memory database in the background. Xite + /// dbs live only in RAM (see the startup note on `rebuild_merger_dbs`), + /// so after a restart the dashboard's first feedQuery used to trigger + /// every rebuild lazily - racing the 10s per-feed deadline and the + /// post-boot update churn, which left the feed empty until a later + /// file_done event asked again. Warming right after restore runs the + /// rebuilds before the network loops contend for the per-xite locks. + pub fn spawn_db_warmup(self: &Arc) { + let state = self.clone(); + tokio::spawn(async move { + let started = std::time::Instant::now(); + let mut seen = std::collections::HashSet::new(); + let candidates: Vec = { + let xites = state.xites.read().await; + xites + .iter() + .filter(|(_, xite)| xite.storage.exists("dbschema.json")) + .filter_map(|(key, xite)| { + seen.insert(canonical_address(xite.content.as_ref(), key)) + .then(|| key.clone()) + }) + .collect() + }; + let mut warmed = 0usize; + let mut skipped = 0usize; + for address in &candidates { + // A db already present (a merger rebuild may have filled it) + // needs no warm-up. + if state.current_receipt_for(address).await.is_some() { + skipped += 1; + continue; + } + let one = std::time::Instant::now(); + let built = state.rebuild_xite_db(address).await; + warmed += usize::from(built); + // A slow or refused rebuild is the feed-latency signal this + // warm-up exists to surface - name the xite so it can be + // chased instead of averaged away. + if !built || one.elapsed() > std::time::Duration::from_secs(2) { + state + .log( + "INFO", + format!( + "Db warm-up for {address}: {} in {:.1}s", + if built { "rebuilt" } else { "not rebuilt" }, + one.elapsed().as_secs_f32() + ), + ) + .await; + } + } + state + .log( + "INFO", + format!( + "Db warm-up: {warmed} rebuilt, {skipped} already present, {} total in {:.1}s", + candidates.len(), + started.elapsed().as_secs_f32() + ), + ) + .await; + }); + } + pub fn spawn_optional_retry_loop(self: &Arc) { let state = self.clone(); tokio::spawn(async move { diff --git a/crates/epix-xite/src/storage.rs b/crates/epix-xite/src/storage.rs index 9a13f8d..3ac6c4c 100644 --- a/crates/epix-xite/src/storage.rs +++ b/crates/epix-xite/src/storage.rs @@ -204,7 +204,6 @@ fn list_regular_files_beneath(root: &Path, max_entries: usize) -> std::io::Resul #[cfg(windows)] fn create_directory_durable(directory: &Path) -> std::io::Result<()> { - use std::os::windows::ffi::OsStrExt; use windows_sys::Win32::Storage::FileSystem::{MoveFileExW, MOVEFILE_WRITE_THROUGH}; if directory.is_dir() { @@ -223,16 +222,10 @@ fn create_directory_durable(directory: &Path) -> std::io::Result<()> { std::process::id() )); std::fs::create_dir(&temporary)?; - let source = temporary - .as_os_str() - .encode_wide() - .chain(Some(0)) - .collect::>(); - let destination = directory - .as_os_str() - .encode_wide() - .chain(Some(0)) - .collect::>(); + // Raw MoveFileExW paths are MAX_PATH-limited; the verbatim form keeps + // deep staging trees working. + let source = epix_fs::verbatim_wide_null(&temporary)?; + let destination = epix_fs::verbatim_wide_null(directory)?; let result = unsafe { MoveFileExW(source.as_ptr(), destination.as_ptr(), MOVEFILE_WRITE_THROUGH) }; @@ -449,12 +442,12 @@ fn write_atomic_durable_beneath(root: &Path, inner_path: &str, bytes: &[u8]) -> use std::os::windows::fs::{MetadataExt, OpenOptionsExt}; use windows_sys::Win32::Storage::FileSystem::{ FILE_ATTRIBUTE_REPARSE_POINT, FILE_FLAG_BACKUP_SEMANTICS, - FILE_FLAG_OPEN_REPARSE_POINT, FILE_SHARE_READ, + FILE_FLAG_OPEN_REPARSE_POINT, FILE_SHARE_READ, FILE_SHARE_WRITE, }; let directory = std::fs::OpenOptions::new() .read(true) - .share_mode(FILE_SHARE_READ) + .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE) .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT) .open(path)?; let metadata = directory.metadata()?; @@ -550,6 +543,7 @@ fn write_atomic_durable_beneath(root: &Path, inner_path: &str, bytes: &[u8]) -> )); let temporary_path = current.join(&temporary_name); let mut file = match std::fs::OpenOptions::new() + .write(true) .access_mode(FILE_GENERIC_WRITE | DELETE) .share_mode(FILE_SHARE_READ | FILE_SHARE_DELETE) .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT) @@ -1127,6 +1121,29 @@ mod tests { assert!(leftovers.is_empty(), "temp files left behind: {leftovers:?}"); } + /// The staging tree nests three 64-char hash directories, which pushes + /// absolute paths past Windows' 260-char MAX_PATH. Every durable + /// operation must survive that depth (raw Win32 calls need verbatim + /// paths; std converts its own). + #[test] + fn durable_write_survives_paths_past_max_path() { + let dir = tempfile::tempdir().unwrap(); + let deep = dir + .path() + .join("a".repeat(64)) + .join("b".repeat(64)) + .join("c".repeat(64)); + let storage = XiteStorage::new(&deep); + let inner = "backups/replaced/data/users/someone.epix/content.json"; + assert!(deep.join(inner).as_os_str().len() > 260, "test path must exceed MAX_PATH"); + + storage.write_atomic_durable(inner, b"deep v1").unwrap(); + storage.write_atomic_durable(inner, b"deep v2").unwrap(); + assert_eq!(storage.read(inner).unwrap(), b"deep v2"); + storage.delete(inner).unwrap(); + assert!(!storage.exists(inner)); + } + #[test] fn durable_atomic_write_replaces_without_leaving_a_sibling() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/epix-xite/src/xite.rs b/crates/epix-xite/src/xite.rs index 1a4218e..5b0b5e0 100644 --- a/crates/epix-xite/src/xite.rs +++ b/crates/epix-xite/src/xite.rs @@ -2678,11 +2678,17 @@ mod tests { let mut swapped = movie.clone(); swapped[0] = 1; storage.write("video/movie.bin", &swapped).unwrap(); - let f = std::fs::File::options() - .append(true) - .open(storage.path("video/movie.bin").unwrap()) - .unwrap(); - f.set_modified(mtime).unwrap(); + // The touch handle must not stay open across a sign: the Windows read + // path denies write sharing, so a held append handle fails the read. + let touch = |time: std::time::SystemTime| { + std::fs::File::options() + .append(true) + .open(storage.path("video/movie.bin").unwrap()) + .unwrap() + .set_modified(time) + .unwrap(); + }; + touch(mtime); xite.sign(&pk, 1001.0).unwrap(); let second = xite.content.clone().unwrap(); assert_eq!( @@ -2702,7 +2708,7 @@ mod tests { ); // Now touch the mtime: a normal sign re-reads changed files by itself. - f.set_modified(std::time::SystemTime::now() + std::time::Duration::from_secs(5)).unwrap(); + touch(std::time::SystemTime::now() + std::time::Duration::from_secs(5)); xite.sign(&pk, 1003.0).unwrap(); let third = xite.content.clone().unwrap(); assert_ne!(