diff --git a/developer-docs/zh-CN/records/README.md b/developer-docs/zh-CN/records/README.md index 294ffe213..8e0fbe42f 100644 --- a/developer-docs/zh-CN/records/README.md +++ b/developer-docs/zh-CN/records/README.md @@ -4,6 +4,7 @@ | 文档 | 状态 | 阅读目的 | | --- | --- | --- | +| [Issue #486 Future 体积与分配证据](./issue-486-future-size-evidence.md) | 历史快照 | 保存归档、上传和 WebDAV 的改前改后 Future 体积、分配数据与 lint 分类 | | [静态配置密钥处理备忘](./static-config-secret-handling.md) | 草稿 | 评估敏感配置的脱敏、内存驻留和 `SecretString` 取舍 | | [服务层模块化重构历史方案](./service-modularization-refactor-plan.md) | 历史快照 | 保存服务目录迁移前的分析和仍然有效的边界建议 | | [用户文档 IA 重构迁移地图](./user-docs-ia-migration-map.md) | 执行中 | issue #435 的页面处置、概念权威页、重定向规则和中英同步策略，随各批次更新 | diff --git a/developer-docs/zh-CN/records/issue-486-future-size-evidence.md b/developer-docs/zh-CN/records/issue-486-future-size-evidence.md new file mode 100644 index 000000000..62f509a95 --- /dev/null +++ b/developer-docs/zh-CN/records/issue-486-future-size-evidence.md @@ -0,0 +1,83 @@ +# Issue #486 Future 体积与分配证据 + +> 状态：历史快照。本文记录基线 revision `9b55dbfd18892e09227bfb0ff5de950586071dd5` +> 与 issue #486 实现分支在 2026-08-08 的聚焦测量，不作为当前实现规范。当前行为以相关源码、测试和 +> `clippy::large_futures` 输出为准。 + +## 测量环境 + +- `rustc 1.97.1 (8bab26f4f 2026-07-14)`，LLVM `22.1.6` +- `cargo 1.97.1 (c980f4866 2026-06-30)` +- `x86_64-unknown-linux-gnu`，Linux `6.12.85+deb13-amd64` +- 基线与改后 checkout 共享同一 `target`，分别重新编译 AsterDrive crate +- Future 体积来自同一条全 feature Clippy 命令；归档 copy helper 的精确体积来自 + `std::mem::size_of_val` +- 分配数据由线程局部启用的 `System` counting allocator 记录；输入、状态构造、认证 warm-up 和 + 5 MiB 上传 `Bytes` 均在测量区间外创建 + +## Future 体积 + +默认 16 KiB 阈值命令： + +```bash +RUSTC_WRAPPER= cargo clippy -p aster_drive --lib --all-features \ + --message-format=short -- \ + -W clippy::unused_async \ + -W clippy::large_futures +``` + +| 边界 | 基线 | 改后 | 结论 | +| --- | ---: | ---: | --- | +| 归档预览 copy helper | 65,792 B | 272 B | 内联 64 KiB 数组已移出状态机 | +| 归档解包 copy helper | 65,792 B | 272 B | 内联 64 KiB 数组已移出状态机 | +| 解包任务处理 Future | 67,448 B | 10,712 B | 改后值由临时 8 KiB 阈值复核 | +| 解包 task-spec 上层 Future | 67,736 B | 11,000 B | 不再触发默认阈值 | +| 预览任务处理 Future | 67,312 B | < 8,192 B | 改后在临时 8 KiB 阈值下也未报告 | +| 预览 task-spec 上层 Future | 67,600 B | < 8,192 B | 改后在临时 8 KiB 阈值下也未报告 | +| 个人上传 route | 23,616 B | 23,584 B | 同步化 helper 后减少 32 B，无装箱 | +| 团队上传 route | 23,624 B | 23,592 B | 同步化 helper 后减少 32 B，无装箱 | +| 上传 service 六个入口 | 16,432/23,264 B | 16,400/23,232 B | 每个减少 32 B，无装箱 | +| WebDAV dispatch | 27,280 B | 27,280 B | 未改变 dispatch 结构 | + +临时 8 KiB 配置只用于取得归档上层改后数值，没有修改仓库阈值。该检查仍显示其他 8–16 KiB +Future，因此不适合作为通用 CI 门槛。 + +## 分配次数与字节数 + +归档 copy 测量使用空的精确长度输入和 sink writer，只覆盖 Future 首次 poll 到完成。上传测量使用 +预先构造的 5 MiB `Bytes` 执行第一个 offset-staging chunk。WebDAV 测量先 warm-up 认证缓存，再执行 +一次完整的 authenticated `OPTIONS /webdav/` dispatch。 + +| 操作 | 基线分配 | 改后分配 | 差值 | +| --- | ---: | ---: | ---: | +| 归档预览 copy | 0 次 / 0 B | 1 次 / 65,536 B | +1 次 / +65,536 B | +| 归档解包 copy | 0 次 / 0 B | 1 次 / 65,536 B | +1 次 / +65,536 B | +| 5 MiB chunk upload | 346 次 / 2,184,661 B | 346 次 / 2,184,661 B | 0 次 / 0 B | +| WebDAV authenticated OPTIONS | 73 次 / 34,233 B | 73 次 / 34,233 B | 0 次 / 0 B | + +归档新增分配就是每次 copy 的一个固定 64 KiB `Box<[u8]>`，完整循环复用该缓冲区。上传和 WebDAV +没有新增每请求分配，因而没有引入 `Box::pin`、动态 Future dispatch 或其他热路径分配边界。 + +## Lint 分类 + +默认阈值下，主 crate 的 `large_futures` 从 17 条降至 9 条： + +- 归档 8 条级联警告全部消失； +- 余下 8 条属于 upload route/service，1 条属于 WebDAV dispatch； +- 四条 migration 警告保持不变，仍按顺序执行的 migration 路径单独处理。 + +五个生产 helper 的 `unused_async` 已清理。余下三条均为 +`src/services/auth/local/tokens/refresh.rs` 的 debug/test-support hook：一次 contention 通知以及 test hook +的安装、清理入口。相邻的 pause hook 确实等待通知且未触发该 lint；这些入口均不属于本次生产路径清理。 + +## 回归边界 + +两个 archive copy helper 的单元测试均覆盖： + +- 精确长度及内容保持； +- 输入提前结束； +- 输入超过声明长度； +- shutdown/cancellation； +- writer 失败传播； +- Future 小于 16 KiB； +- 每次 copy 恰好一次 65,536 B 缓冲区分配。 diff --git a/src/lib.rs b/src/lib.rs index 8ff95573a..36a66d097 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -25,3 +25,8 @@ pub mod storage; #[cfg(test)] pub(crate) mod test_support; pub mod webdav; + +#[cfg(test)] +#[global_allocator] +static TEST_ALLOCATOR: test_support::allocations::CountingAllocator = + test_support::allocations::CountingAllocator; diff --git a/src/services/content/version.rs b/src/services/content/version.rs index 998199710..e35184a0e 100644 --- a/src/services/content/version.rs +++ b/src/services/content/version.rs @@ -477,7 +477,7 @@ pub async fn cleanup_excess(state: &PrimaryAppState, file_id: i64) -> Result<()> let db = state.writer_db(); let file = file_repo::find_by_id(db, file_id).await?; let scope = resource_scope_from_file(&file)?; - let max_versions = get_max_versions(state).await; + let max_versions = get_max_versions(state); let mut deleted_count = 0u64; let mut reclaimed_bytes = 0i64; @@ -614,7 +614,7 @@ async fn cleanup_blobs_if_unused_by_counts( Ok(()) } -async fn get_max_versions(state: &PrimaryAppState) -> u64 { +fn get_max_versions(state: &PrimaryAppState) -> u64 { state .runtime_config .get_u64("max_versions_per_file") diff --git a/src/services/files/archive/preview/mod.rs b/src/services/files/archive/preview/mod.rs index 334d62b39..ba7ba0e66 100644 --- a/src/services/files/archive/preview/mod.rs +++ b/src/services/files/archive/preview/mod.rs @@ -294,7 +294,7 @@ where E: Fn(String) -> AsterError, { let mut copied = 0_u64; - let mut buffer = [0_u8; 64 * 1024]; + let mut buffer = vec![0_u8; 64 * 1024].into_boxed_slice(); loop { context.ensure_active()?; diff --git a/src/services/files/archive/preview/tests.rs b/src/services/files/archive/preview/tests.rs index 87fc93634..d88d64f78 100644 --- a/src/services/files/archive/preview/tests.rs +++ b/src/services/files/archive/preview/tests.rs @@ -20,6 +20,8 @@ use aster_forge_config::{ConfigSource, ConfigValueType}; use aster_forge_db::system_config; use aster_forge_tasks::TaskLease; +static COPY_ALLOCATION_MEASUREMENT_INPUT: [u8; 128 * 1024] = [0; 128 * 1024]; + struct PreviewMemoryRangeDriver { data: Vec, range_calls: AtomicUsize, @@ -709,6 +711,66 @@ async fn bounded_copy_stops_before_reading_when_shutdown_requested() { assert!(error.message().contains("shutdown")); } +#[tokio::test] +async fn bounded_copy_propagates_write_failure() { + let context = test_execution_context(); + let mut reader = &b"zip"[..]; + let mut output = crate::test_support::FailingAsyncWriter; + + let error = copy_async_reader_to_writer_with_execution_and_expected_size( + &context, + &mut reader, + &mut output, + 3, + "source archive", + |message| { + archive_preview_validation_error( + ApiErrorCode::ArchivePreviewSourceSizeMismatch, + message, + ) + }, + ) + .await + .expect_err("writer failure should abort the copy"); + + assert!( + error + .message() + .contains("write archive preview source stream chunk") + ); +} + +#[tokio::test] +async fn bounded_copy_future_size_and_allocations_are_bounded() { + let context = test_execution_context(); + let mut reader = COPY_ALLOCATION_MEASUREMENT_INPUT.as_slice(); + let mut output = tokio::io::sink(); + let future = copy_async_reader_to_writer_with_execution_and_expected_size( + &context, + &mut reader, + &mut output, + COPY_ALLOCATION_MEASUREMENT_INPUT.len() as u64, + "source archive", + |message| { + archive_preview_validation_error( + ApiErrorCode::ArchivePreviewSourceSizeMismatch, + message, + ) + }, + ); + let future_size = std::mem::size_of_val(&future); + let (result, allocations) = crate::test_support::allocations::measure_future(future).await; + + result.expect("multi-buffer exact-size stream should copy"); + eprintln!("archive preview copy future: {future_size} bytes; allocations: {allocations:?}"); + assert!( + future_size < 16 * 1024, + "archive preview copy future is {future_size} bytes" + ); + assert_eq!(allocations.count, 1); + assert_eq!(allocations.bytes, 64 * 1024); +} + fn test_execution_context() -> TaskExecutionContext { TaskExecutionContext::new( TaskLease::new(42, 7), diff --git a/src/services/files/file/lock.rs b/src/services/files/file/lock.rs index b3485ef7f..2e344f814 100644 --- a/src/services/files/file/lock.rs +++ b/src/services/files/file/lock.rs @@ -39,7 +39,7 @@ pub(crate) async fn set_lock_in_scope( let file = crate::services::workspace::storage::verify_file_access(state, scope, file_id).await?; - publish_file_lock_change(state, scope, &file, locked).await?; + publish_file_lock_change(state, scope, &file, locked)?; tracing::debug!( scope = ?scope, file_id = file.id, @@ -67,7 +67,7 @@ pub async fn set_lock( ))) } -async fn publish_file_lock_change( +fn publish_file_lock_change( state: &impl StorageChangeRuntimeState, scope: WorkspaceStorageScope, file: &file::Model, diff --git a/src/services/files/upload/lifecycle.rs b/src/services/files/upload/lifecycle.rs index c36f0ddb2..aecab6b4b 100644 --- a/src/services/files/upload/lifecycle.rs +++ b/src/services/files/upload/lifecycle.rs @@ -465,7 +465,7 @@ async fn cancel_upload_impl(state: &PrimaryAppState, session: upload_session::Mo pub async fn cancel_upload(state: &PrimaryAppState, upload_id: &str, user_id: i64) -> Result<()> { let session = load_upload_session(state, personal_scope(user_id), upload_id).await?; - let mode = upload_session_mode_label_for_cancel(state, &session).await; + let mode = upload_session_mode_label_for_cancel(&session); cancel_upload_impl(state, session) .await .inspect(|_| record_upload_cancel_metric(state, mode, true)) @@ -479,18 +479,15 @@ pub async fn cancel_upload_for_team( user_id: i64, ) -> Result<()> { let session = load_upload_session(state, team_scope(team_id, user_id), upload_id).await?; - let mode = upload_session_mode_label_for_cancel(state, &session).await; + let mode = upload_session_mode_label_for_cancel(&session); cancel_upload_impl(state, session) .await .inspect(|_| record_upload_cancel_metric(state, mode, true)) .inspect_err(|_| record_upload_cancel_metric(state, mode, false)) } -async fn upload_session_mode_label_for_cancel( - state: &PrimaryAppState, - session: &upload_session::Model, -) -> &'static str { - match upload_session_mode_label(state, session).await { +fn upload_session_mode_label_for_cancel(session: &upload_session::Model) -> &'static str { + match upload_session_mode_label(session) { Ok(mode) => mode, Err(error) => { // Metrics classification failure is recorded explicitly instead of blocking cleanup. @@ -503,10 +500,7 @@ async fn upload_session_mode_label_for_cancel( } } -async fn upload_session_mode_label( - _state: &PrimaryAppState, - session: &upload_session::Model, -) -> Result<&'static str> { +fn upload_session_mode_label(session: &upload_session::Model) -> Result<&'static str> { Ok(resolve_upload_session_kind(session)?.as_str()) } diff --git a/src/services/files/upload/provider_relay.rs b/src/services/files/upload/provider_relay.rs index 6b35e7dd8..b19d2568c 100644 --- a/src/services/files/upload/provider_relay.rs +++ b/src/services/files/upload/provider_relay.rs @@ -52,7 +52,7 @@ pub(super) async fn upload_bytes( chunk_number: i32, data: Bytes, ) -> Result { - let context = load_context(state, &session).await?; + let context = load_context(state, &session)?; upload_bytes_with_context(state, session, chunk_number, data, &context).await } @@ -112,7 +112,7 @@ pub(super) async fn upload_payload( chunk_number: i32, payload: actix_web::web::Payload, ) -> Result { - let context = load_context(state, &session).await?; + let context = load_context(state, &session)?; upload_payload_with_context(state, session, chunk_number, payload, &context).await } @@ -295,7 +295,7 @@ pub(super) async fn reconcile_progress( state: &PrimaryAppState, session: &upload_session::Model, ) -> Result> { - let context = load_context(state, session).await?; + let context = load_context(state, session)?; reconcile_progress_with_context(state, session, &context).await } @@ -339,7 +339,7 @@ fn validate_chunk_request(session: &upload_session::Model, chunk_number: i32) -> Ok(()) } -async fn load_context( +fn load_context( state: &PrimaryAppState, session: &upload_session::Model, ) -> Result { diff --git a/src/services/remote/master_binding.rs b/src/services/remote/master_binding.rs index 9c41fd4e8..aa1aec601 100644 --- a/src/services/remote/master_binding.rs +++ b/src/services/remote/master_binding.rs @@ -165,7 +165,7 @@ pub async fn authorize_presigned_put_request( )); } - let binding = authorize_presigned_binding_request(state, req).await?; + let binding = authorize_presigned_binding_request(state, req)?; resolve_authorized_ingress( state, binding, @@ -185,7 +185,7 @@ pub async fn authorize_presigned_get_request( )); } - let binding = authorize_presigned_binding_request(state, req).await?; + let binding = authorize_presigned_binding_request(state, req)?; resolve_authorized_ingress(state, binding, remote_storage_target_key(req)?, 0).await } @@ -285,7 +285,7 @@ async fn authorize_binding_request( Ok(binding) } -async fn authorize_presigned_binding_request( +fn authorize_presigned_binding_request( state: &S, req: &actix_web::HttpRequest, ) -> Result { diff --git a/src/services/task/archive/extract/staging.rs b/src/services/task/archive/extract/staging.rs index 6f48d3609..08b2e7a07 100644 --- a/src/services/task/archive/extract/staging.rs +++ b/src/services/task/archive/extract/staging.rs @@ -253,7 +253,7 @@ where W: tokio::io::AsyncWrite + Unpin, { let mut copied = 0_u64; - let mut buffer = [0_u8; 64 * 1024]; + let mut buffer = vec![0_u8; 64 * 1024].into_boxed_slice(); loop { context.ensure_active()?; @@ -539,6 +539,8 @@ mod tests { use super::copy_async_reader_to_writer_with_execution_and_expected_size; + static COPY_ALLOCATION_MEASUREMENT_INPUT: [u8; 128 * 1024] = [0; 128 * 1024]; + #[tokio::test] async fn source_archive_copy_writes_bytes_when_declared_size_matches() { let context = TaskExecutionContext::new( @@ -638,4 +640,60 @@ mod tests { assert!(is_task_worker_shutdown_requested(&error)); } + + #[tokio::test] + async fn source_archive_copy_propagates_write_failure() { + let context = TaskExecutionContext::new( + TaskLease::new(42, 7), + std::time::Duration::from_secs(60), + CancellationToken::new(), + ); + let mut reader = &b"archive"[..]; + let mut writer = crate::test_support::FailingAsyncWriter; + + let error = copy_async_reader_to_writer_with_execution_and_expected_size( + &context, + &mut reader, + &mut writer, + 7, + "source archive", + ) + .await + .expect_err("writer failure should abort the copy"); + + assert!( + error + .message() + .contains("write source archive stream chunk") + ); + } + + #[tokio::test] + async fn source_archive_copy_future_size_and_allocations_are_bounded() { + let context = TaskExecutionContext::new( + TaskLease::new(42, 7), + std::time::Duration::from_secs(60), + CancellationToken::new(), + ); + let mut reader = COPY_ALLOCATION_MEASUREMENT_INPUT.as_slice(); + let mut writer = tokio::io::sink(); + let future = copy_async_reader_to_writer_with_execution_and_expected_size( + &context, + &mut reader, + &mut writer, + COPY_ALLOCATION_MEASUREMENT_INPUT.len() as u64, + "source archive", + ); + let future_size = std::mem::size_of_val(&future); + let (result, allocations) = crate::test_support::allocations::measure_future(future).await; + + result.expect("multi-buffer exact-size stream should copy"); + eprintln!("archive extract copy future: {future_size} bytes; allocations: {allocations:?}"); + assert!( + future_size < 16 * 1024, + "archive extract copy future is {future_size} bytes" + ); + assert_eq!(allocations.count, 1); + assert_eq!(allocations.bytes, 64 * 1024); + } } diff --git a/src/test_support.rs b/src/test_support.rs index 65a92d088..4add9aacd 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -1,5 +1,126 @@ use std::{collections::BTreeSet, path::Path}; +pub(crate) mod allocations { + use std::alloc::{GlobalAlloc, Layout, System}; + use std::cell::Cell; + use std::future::Future; + + thread_local! { + static MEASURING: Cell = const { Cell::new(false) }; + static ALLOCATION_COUNT: Cell = const { Cell::new(0) }; + static ALLOCATED_BYTES: Cell = const { Cell::new(0) }; + } + + pub(crate) struct CountingAllocator; + + // SAFETY: every operation delegates to `System` with the original pointer and layout. The + // thread-local counters are observational and do not alter allocator behavior. + unsafe impl GlobalAlloc for CountingAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + // SAFETY: the caller upholds `GlobalAlloc::alloc`'s layout contract. + let pointer = unsafe { System.alloc(layout) }; + if !pointer.is_null() { + record_allocation(layout.size()); + } + pointer + } + + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + // SAFETY: the caller upholds `GlobalAlloc::alloc_zeroed`'s layout contract. + let pointer = unsafe { System.alloc_zeroed(layout) }; + if !pointer.is_null() { + record_allocation(layout.size()); + } + pointer + } + + unsafe fn dealloc(&self, pointer: *mut u8, layout: Layout) { + // SAFETY: the caller supplies the pointer and layout returned by this allocator. + unsafe { System.dealloc(pointer, layout) }; + } + + unsafe fn realloc(&self, pointer: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + // SAFETY: the caller supplies the original allocation and a valid new size. + let resized = unsafe { System.realloc(pointer, layout, new_size) }; + if !resized.is_null() { + record_allocation(new_size); + } + resized + } + } + + fn record_allocation(bytes: usize) { + MEASURING.with(|measuring| { + if measuring.get() { + ALLOCATION_COUNT.with(|count| count.set(count.get().saturating_add(1))); + ALLOCATED_BYTES.with(|total| total.set(total.get().saturating_add(bytes))); + } + }); + } + + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub(crate) struct AllocationMeasurement { + pub(crate) count: usize, + pub(crate) bytes: usize, + } + + struct MeasurementGuard; + + impl MeasurementGuard { + fn start() -> Self { + MEASURING.with(|measuring| { + assert!(!measuring.replace(true), "nested allocation measurement"); + }); + ALLOCATION_COUNT.with(|count| count.set(0)); + ALLOCATED_BYTES.with(|bytes| bytes.set(0)); + Self + } + } + + impl Drop for MeasurementGuard { + fn drop(&mut self) { + MEASURING.with(|measuring| measuring.set(false)); + } + } + + pub(crate) async fn measure_future(future: F) -> (F::Output, AllocationMeasurement) { + let guard = MeasurementGuard::start(); + let output = future.await; + let measurement = AllocationMeasurement { + count: ALLOCATION_COUNT.with(Cell::get), + bytes: ALLOCATED_BYTES.with(Cell::get), + }; + drop(guard); + (output, measurement) + } +} + +pub(crate) struct FailingAsyncWriter; + +impl tokio::io::AsyncWrite for FailingAsyncWriter { + fn poll_write( + self: std::pin::Pin<&mut Self>, + _context: &mut std::task::Context<'_>, + _buffer: &[u8], + ) -> std::task::Poll> { + std::task::Poll::Ready(Err(std::io::Error::other("injected write failure"))) + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _context: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _context: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } +} + pub(crate) fn snapshot_dir_tree(path: &Path) -> std::io::Result> { fn walk(root: &Path, current: &Path, entries: &mut BTreeSet) -> std::io::Result<()> { for entry in std::fs::read_dir(current)? {