diff --git a/CHANGELOG.md b/CHANGELOG.md index e0d7169..2891b68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Removed ### Security +## [2.2.0] - 2026-09-28 + +### Fixed + +- Stack overflow when filtering out many entries, e.g. on tokio worker threads (#13). + +### Changed + +- Directories filtered with `Filtering::IgnoreDir` are no longer read, so they + no longer produce IO errors (e.g. permission denied). +- Update dependencies: `futures-lite` 2.6, `async-fs` 2.2. + ## [2.1.0] - 2025-01-27 ### Added diff --git a/Cargo.toml b/Cargo.toml index 0b1e82b..43ea76d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "async-walkdir" -version = "2.1.0" +version = "2.2.0" authors = ["Ririsoft "] edition = "2021" description = "Asynchronous directory traversal for Rust." @@ -14,9 +14,10 @@ exclude = [ "/.github/*" ] readme = "README.md" [dependencies] -futures-lite = "2.1" -async-fs = "2.1" +futures-lite = "2.6" +async-fs = "2.2" thiserror = "2.0" [dev-dependencies] -tempfile = "3.15" +tempfile = "3.27" +tokio = { version = "1.53", features = ["rt-multi-thread"] } diff --git a/src/lib.rs b/src/lib.rs index 2c1e2ad..97b4a0f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -90,7 +90,6 @@ use std::task::{Context, Poll}; use async_fs::{read_dir, ReadDir}; use futures_lite::future::Boxed as BoxedFut; -use futures_lite::future::FutureExt; use futures_lite::stream::{self, Stream, StreamExt}; #[doc(no_inline)] @@ -197,85 +196,55 @@ enum State { type UnfoldState = (Result, State); -fn walk( +// Iterative on purpose: filtered-out entries must not grow the stack (see issue #13). +async fn walk( mut dirs: Vec<(PathBuf, ReadDir)>, - filter: Option, -) -> BoxedFut>> + mut filter: Option, +) -> Option> where F: FnMut(DirEntry) -> Fut + Send + 'static, Fut: Future + Send, { - async move { - if let Some((path, dir)) = dirs.last_mut() { - match dir.next().await { - Some(Ok(entry)) => walk_entry(entry, dirs, filter).await, - Some(Err(source)) => Some(( - Err(InnerError::Io { - path: path.to_path_buf(), - source, - } - .into()), - State::Walk((dirs, filter)), - )), - None => { - dirs.pop(); - walk(dirs, filter).await - } + loop { + let (path, dir) = dirs.last_mut()?; + let entry = match dir.next().await { + Some(Ok(entry)) => entry, + Some(Err(source)) => return io_error(path.to_path_buf(), source, dirs, filter), + None => { + dirs.pop(); + continue; } - } else { - None + }; + let ft = match entry.file_type().await { + Ok(ft) => ft, + Err(source) => return io_error(entry.path(), source, dirs, filter), + }; + let filtering = match filter.as_mut() { + Some(filter) => filter(entry.clone()).await, + None => Filtering::Continue, + }; + if ft.is_dir() && filtering != Filtering::IgnoreDir { + let path = entry.path(); + let rd = match read_dir(&path).await { + Ok(rd) => rd, + Err(source) => return io_error(path, source, dirs, filter), + }; + dirs.push((path, rd)); + } + if filtering == Filtering::Continue { + return Some((Ok(entry), State::Walk((dirs, filter)))); } } - .boxed() } -fn walk_entry( - entry: DirEntry, - mut dirs: Vec<(PathBuf, ReadDir)>, - mut filter: Option, -) -> BoxedFut>> -where - F: FnMut(DirEntry) -> Fut + Send + 'static, - Fut: Future + Send, -{ - async move { - match entry.file_type().await { - Err(source) => Some(( - Err(InnerError::Io { - path: entry.path(), - source, - } - .into()), - State::Walk((dirs, filter)), - )), - Ok(ft) => { - let filtering = match filter.as_mut() { - Some(filter) => filter(entry.clone()).await, - None => Filtering::Continue, - }; - if ft.is_dir() { - let path = entry.path(); - let rd = match read_dir(&path).await { - Err(source) => { - return Some(( - Err(InnerError::Io { path, source }.into()), - State::Walk((dirs, filter)), - )) - } - Ok(rd) => rd, - }; - if filtering != Filtering::IgnoreDir { - dirs.push((path, rd)); - } - } - match filtering { - Filtering::Continue => Some((Ok(entry), State::Walk((dirs, filter)))), - Filtering::IgnoreDir | Filtering::Ignore => walk(dirs, filter).await, - } - } - } - } - .boxed() +fn io_error( + path: PathBuf, + source: std::io::Error, + dirs: Vec<(PathBuf, ReadDir)>, + filter: Option, +) -> Option> { + let err = InnerError::Io { path, source }.into(); + Some((Err(err), State::Walk((dirs, filter)))) } #[cfg(test)] @@ -309,6 +278,30 @@ mod tests { } _ => panic!("want IO error"), } + assert!(wd.next().await.is_none()); + }) + } + + #[test] + fn walk_dir_read_dir_error() -> Result<()> { + block_on(async { + let root = tempfile::tempdir()?; + let d1 = root.path().join("d1"); + async_fs::create_dir_all(&d1).await?; + // Removing the directory from the filter makes the subsequent read_dir fail. + let mut wd = WalkDir::new(root.path()).filter(|entry| async move { + async_fs::remove_dir(entry.path()).await.unwrap(); + Filtering::Continue + }); + match wd.next().await.unwrap() { + Err(e) => { + assert_eq!(e.path().unwrap(), d1.as_path()); + assert_eq!(e.io().unwrap().kind(), ErrorKind::NotFound); + } + _ => panic!("want IO error"), + } + assert!(wd.next().await.is_none()); + Ok(()) }) } @@ -447,7 +440,23 @@ mod test_unix { use futures_lite::future::block_on; use futures_lite::stream::StreamExt; - use super::WalkDir; + use super::{Filtering, WalkDir}; + + #[test] + fn filter_ignore_dir_does_not_read_it() -> Result<()> { + block_on(async { + let root = tempfile::tempdir()?; + let d1 = root.path().join("d1"); + async_fs::create_dir_all(&d1).await?; + let mut perms = async_fs::metadata(&d1).await?.permissions(); + perms.set_mode(0o222); + async_fs::set_permissions(&d1, perms).await?; + let mut wd = WalkDir::new(&root).filter(|_| async { Filtering::IgnoreDir }); + assert!(wd.next().await.is_none()); + Ok(()) + }) + } + #[test] fn walk_dir_error_path() -> Result<()> { block_on(async { diff --git a/tests/issue_13_stack_overflow.rs b/tests/issue_13_stack_overflow.rs new file mode 100644 index 0000000..459d085 --- /dev/null +++ b/tests/issue_13_stack_overflow.rs @@ -0,0 +1,101 @@ +// Regression tests for https://github.com/ririsoft/async-walkdir/issues/13 +// +// Filtering out many consecutive entries used to overflow the stack of small-stack +// threads (such as tokio workers) because the traversal recursed once per ignored entry. +// A stack overflow aborts the whole process, hence these tests live in their own binary. + +use std::path::{Path, PathBuf}; + +use async_walkdir::{Filtering, WalkDir}; +use futures_lite::stream::StreamExt; + +// The old recursive code overflows this stack after ~1000-2000 ignored entries in release +// (fewer in debug), so 4000 gives margin. The stack is large enough for the fixed code's +// constant usage on every CI platform (64 KiB was too small on Windows debug builds). +const IGNORED_ENTRIES: usize = 4_000; +const MATCHING_FILES: usize = 5; +const STACK_SIZE: usize = 512 * 1024; + +fn run_on_small_stack(fut: F) -> T +where + F: std::future::Future + Send + 'static, + T: Send + 'static, +{ + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_stack_size(STACK_SIZE) + .build() + .unwrap(); + runtime.block_on(async { tokio::spawn(fut).await.unwrap() }) +} + +fn sorted(mut v: Vec) -> Vec { + v.sort(); + v +} + +async fn collect(mut wd: WalkDir) -> Vec { + let mut got = Vec::new(); + while let Some(entry) = wd.next().await { + got.push(entry.unwrap().path()); + } + got +} + +fn is_m3u8(path: &Path) -> bool { + path.extension().is_some_and(|ext| ext == "m3u8") +} + +#[test] +fn filter_ignore_many_files() -> std::io::Result<()> { + let root = tempfile::tempdir()?; + // A sub directory ensures popping exhausted directories is covered too. + let sub = root.path().join("sub"); + std::fs::create_dir(&sub)?; + let mut want = Vec::new(); + for i in 0..IGNORED_ENTRIES { + let dir = if i % 2 == 0 { root.path() } else { &sub }; + std::fs::write(dir.join(format!("segment_{i}.ts")), [])?; + } + for i in 0..MATCHING_FILES { + let f = root.path().join(format!("playlist_{i}.m3u8")); + std::fs::write(&f, [])?; + want.push(f); + } + + let wd = WalkDir::new(root.path()).filter(|entry| async move { + let path = entry.path(); + if path.is_file() && is_m3u8(&path) { + return Filtering::Continue; + } + Filtering::Ignore + }); + let got = run_on_small_stack(collect(wd)); + assert_eq!(sorted(got), sorted(want)); + Ok(()) +} + +#[test] +fn filter_ignore_dir_many_dirs() -> std::io::Result<()> { + let root = tempfile::tempdir()?; + for i in 0..IGNORED_ENTRIES { + let d = root.path().join(format!("ignored_{i}")); + std::fs::create_dir(&d)?; + std::fs::write(d.join("f.m3u8"), [])?; + } + let kept = root.path().join("kept"); + std::fs::create_dir(&kept)?; + let f = kept.join("f.m3u8"); + std::fs::write(&f, [])?; + let want = vec![kept, f]; + + let wd = WalkDir::new(root.path()).filter(|entry| async move { + if entry.file_name().to_string_lossy().starts_with("ignored_") { + return Filtering::IgnoreDir; + } + Filtering::Continue + }); + let got = run_on_small_stack(collect(wd)); + assert_eq!(sorted(got), sorted(want)); + Ok(()) +}