weavatrix_worktree/operation/
mod.rs

1mod commit;
2mod contract;
3mod journal;
4mod model;
5mod pending;
6mod projection;
7mod recovery;
8mod recovery_model;
9mod stage;
10#[cfg(test)]
11mod tests;
12mod undo;
13
14use crate::{
15    WorktreePlan,
16    error::{TransactionPhase, WorktreeError, WorktreeErrorCode},
17    filesystem::{ControlDir, FsRoot},
18    journal::FinishOutcome,
19    options::WorktreeOptions,
20    report::{OperationChange, WorktreeDryRunReport},
21    transaction::acquire,
22};
23
24use journal::{Record, Writer};
25use model::StagedPath;
26use pending::ensure_no_pending;
27pub use undo::{
28    ParseUndoIdError, RetainedApplyReport, UndoId, UndoReceipt, UndoRetention, UndoRollbackReport,
29    UndoStoreUsage, WorktreeSnapshotFingerprint,
30};
31pub(crate) use undo::{undo_discard, undo_receipts, undo_rollback, undo_usage};
32
33/// A durably staged create/delete/modify/rename transaction holding the root lock.
34#[must_use = "commit, abort, or later recover the prepared transaction"]
35pub struct PreparedWorktreeTransaction {
36    transaction_id: String,
37    contract_hash: crate::Sha256Hash,
38    operation: String,
39    preview: WorktreeDryRunReport,
40    operations: Vec<OperationChange>,
41    paths: Vec<StagedPath>,
42    options: WorktreeOptions,
43    root: FsRoot,
44    journal: Writer,
45    control: ControlDir,
46    _lock: std::fs::File,
47}
48
49impl core::fmt::Debug for PreparedWorktreeTransaction {
50    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
51        formatter
52            .debug_struct("PreparedWorktreeTransaction")
53            .field("transaction_id", &self.transaction_id)
54            .field("operation", &self.operation)
55            .field("prepared_paths", &self.paths.len())
56            .finish_non_exhaustive()
57    }
58}
59
60impl PreparedWorktreeTransaction {
61    #[must_use]
62    pub fn transaction_id(&self) -> &str {
63        &self.transaction_id
64    }
65
66    #[must_use]
67    pub const fn preview(&self) -> &WorktreeDryRunReport {
68        &self.preview
69    }
70}
71
72pub(crate) fn dry_run_operation_plan(
73    root: &FsRoot,
74    options: WorktreeOptions,
75    plan: &WorktreePlan,
76) -> Result<WorktreeDryRunReport, WorktreeError> {
77    let validated = contract::validate(plan, options)?;
78    projection::project(root, options, &validated).map(|value| projection::preview(&value))
79}
80
81pub(crate) fn prepare_operation_plan(
82    root: &FsRoot,
83    options: WorktreeOptions,
84    plan: &WorktreePlan,
85) -> Result<PreparedWorktreeTransaction, WorktreeError> {
86    let validated = contract::validate(plan, options)?;
87    let locked = acquire(root)?;
88    ensure_no_pending(&locked.control)?;
89    let projected = projection::project(root, options, &validated)?;
90    let preview = projection::preview(&projected);
91    let path_count = projected.paths.len();
92    let transaction_id = random_id()?;
93    let file = locked.control.create_operation_journal().map_err(|error| {
94        WorktreeError::with_source(
95            WorktreeErrorCode::Io,
96            TransactionPhase::Prepare,
97            "failed to create the exclusive operation journal",
98            error,
99        )
100    })?;
101    locked.control.sync().map_err(|error| {
102        WorktreeError::with_source(
103            WorktreeErrorCode::DurabilityFailed,
104            TransactionPhase::Prepare,
105            "failed to synchronize the operation journal directory",
106            error,
107        )
108        .requiring_recovery()
109    })?;
110    let mut journal =
111        Writer::new(file, options.limits.max_journal_bytes as u64).map_err(|error| {
112            operation_journal_error(
113                TransactionPhase::Prepare,
114                "invalid new operation journal",
115                error,
116            )
117            .requiring_recovery()
118        })?;
119    append_header(
120        &mut journal,
121        plan,
122        validated.fingerprint(),
123        &transaction_id,
124        path_count,
125    )?;
126    let (operation, operations, paths) =
127        match stage::stage_all(projected, &transaction_id, options, &mut journal) {
128            Ok(value) => value,
129            Err(error) if error.recovery_required() => {
130                return Err(error.in_transaction(transaction_id));
131            }
132            Err(error) => {
133                finish_failed_prepare(&mut journal, &locked.control)?;
134                return Err(error.in_transaction(transaction_id));
135            }
136        };
137    journal
138        .append(&Record::Prepared {
139            operation_count: u32::try_from(operations.len())
140                .map_err(|_| too_large("operation count does not fit the journal contract"))?,
141            path_count: u32::try_from(paths.len())
142                .map_err(|_| too_large("path count does not fit the journal contract"))?,
143        })
144        .map_err(|error| {
145            operation_journal_error(
146                TransactionPhase::Prepare,
147                "failed to record durable operation preparation",
148                error,
149            )
150            .requiring_recovery()
151        })?;
152    Ok(PreparedWorktreeTransaction {
153        transaction_id,
154        contract_hash: crate::Sha256Hash::parse(&validated.fingerprint().to_string())
155            .expect("validated plan fingerprint is SHA-256"),
156        operation,
157        preview,
158        operations,
159        paths,
160        options,
161        root: root.try_clone().map_err(|error| {
162            WorktreeError::with_source(
163                WorktreeErrorCode::Io,
164                TransactionPhase::Prepare,
165                "failed to retain the worktree root capability",
166                error,
167            )
168            .requiring_recovery()
169        })?,
170        journal,
171        control: locked.control,
172        _lock: locked.file,
173    })
174}
175
176pub(crate) use recovery::recover_operation_transaction;
177
178fn append_header(
179    journal: &mut Writer,
180    plan: &WorktreePlan,
181    contract_hash: weavatrix_refactor_plan::PlanFingerprint,
182    transaction_id: &str,
183    path_count: usize,
184) -> Result<(), WorktreeError> {
185    journal
186        .append(&Record::Header {
187            transaction_id: transaction_id.to_owned(),
188            contract_hash: contract_hash.to_string(),
189            operation: plan.operation.clone(),
190            operation_count: u32::try_from(plan.operations.len())
191                .map_err(|_| too_large("operation count does not fit the journal contract"))?,
192            path_count: u32::try_from(path_count)
193                .map_err(|_| too_large("path count does not fit the journal contract"))?,
194        })
195        .map_err(|error| {
196            operation_journal_error(
197                TransactionPhase::Prepare,
198                "failed to synchronize the operation journal header",
199                error,
200            )
201            .requiring_recovery()
202        })?;
203    Ok(())
204}
205
206fn finish_failed_prepare(journal: &mut Writer, control: &ControlDir) -> Result<(), WorktreeError> {
207    journal
208        .append(&Record::Finished {
209            outcome: FinishOutcome::Aborted,
210        })
211        .map_err(|error| {
212            operation_journal_error(
213                TransactionPhase::Cleanup,
214                "failed to record aborted operation preparation",
215                error,
216            )
217            .requiring_recovery()
218        })?;
219    control.remove_operation_journal().map_err(|error| {
220        WorktreeError::with_source(
221            WorktreeErrorCode::RecoveryRequired,
222            TransactionPhase::Cleanup,
223            "failed to remove the aborted operation journal",
224            error,
225        )
226        .requiring_recovery()
227    })
228}
229
230fn random_id() -> Result<String, WorktreeError> {
231    const HEX: &[u8; 16] = b"0123456789abcdef";
232    let mut bytes = [0_u8; 16];
233    getrandom::fill(&mut bytes).map_err(|error| {
234        WorktreeError::with_source(
235            WorktreeErrorCode::Io,
236            TransactionPhase::Prepare,
237            "failed to generate an operation transaction identifier",
238            error,
239        )
240    })?;
241    Ok(bytes
242        .iter()
243        .fold(String::with_capacity(32), |mut value, byte| {
244            value.push(char::from(HEX[usize::from(byte >> 4)]));
245            value.push(char::from(HEX[usize::from(byte & 15)]));
246            value
247        }))
248}
249
250fn operation_journal_error(
251    phase: TransactionPhase,
252    message: &str,
253    source: impl std::error::Error + Send + Sync + 'static,
254) -> WorktreeError {
255    WorktreeError::with_source(WorktreeErrorCode::JournalCorrupt, phase, message, source)
256}
257
258fn too_large(message: &str) -> WorktreeError {
259    WorktreeError::new(
260        WorktreeErrorCode::TransactionTooLarge,
261        TransactionPhase::Prepare,
262        message,
263    )
264}
weavatrix_worktree/operation/mod.rs

weavatrix_worktree/operation/
mod.rs
