qubit_fs/write/
async_write_all_operation.rs

1// =============================================================================
2//    Copyright (c) 2026 Haixing Hu.
3//
4//    SPDX-License-Identifier: Apache-2.0
5//
6//    Licensed under the Apache License, Version 2.0.
7// =============================================================================
8//! Owning asynchronous whole-file write operation.
9use std::fmt::Debug;
10use std::fmt::Formatter;
11use std::fmt::Result as FmtResult;
12use std::io::Error as IoError;
13
14use qubit_io::AsyncOutput;
15
16use crate::AsyncFileSystem;
17use crate::error::FsError;
18use crate::error::FsErrorKind;
19use crate::error::FsOperation;
20use crate::error::OpenFailureStage;
21use crate::metadata::WriteOutcome;
22use crate::path::Path;
23use crate::write::AsyncWriteAllOperationFailure;
24use crate::write::AsyncWriteAllOperationState;
25use crate::write::AsyncWriterRecovery;
26use crate::write::WriteFailureState;
27use crate::write::WriteOptions;
28use crate::write::WriterState;
29use crate::write::internal::WriteAllCancellationGuard;
30use crate::write::internal::WriteAllRecoverySnapshot;
31use crate::write::internal::open_failure_state;
32/// Owning asynchronous whole-file write that survives cancellation.
33///
34/// # Examples
35///
36/// The example runs against an isolated in-memory fixture. Applications obtain
37/// their configured facade from a provider or registry integration.
38///
39/// ```rust
40/// # mod support { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/common/rustdoc_support.rs")); }
41/// # use support::*;
42/// # let (filesystem, _) = async_recording_spi::async_recording_file_system(Default::default());
43/// # poll_support::ready(async {
44/// use qubit_fs::Path;
45/// use qubit_fs::write::WriteOptions;
46/// use qubit_fs::write::AsyncWriteAllOperationState;
47///
48/// let mut operation = filesystem.begin_write_all(
49///     Path::parse("/report")?, b"bytes".to_vec(), WriteOptions::default(),
50/// )?;
51/// operation.execute().await?;
52/// assert_eq!(AsyncWriteAllOperationState::Completed, operation.state());
53/// assert_eq!(5, operation.written_bytes());
54/// assert!(!operation.has_recovery());
55/// # Ok::<(), Box<dyn std::error::Error>>(())
56/// # }).unwrap();
57/// ```
58#[must_use]
59pub struct AsyncWriteAllOperation {
60    /// Filesystem retained independently of the caller.
61    filesystem: AsyncFileSystem,
62    /// Target retained for recovery inspection.
63    path: Path,
64    /// Request payload, held only until execution ends or is cancelled.
65    bytes: Vec<u8>,
66    /// Validated request options.
67    options: WriteOptions,
68    /// Historical execution state.
69    state: AsyncWriteAllOperationState,
70    /// Opened writer retained while recovery may be required.
71    writer: Option<AsyncWriterRecovery>,
72    /// Frozen publication and confirmed-progress snapshot.
73    recovery: WriteAllRecoverySnapshot,
74}
75impl AsyncWriteAllOperation {
76    /// Creates an operation after facade preflight has succeeded.
77    pub(crate) fn new(filesystem: AsyncFileSystem, path: Path, bytes: Vec<u8>, options: WriteOptions) -> Self {
78        Self {
79            filesystem,
80            path,
81            bytes,
82            options,
83            state: AsyncWriteAllOperationState::Ready,
84            writer: None,
85            recovery: WriteAllRecoverySnapshot::new(),
86        }
87    }
88    /// Returns the filesystem retained for execution and recovery inspection.
89    #[inline]
90    #[must_use]
91    pub const fn filesystem(&self) -> &AsyncFileSystem {
92        &self.filesystem
93    }
94    /// Returns the requested target path.
95    #[inline]
96    #[must_use]
97    pub const fn path(&self) -> &Path {
98        &self.path
99    }
100    /// Returns the operation state.
101    #[inline]
102    #[must_use = "inspect publication state before choosing a recovery action"]
103    pub const fn state(&self) -> AsyncWriteAllOperationState {
104        self.state
105    }
106    /// Reports whether an opened writer is retained for recovery.
107    #[inline]
108    #[must_use]
109    pub const fn has_recovery(&self) -> bool {
110        self.writer.is_some()
111    }
112    /// Returns mutable access to the retained recovery writer.
113    #[inline]
114    #[must_use]
115    pub fn recovery(&mut self) -> Option<&mut AsyncWriterRecovery> {
116        self.writer.as_mut()
117    }
118    /// Transfers recovery responsibility to the caller.
119    ///
120    /// Subsequent writer recovery does not change this operation's historical
121    /// publication state or byte count.
122    #[inline]
123    #[must_use]
124    pub fn take_recovery(&mut self) -> Option<AsyncWriterRecovery> {
125        self.writer.take()
126    }
127    /// Returns the frozen count of bytes confirmed before execution stopped.
128    #[inline]
129    #[must_use]
130    pub const fn written_bytes(&self) -> u64 {
131        self.recovery.written_bytes
132    }
133    /// Executes the operation once, retaining recovery state on failure.
134    ///
135    /// # Cancellation
136    ///
137    /// Keep this operation outside the cancellation scope and cancel only this
138    /// future. Dropping an unpolled future leaves the operation ready. After
139    /// execution starts, cancellation records indeterminate publication and
140    /// retains any opened writer. A missing writer does not prove no effects.
141    /// The payload is released when this future finishes or is dropped.
142    ///
143    /// # Returns
144    ///
145    /// The confirmed write outcome. Success freezes the acknowledged byte count
146    /// and releases the committed writer.
147    ///
148    /// # Errors
149    ///
150    /// Returns the provider failure with confirmed progress, or `InvalidState`
151    /// if execution has already started. Repeated execution preserves the
152    /// historical state and does not invoke the provider again.
153    pub async fn execute(&mut self) -> Result<WriteOutcome, AsyncWriteAllOperationFailure> {
154        if self.state != AsyncWriteAllOperationState::Ready {
155            return Err(AsyncWriteAllOperationFailure::new(
156                invalid_state(&self.path, &self.filesystem),
157                self.recovery.state,
158                self.recovery.written_bytes,
159            ));
160        }
161        let Self {
162            filesystem,
163            path,
164            bytes,
165            options,
166            state,
167            writer,
168            recovery,
169        } = self;
170        let bytes = std::mem::take(bytes);
171        let mut guard = WriteAllCancellationGuard::start(state, writer, recovery);
172        let result = execute_write(filesystem, path, &bytes, options, guard.writer_mut()).await;
173        guard.finish(&result);
174        result
175    }
176}
177/// Runs provider stages while keeping the session in the recovery slot.
178async fn execute_write(
179    filesystem: &AsyncFileSystem,
180    path: &Path,
181    bytes: &[u8],
182    options: &WriteOptions,
183    slot: &mut Option<AsyncWriterRecovery>,
184) -> Result<WriteOutcome, AsyncWriteAllOperationFailure> {
185    if slot.is_none() {
186        match filesystem.open_writer(path, options.clone()).await {
187            Ok(writer) => *slot = Some(AsyncWriterRecovery::Opened(Box::new(writer))),
188            Err(failure) => {
189                let (error, stage, recovery) = failure.into_parts();
190                *slot = recovery.map(AsyncWriterRecovery::Rejected);
191                let state = match stage {
192                    OpenFailureStage::Preflight => WriteFailureState::NotPublished,
193                    OpenFailureStage::ProviderOpen => open_failure_state(&error),
194                    OpenFailureStage::OutcomeValidation => WriteFailureState::Indeterminate,
195                };
196                return Err(AsyncWriteAllOperationFailure::new(error, state, 0));
197            }
198        }
199    }
200    let writer = slot
201        .as_mut()
202        .and_then(AsyncWriterRecovery::opened_mut)
203        .expect("writer is retained after open");
204    if let Err(error) = writer.write_fully_async(bytes).await {
205        let error = contextual(filesystem, error, path);
206        let state = state_for(error.has_indeterminate_effect(), writer.state());
207        return Err(AsyncWriteAllOperationFailure::new(error, state, writer.written_bytes()));
208    }
209    if let Err(error) = writer.flush_async().await {
210        let error = contextual(filesystem, error, path);
211        let state = state_for(error.has_indeterminate_effect(), writer.state());
212        return Err(AsyncWriteAllOperationFailure::new(error, state, writer.written_bytes()));
213    }
214    match writer.commit_async().await {
215        Ok(outcome) => Ok(outcome),
216        Err(failure) => {
217            let (error, state) = failure.into_parts();
218            Err(AsyncWriteAllOperationFailure::new(error, state, writer.written_bytes()))
219        }
220    }
221}
222/// Restores filesystem and target context to a stream failure.
223fn contextual(filesystem: &AsyncFileSystem, error: IoError, path: &Path) -> FsError {
224    filesystem.core().enrich(
225        FsError::from_stream_io(error, FsOperation::Write, path),
226        Some(path),
227        FsOperation::Write,
228    )
229}
230/// Classifies a post-open stream failure using writer publication facts.
231fn state_for(indeterminate: bool, state: WriterState) -> WriteFailureState {
232    if indeterminate {
233        WriteFailureState::Indeterminate
234    } else {
235        state.publication_failure_state()
236    }
237}
238/// Builds a repeated-execution error without changing historical facts.
239fn invalid_state(path: &Path, filesystem: &AsyncFileSystem) -> FsError {
240    FsError::new(
241        FsErrorKind::InvalidState,
242        FsOperation::Write,
243        "async whole-file write cannot execute in its current state",
244    )
245    .with_path(path.clone())
246    .with_provider(filesystem.properties().info().provider_id())
247}
248
249impl Debug for AsyncWriteAllOperation {
250    /// Formats lifecycle facts without payload or provider session contents.
251    fn fmt(&self, formatter: &mut Formatter<'_>) -> FmtResult {
252        formatter
253            .debug_struct("AsyncWriteAllOperation")
254            .field("path", &self.path)
255            .field("state", &self.state)
256            .field("written_bytes", &self.recovery.written_bytes)
257            .field("has_recovery", &self.writer.is_some())
258            .finish()
259    }
260}
qubit_fs/write/async_write_all_operation.rs

qubit_fs/write/
async_write_all_operation.rs
