Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
9 changes: 5 additions & 4 deletions Cargo.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "async-walkdir"
version = "2.1.0"
version = "2.2.0"
authors = ["Ririsoft <riri@ririsoft.com>"]
edition = "2021"
description = "Asynchronous directory traversal for Rust."
Expand All @@ -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"] }
151 changes: 80 additions & 71 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)]
Expand Down Expand Up @@ -197,85 +196,55 @@ enum State<F> {

type UnfoldState<F> = (Result<DirEntry>, State<F>);

fn walk<F, Fut>(
// Iterative on purpose: filtered-out entries must not grow the stack (see issue #13).
async fn walk<F, Fut>(
mut dirs: Vec<(PathBuf, ReadDir)>,
filter: Option<F>,
) -> BoxedFut<Option<UnfoldState<F>>>
mut filter: Option<F>,
) -> Option<UnfoldState<F>>
where
F: FnMut(DirEntry) -> Fut + Send + 'static,
Fut: Future<Output = Filtering> + 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<F, Fut>(
entry: DirEntry,
mut dirs: Vec<(PathBuf, ReadDir)>,
mut filter: Option<F>,
) -> BoxedFut<Option<UnfoldState<F>>>
where
F: FnMut(DirEntry) -> Fut + Send + 'static,
Fut: Future<Output = Filtering> + 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<F>(
path: PathBuf,
source: std::io::Error,
dirs: Vec<(PathBuf, ReadDir)>,
filter: Option<F>,
) -> Option<UnfoldState<F>> {
let err = InnerError::Io { path, source }.into();
Some((Err(err), State::Walk((dirs, filter))))
}

#[cfg(test)]
Expand Down Expand Up @@ -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(())
})
}

Expand Down Expand Up @@ -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 {
Expand Down
101 changes: 101 additions & 0 deletions tests/issue_13_stack_overflow.rs
Original file line number Diff line number Diff line change
@@ -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<F, T>(fut: F) -> T
where
F: std::future::Future<Output = T> + 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<PathBuf>) -> Vec<PathBuf> {
v.sort();
v
}

async fn collect(mut wd: WalkDir) -> Vec<PathBuf> {
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(())
}
Loading