soma_infra/
file_transfer_engine.rs

1use std::path::Path;
2
3use soma_fleet::{HostRecord, TransferRequest};
4use soma_ops::MutationSendState;
5use tokio_util::sync::CancellationToken;
6
7use crate::file_transfer::receipt_identity;
8use crate::{
9    FileTransferFingerprint, FileTransferPathRole, InfraError, InfraResult,
10    MAX_FILE_TRANSFER_BYTES, MutationFailure, MutationResult, VerifiedFileTransferClient,
11    VerifiedFileTransferOutcome, VerifiedFileTransferRequest,
12};
13
14/// Verified bounded file-transfer coordinator.
15#[derive(Debug, Clone, Copy, Default)]
16pub struct FileTransferEngine;
17
18impl FileTransferEngine {
19    /// Captures source and destination pre-state.
20    pub async fn inspect(
21        &self,
22        client: &dyn VerifiedFileTransferClient,
23        source: &HostRecord,
24        source_path: &Path,
25        destination: &HostRecord,
26        destination_path: &Path,
27        cancellation: &CancellationToken,
28    ) -> InfraResult<FileTransferFingerprint> {
29        let source_identity = client
30            .inspect_transfer_file(
31                source,
32                source_path,
33                FileTransferPathRole::Source,
34                false,
35                cancellation,
36            )
37            .await?
38            .ok_or_else(|| InfraError::InvalidRequest {
39                domain: "file-transfer",
40                message: "source file is absent".into(),
41            })?;
42        if source_identity.bytes > MAX_FILE_TRANSFER_BYTES {
43            return Err(InfraError::InvalidRequest {
44                domain: "file-transfer",
45                message: format!(
46                    "source exceeds the {MAX_FILE_TRANSFER_BYTES}-byte transfer limit"
47                ),
48            });
49        }
50        let destination_before = client
51            .inspect_transfer_file(
52                destination,
53                destination_path,
54                FileTransferPathRole::Destination,
55                true,
56                cancellation,
57            )
58            .await?;
59        Ok(FileTransferFingerprint {
60            source_host: source.id().clone(),
61            source_revision: source.revision().clone(),
62            source: source_identity,
63            destination_host: destination.id().clone(),
64            destination_revision: destination.revision().clone(),
65            destination_path: destination_path.to_path_buf(),
66            destination_before,
67        })
68    }
69
70    /// Executes a transfer and independently verifies destination content.
71    pub async fn execute(
72        &self,
73        client: &dyn VerifiedFileTransferClient,
74        source: &HostRecord,
75        destination: &HostRecord,
76        request: &VerifiedFileTransferRequest,
77        cancellation: &CancellationToken,
78    ) -> MutationResult<VerifiedFileTransferOutcome> {
79        if cancellation.is_cancelled() {
80            return Err(not_sent(soma_fleet::FleetError::Cancelled.into()));
81        }
82        if request.deadline <= soma_ops::Timestamp::now() {
83            return Err(not_sent(soma_fleet::FleetError::DeadlineExceeded.into()));
84        }
85        let current = self
86            .inspect(
87                client,
88                source,
89                &request.fingerprint.source.path,
90                destination,
91                &request.fingerprint.destination_path,
92                cancellation,
93            )
94            .await
95            .map_err(not_sent)?;
96        if current != request.fingerprint {
97            return Err(not_sent(InfraError::InvalidRequest {
98                domain: "file-transfer",
99                message: "source or destination state changed after planning".into(),
100            }));
101        }
102        let transfer = TransferRequest::new(
103            source.id().clone(),
104            request.fingerprint.source.path.clone(),
105            destination.id().clone(),
106            request.fingerprint.destination_path.clone(),
107            MAX_FILE_TRANSFER_BYTES,
108            request.deadline,
109        )
110        .map_err(soma_fleet::FleetError::from)
111        .map_err(|error| not_sent(error.into()))?
112        .with_expected_source_sha256(request.fingerprint.source.sha256.clone());
113        let receipt = client
114            .transfer(source, destination, &transfer, cancellation)
115            .await
116            .map_err(|error| MutationFailure::new(MutationSendState::Unknown, error.into()))?;
117        let destination_after = client
118            .inspect_transfer_file(
119                destination,
120                &request.fingerprint.destination_path,
121                FileTransferPathRole::Destination,
122                false,
123                cancellation,
124            )
125            .await
126            .map_err(|error| MutationFailure::new(MutationSendState::Sent, error))?
127            .ok_or_else(|| {
128                MutationFailure::new(
129                    MutationSendState::Sent,
130                    InfraError::InvalidRequest {
131                        domain: "file-transfer",
132                        message: "destination is absent after transfer".into(),
133                    },
134                )
135            })?;
136        let (source_digest, destination_digest) = receipt_identity(&receipt)
137            .map_err(|error| MutationFailure::new(MutationSendState::Sent, error))?;
138        let verified = receipt.verified()
139            && receipt.bytes() == request.fingerprint.source.bytes
140            && destination_after.bytes == request.fingerprint.source.bytes
141            && source_digest == request.fingerprint.source.sha256
142            && destination_digest == destination_after.sha256
143            && source_digest == destination_digest;
144        if !verified {
145            return Err(MutationFailure::new(
146                MutationSendState::Sent,
147                InfraError::InvalidRequest {
148                    domain: "file-transfer",
149                    message: "source and destination transfer evidence does not match".into(),
150                },
151            ));
152        }
153        let changed = request
154            .fingerprint
155            .destination_before
156            .as_ref()
157            .is_none_or(|before| before.sha256 != destination_after.sha256);
158        Ok(VerifiedFileTransferOutcome {
159            before: request.fingerprint.clone(),
160            destination_after,
161            bytes: receipt.bytes(),
162            send_state: MutationSendState::Sent,
163            verified,
164            changed,
165        })
166    }
167}
168
169fn not_sent(error: InfraError) -> MutationFailure {
170    MutationFailure::new(MutationSendState::NotSent, error)
171}
172
173#[cfg(test)]
174#[path = "file_transfer_engine_tests.rs"]
175mod tests;
soma_infra/file_transfer_engine.rs

soma_infra/
file_transfer_engine.rs
