scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix: migrate blocking fs calls off Tokio paths

Browse source 
Phase 1 of the std::fs lint initiative. Refactors blocking std::fs entry
points that ran inside async contexts. Caller chains either converted to
async (using tokio::fs) or wrapped in tokio::task::spawn_blocking where
sync callers were already natural (Command builders, flock semantics).

HIGH (per-request async hot paths):
  - fabro-sandbox local.rs: wrap recursive std::fs::read_dir traversal in
    spawn_blocking. Fixes /api/runs/{id}/files stalling workers under
    concurrent or deep listings.
  - fabro-server static_files.rs: convert serve/serve_install/serve_with_mode
    and the static-asset load chain to async; use tokio::fs::read for the
    debug-only disk fallback. Cascades through install.rs build_install_router
    (now async) and ~17 test call sites.

LOW (async but not per-request):
  - fabro-workflow artifact.rs: sync_artifacts_to_env, offload_large_values
    → tokio::fs::read_to_string.
  - fabro-workflow artifact_snapshot.rs: compute_artifact_info → async +
    tokio::fs::read.
  - fabro-server ip_allowlist.rs: load_cache and store_cache → async +
    tokio::fs::{read,write,create_dir_all}.
  - fabro-server server.rs: wrap worker_command invocation in spawn_blocking
    at the async boundary in execute_run_subprocess; keep the sync
    worker_command + current_server_target signatures intact.
  - fabro-cli server/start.rs: wrap the OpenOptions::open call in
    acquire_lock in spawn_blocking; file-lock semantics require a real
    std::fs::File, and the flock polling loop stays async with time::sleep.

Deferred:
  - fabro-llm load_file_as_base64 (file:// attachment loader): 7 call sites
    across 4 providers, each inside sync translators. Left for Phase 3
    annotation with a FOLLOW-UP marker; file:// URLs are rare in practice.

Verified: workspace builds, 4131 tests pass, 182 skipped. The lint that
enforces this discipline lands in the next commit.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-19 17:00:56 -04:00
parent 95b101a26f
commit 9d1c0d98c7
No known key found for this signature in database
11 changed files with 118 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
lib/crates/fabro-cli/src/commands/server/start.rs
View file

@ -513,12 +513,17 @@ async fn acquire_lock(storage_dir: &Path) -> Result<std::fs::File> {
std::fs::create_dir_all(parent)
.with_context(|| format!("creating server lock directory {}", parent.display()))?;
}
let lock_file = std::fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(false)
.open(&lock_path)
.with_context(|| format!("opening server lock file {}", lock_path.display()))?;
let lock_path_for_open = lock_path.clone();
let lock_file = tokio::task::spawn_blocking(move || {
std::fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(false)
.open(&lock_path_for_open)
})
.await
.context("lock-file open task failed")?
.with_context(|| format!("opening server lock file {}", lock_path.display()))?;
let poll_interval = Duration::from_millis(50);
let timeout = Duration::from_secs(5);

10
lib/crates/fabro-sandbox/src/local.rs
View file

@ -197,9 +197,13 @@ impl Sandbox for LocalSandbox {
let full_path = self.resolve_path(path);
let max_depth = depth.unwrap_or(1);
let mut entries = Vec::new();
list_recursive(&full_path, "", 0, max_depth, &mut entries)?;
Ok(entries)
tokio::task::spawn_blocking(move || {
let mut entries = Vec::new();
list_recursive(&full_path, "", 0, max_depth, &mut entries)?;
Ok(entries)
})
.await
.map_err(|e| format!("list_directory task failed: {e}"))?
}
async fn exec_command(

10
lib/crates/fabro-server/src/install.rs
View file

@ -310,8 +310,8 @@ impl TryFrom<GithubAppOwnerInput> for GitHubAppOwner {
}
}
pub fn build_install_router(state: InstallAppState) -> Router {
static_files::assert_install_mode_shell_ready();
pub async fn build_install_router(state: InstallAppState) -> Router {
static_files::assert_install_mode_shell_ready().await;
Router::new()
.route("/health", get(health))
@ -346,7 +346,7 @@ pub fn build_install_router(state: InstallAppState) -> Router {
Ok::<_, Infallible>(StatusCode::NOT_FOUND.into_response())
} else if matches!(req.method(), &Method::GET | &Method::HEAD) {
let headers = req.headers().clone();
Ok::<_, Infallible>(static_files::serve_install(&path, &headers))
Ok::<_, Infallible>(static_files::serve_install(&path, &headers).await)
} else {
Ok::<_, Infallible>(StatusCode::NOT_FOUND.into_response())
}
@ -397,7 +397,7 @@ where
let bound_listener = bind_install_listener(&bind_request).await?;
state.set_install_bind(&bound_listener.bind);
let state = state.with_finish_callback(finish_callback);
let router = build_install_router(state);
let router = build_install_router(state).await;
let bind = bound_listener.bind.clone();
on_ready(&bind)?;
@ -956,7 +956,7 @@ async fn post_install_finish(
}
async fn render_install_shell(headers: HeaderMap, uri: OriginalUri) -> Response {
static_files::serve_install(uri.path(), &headers)
static_files::serve_install(uri.path(), &headers).await
}
fn token_is_valid(state: &InstallAppState, headers: &HeaderMap, query_token: Option<&str>) -> bool {

17
lib/crates/fabro-server/src/ip_allowlist.rs
View file

@ -69,7 +69,7 @@ impl GitHubMetaResolver {
}
async fn resolve_hooks(&self) -> Result<Vec<IpNet>> {
let cached = self.load_cache()?;
let cached = self.load_cache().await?;
let mut request = self
.client
.get(&self.meta_url)
@ -124,12 +124,13 @@ impl GitHubMetaResolver {
self.store_cache(&GitHubMetaCache {
etag,
hooks: payload.hooks,
})?;
})
.await?;
Ok(hooks)
}
fn load_cache(&self) -> Result<Option<GitHubMetaCache>> {
match std::fs::read(&self.cache_path) {
async fn load_cache(&self) -> Result<Option<GitHubMetaCache>> {
match tokio::fs::read(&self.cache_path).await {
Ok(contents) => match serde_json::from_slice(&contents) {
Ok(cache) => Ok(Some(cache)),
Err(error) => {
@ -148,14 +149,16 @@ impl GitHubMetaResolver {
}
}
fn store_cache(&self, cache: &GitHubMetaCache) -> Result<()> {
async fn store_cache(&self, cache: &GitHubMetaCache) -> Result<()> {
if let Some(parent) = self.cache_path.parent() {
std::fs::create_dir_all(parent)
tokio::fs::create_dir_all(parent)
.await
.with_context(|| format!("creating {}", parent.display()))?;
}
let contents = serde_json::to_vec(cache).context("serializing GitHub meta cache")?;
std::fs::write(&self.cache_path, contents)
tokio::fs::write(&self.cache_path, contents)
.await
.with_context(|| format!("writing {}", self.cache_path.display()))?;
Ok(())
}

18
lib/crates/fabro-server/src/server.rs
View file

@ -980,7 +980,7 @@ pub fn build_router_with_options(
&& matches!(req.method(), &Method::GET | &Method::HEAD)
{
let headers = req.headers().clone();
Ok::<_, std::convert::Infallible>(static_files::serve(&path, &headers))
Ok::<_, std::convert::Infallible>(static_files::serve(&path, &headers).await)
} else {
Ok::<_, std::convert::Infallible>(StatusCode::NOT_FOUND.into_response())
}
@ -4460,7 +4460,21 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
run_store.subscribe(),
));
let mut child = match worker_command(state.as_ref(), run_id, execution_mode, &run_dir)
let state_for_build = Arc::clone(&state);
let run_dir_for_build = run_dir.clone();
let build_cmd_result = tokio::task::spawn_blocking(move || {
worker_command(
state_for_build.as_ref(),
run_id,
execution_mode,
&run_dir_for_build,
)
})
.await;
let mut child = match build_cmd_result
.map_err(|err| anyhow::anyhow!("worker_command task failed: {err}"))
.and_then(|inner| inner)
.and_then(|mut cmd| cmd.spawn().context("spawning run worker process"))
{
Ok(child) => child,

65
lib/crates/fabro-server/src/static_files.rs
View file

@ -7,18 +7,21 @@ use axum::response::{IntoResponse, Response};
const INSTALL_MODE_MARKER: &str = "__FABRO_MODE__ = \"install\"";
pub fn serve(path: &str, headers: &HeaderMap) -> Response {
serve_with_mode(path, headers, SpaMode::Normal)
pub async fn serve(path: &str, headers: &HeaderMap) -> Response {
serve_with_mode(path, headers, SpaMode::Normal).await
}
pub fn serve_install(path: &str, headers: &HeaderMap) -> Response {
serve_with_mode(path, headers, SpaMode::Install)
pub async fn serve_install(path: &str, headers: &HeaderMap) -> Response {
serve_with_mode(path, headers, SpaMode::Install).await
}
pub(crate) fn assert_install_mode_shell_ready() {
let shell = cached_install_mode_shell().clone().unwrap_or_else(|| {
load_injected_install_shell().expect("install-mode SPA shell asset missing")
});
pub(crate) async fn assert_install_mode_shell_ready() {
let shell = match cached_install_mode_shell().await {
Some(shell) => shell,
None => load_injected_install_shell()
.await
.expect("install-mode SPA shell asset missing"),
};
let html = String::from_utf8(shell).expect("install-mode SPA shell must be valid UTF-8");
assert!(
html.contains(INSTALL_MODE_MARKER),
@ -26,17 +29,21 @@ pub(crate) fn assert_install_mode_shell_ready() {
);
}
fn cached_install_mode_shell() -> Option<Vec<u8>> {
async fn cached_install_mode_shell() -> Option<Vec<u8>> {
static SHELL: OnceLock<Option<Vec<u8>>> = OnceLock::new();
if cfg!(debug_assertions) {
// In debug builds the SPA is reloaded from disk on every request.
return load_injected_install_shell();
return load_injected_install_shell().await;
}
SHELL.get_or_init(load_injected_install_shell).clone()
if let Some(cached) = SHELL.get() {
return cached.clone();
}
let loaded = load_injected_install_shell().await;
SHELL.get_or_init(|| loaded).clone()
}
fn load_injected_install_shell() -> Option<Vec<u8>> {
Some(inject_install_mode(load_asset("index.html")?))
async fn load_injected_install_shell() -> Option<Vec<u8>> {
Some(inject_install_mode(load_asset("index.html").await?))
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
@ -45,14 +52,14 @@ enum SpaMode {
Install,
}
fn serve_with_mode(path: &str, headers: &HeaderMap, mode: SpaMode) -> Response {
async fn serve_with_mode(path: &str, headers: &HeaderMap, mode: SpaMode) -> Response {
let normalized = normalize(path);
if is_source_map(&normalized) {
return (StatusCode::NOT_FOUND, "Static asset not found").into_response();
}
if let Some(asset) = load_asset_for_mode(&normalized, mode) {
if let Some(asset) = load_asset_for_mode(&normalized, mode).await {
return asset_response(&normalized, asset);
}
@ -61,7 +68,7 @@ fn serve_with_mode(path: &str, headers: &HeaderMap, mode: SpaMode) -> Response {
// `Accept: */*`, and similar non-HTML clients get a 404 so typos
// don't silently return 25KB of UI shell.
if accepts_html(headers) {
if let Some(index) = load_asset_for_mode("index.html", mode) {
if let Some(index) = load_asset_for_mode("index.html", mode).await {
return asset_response("index.html", index);
}
}
@ -92,9 +99,9 @@ fn normalize(path: &str) -> String {
}
}
fn load_asset(path: &str) -> Option<Vec<u8>> {
async fn load_asset(path: &str) -> Option<Vec<u8>> {
if cfg!(debug_assertions) {
if let Some(bytes) = read_disk_asset(path) {
if let Some(bytes) = read_disk_asset(path).await {
return Some(bytes);
}
}
@ -102,11 +109,11 @@ fn load_asset(path: &str) -> Option<Vec<u8>> {
fabro_spa::get(path).map(fabro_spa::AssetBytes::into_vec)
}
fn load_asset_for_mode(path: &str, mode: SpaMode) -> Option<Vec<u8>> {
async fn load_asset_for_mode(path: &str, mode: SpaMode) -> Option<Vec<u8>> {
if mode == SpaMode::Install && path == "index.html" {
return cached_install_mode_shell();
return cached_install_mode_shell().await;
}
load_asset(path)
load_asset(path).await
}
fn inject_install_mode(bytes: Vec<u8>) -> Vec<u8> {
@ -129,14 +136,14 @@ fn inject_install_mode(bytes: Vec<u8>) -> Vec<u8> {
injected.into_bytes()
}
fn read_disk_asset(path: &str) -> Option<Vec<u8>> {
read_disk_asset_from_root(&disk_asset_root(), path)
async fn read_disk_asset(path: &str) -> Option<Vec<u8>> {
read_disk_asset_from_root(&disk_asset_root(), path).await
}
fn read_disk_asset_from_root(root: &Path, path: &str) -> Option<Vec<u8>> {
async fn read_disk_asset_from_root(root: &Path, path: &str) -> Option<Vec<u8>> {
let candidate = root.join(path);
if candidate.is_file() {
std::fs::read(candidate).ok()
tokio::fs::read(candidate).await.ok()
} else {
None
}
@ -236,14 +243,16 @@ mod tests {
assert_eq!(cache_control("index.html"), "no-cache");
}
#[test]
fn disk_assets_are_loaded_from_explicit_root() {
#[tokio::test]
async fn disk_assets_are_loaded_from_explicit_root() {
let temp_dir = tempfile::tempdir().unwrap();
let asset_path = temp_dir.path().join("assets/override.txt");
std::fs::create_dir_all(asset_path.parent().unwrap()).unwrap();
std::fs::write(&asset_path, b"override").unwrap();
let bytes = read_disk_asset_from_root(temp_dir.path(), "assets/override.txt").unwrap();
let bytes = read_disk_asset_from_root(temp_dir.path(), "assets/override.txt")
.await
.unwrap();
assert_eq!(bytes, b"override");
}

36
lib/crates/fabro-server/tests/it/api/install.rs
View file

@ -70,7 +70,7 @@ async fn configure_token_install(app: &axum::Router, token: &str) {
#[tokio::test]
async fn install_router_isolated_from_normal_api_surface() {
let app = build_install_router(InstallAppState::for_test("test-install-token"));
let app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let health_response = app
.clone()
@ -128,7 +128,7 @@ async fn install_router_isolated_from_normal_api_surface() {
#[tokio::test]
async fn install_session_requires_valid_install_token() {
let app = build_install_router(InstallAppState::for_test("test-install-token"));
let app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let unauthorized = app
.clone()
@ -166,7 +166,7 @@ async fn install_session_requires_valid_install_token() {
#[tokio::test]
async fn install_endpoints_reject_missing_and_wrong_tokens() {
let app = build_install_router(InstallAppState::for_test("test-install-token"));
let app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let cases = [
("GET", "/install/session", None),
(
@ -249,7 +249,7 @@ async fn install_endpoints_reject_missing_and_wrong_tokens() {
#[tokio::test]
async fn install_endpoints_accept_query_token_when_authorization_header_is_wrong() {
let app = build_install_router(InstallAppState::for_test("test-install-token"));
let app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let response = app
.oneshot(
@ -274,7 +274,8 @@ async fn token_install_finish_persists_settings_env_and_vault() {
"test-install-token",
temp_dir.path(),
&config_path,
));
))
.await;
let llm_response = app
.clone()
@ -394,7 +395,8 @@ async fn token_install_finish_invokes_shutdown_callback_after_accepting() {
.with_finish_callback(Arc::new(move || {
callback_flag.store(true, Ordering::Release);
})),
);
)
.await;
configure_token_install(&app, "test-install-token").await;
@ -451,7 +453,8 @@ async fn install_validation_endpoints_validate_credentials_and_github_token() {
InstallAppState::for_test("test-install-token")
.with_provider_base_url(Provider::Anthropic, format!("{}/v1", llm_mock.url("")))
.with_github_api_base_url(github_mock.url("")),
);
)
.await;
let llm_response = app
.clone()
@ -513,7 +516,8 @@ async fn github_app_manifest_round_trip_updates_install_session() {
let app = build_install_router(
InstallAppState::for_test("test-install-token")
.with_github_api_base_url(github_mock.url("")),
);
)
.await;
let server_response = app
.clone()
@ -618,7 +622,7 @@ async fn github_app_manifest_round_trip_updates_install_session() {
#[tokio::test]
async fn github_app_manifest_rejects_retry_while_pending_and_preserves_prior_token_strategy() {
let app = build_install_router(InstallAppState::for_test("test-install-token"));
let app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let server_response = app
.clone()
@ -741,7 +745,8 @@ async fn github_app_redirect_rejects_invalid_or_missing_state_without_mutating_s
let app = build_install_router(
InstallAppState::for_test("test-install-token")
.with_github_api_base_url(github_mock.url("")),
);
)
.await;
let server_response = app
.clone()
@ -881,7 +886,8 @@ async fn github_app_redirect_exchange_failure_returns_to_wizard_and_keeps_pendin
let app = build_install_router(
InstallAppState::for_test("test-install-token")
.with_github_api_base_url(github_mock.url("")),
);
)
.await;
let server_response = app
.clone()
@ -970,7 +976,7 @@ async fn github_app_redirect_exchange_failure_returns_to_wizard_and_keeps_pendin
#[tokio::test]
async fn install_server_rejects_trailing_slash_canonical_urls() {
let app = build_install_router(InstallAppState::for_test("test-install-token"));
let app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let response = app
.oneshot(
@ -1015,7 +1021,8 @@ async fn install_finish_failure_restores_settings_and_vault_but_leaves_env_keys(
.with_finish_callback(Arc::new(move || {
callback_flag.store(true, Ordering::Release);
})),
);
)
.await;
configure_token_install(&app, "test-install-token").await;
@ -1104,7 +1111,8 @@ async fn install_finish_failure_leaves_home_dev_token_mirror_written() {
let app = build_install_router(
InstallAppState::for_test_with_paths("test-install-token", temp_dir.path(), &config_path)
.with_home(home.clone()),
);
)
.await;
configure_token_install(&app, "test-install-token").await;

2
lib/crates/fabro-server/tests/it/api/install_openai_compatible.rs
View file

@ -7,7 +7,7 @@ use crate::helpers::body_json;
#[tokio::test]
async fn install_llm_endpoints_reject_openai_compatible_in_v1() {
let app = build_install_router(InstallAppState::for_test("test-install-token"));
let app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let test_response = app
.clone()

4
lib/crates/fabro-server/tests/it/openapi_conformance.rs
View file

@ -83,7 +83,7 @@ fn request_for(method: &Method, uri: &str) -> Request<Body> {
async fn all_spec_routes_are_routable() {
let spec = load_spec();
let normal_app = build_router(test_app_state(), AuthMode::Disabled);
let install_app = build_install_router(InstallAppState::for_test("test-install-token"));
let install_app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let paths = spec
.get("paths")
@ -122,7 +122,7 @@ async fn all_spec_routes_are_routable() {
async fn install_and_normal_routes_stay_isolated() {
let spec = load_spec();
let normal_app = build_router(test_app_state(), AuthMode::Disabled);
let install_app = build_install_router(InstallAppState::for_test("test-install-token"));
let install_app = build_install_router(InstallAppState::for_test("test-install-token")).await;
let paths = spec
.get("paths")

5
lib/crates/fabro-workflow/src/artifact.rs
View file

@ -174,7 +174,7 @@ pub async fn sync_artifacts_to_env(
}
}
let content = std::fs::read_to_string(&local_path).map_err(|e| {
let content = tokio::fs::read_to_string(&local_path).await.map_err(|e| {
Error::engine(format!("failed to read local artifact {local_path}: {e}"))
})?;
@ -325,7 +325,8 @@ async fn resolve_explicit_file_ref(value: &str, env: &dyn Sandbox) -> Result<Str
return Ok(value.to_string());
}
let content = std::fs::read_to_string(local_path)
let content = tokio::fs::read_to_string(local_path)
.await
.map_err(|e| Error::engine(format!("failed to read local artifact {local_path}: {e}")))?;
let filename = Path::new(local_path)
.file_name()

7
lib/crates/fabro-workflow/src/artifact_snapshot.rs
View file

@ -257,14 +257,15 @@ fn normalize_paths(discovered: Vec<DiscoveredFile>, root: &str) -> Vec<Discovere
.collect()
}
fn compute_artifact_info(
async fn compute_artifact_info(
relative_path: &str,
local_path: &Path,
) -> std::result::Result<CapturedArtifactInfo, String> {
let mime = mime_guess::from_path(relative_path)
.first_or_octet_stream()
.to_string();
let data = std::fs::read(local_path)
let data = tokio::fs::read(local_path)
.await
.map_err(|e| format!("failed to read {}: {e}", local_path.display()))?;
let bytes = u64::try_from(data.len()).unwrap_or(u64::MAX);
let content_md5 = format!("{:x}", md5::compute(&data));
@ -314,7 +315,7 @@ pub async fn collect_artifacts(
.download_file_to_local(&file.relative_path, &dest)
.await
{
Ok(()) => match compute_artifact_info(&file.relative_path, &dest) {
Ok(()) => match compute_artifact_info(&file.relative_path, &dest).await {
Ok(info) => {
files_copied += 1;
total_bytes += info.bytes;