soma_infra/
file_transfer.rs

1use std::path::{Path, PathBuf};
2
3use async_trait::async_trait;
4use serde::{Deserialize, Serialize};
5use soma_fleet::{FileTransfer, HostId, HostRecord, TopologyRevision, TransferReceipt};
6use soma_ops::{MutationSendState, OperationId, OperationName, Timestamp};
7use tokio_util::sync::CancellationToken;
8
9use crate::{FileReadPolicy, InfraError, InfraResult};
10
11/// Maximum bytes copied by one canonical file-transfer mutation.
12pub const MAX_FILE_TRANSFER_BYTES: u64 = 16 * 1024 * 1024;
13
14/// Explicit source and destination roots for one host.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct FileTransferPolicy {
17    source: FileReadPolicy,
18    destination: FileReadPolicy,
19}
20
21impl FileTransferPolicy {
22    /// Creates a policy with independent source and destination roots.
23    pub fn new<SI, SP, DI, DP>(source_roots: SI, destination_roots: DI) -> InfraResult<Self>
24    where
25        SI: IntoIterator<Item = SP>,
26        SP: Into<PathBuf>,
27        DI: IntoIterator<Item = DP>,
28        DP: Into<PathBuf>,
29    {
30        Ok(Self {
31            source: FileReadPolicy::new(source_roots)?,
32            destination: FileReadPolicy::new(destination_roots)?,
33        })
34    }
35
36    #[cfg(any(feature = "process-driver", test))]
37    pub(crate) fn resolve_source(&self, path: &Path) -> InfraResult<(PathBuf, PathBuf)> {
38        ensure_named_file(self.source.resolve(path)?)
39    }
40
41    #[cfg(any(feature = "process-driver", test))]
42    pub(crate) fn resolve_destination(&self, path: &Path) -> InfraResult<(PathBuf, PathBuf)> {
43        ensure_named_file(self.destination.resolve(path)?)
44    }
45}
46
47/// Stable file content identity.
48#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49pub struct TransferFileIdentity {
50    /// Absolute path.
51    pub path: PathBuf,
52    /// File size.
53    pub bytes: u64,
54    /// Lowercase SHA-256.
55    pub sha256: String,
56}
57
58/// Complete authorization-relevant transfer fingerprint.
59#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
60pub struct FileTransferFingerprint {
61    /// Source host.
62    pub source_host: HostId,
63    /// Source host revision.
64    pub source_revision: TopologyRevision,
65    /// Source file identity.
66    pub source: TransferFileIdentity,
67    /// Destination host.
68    pub destination_host: HostId,
69    /// Destination host revision.
70    pub destination_revision: TopologyRevision,
71    /// Destination absolute path.
72    pub destination_path: PathBuf,
73    /// Existing destination identity, when present.
74    pub destination_before: Option<TransferFileIdentity>,
75}
76
77/// Deadline-bound transfer request.
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct VerifiedFileTransferRequest {
80    /// Operation identity.
81    pub operation_id: OperationId,
82    /// Canonical operation.
83    pub operation: OperationName,
84    /// Planned transfer fingerprint.
85    pub fingerprint: FileTransferFingerprint,
86    /// Absolute execution deadline.
87    pub deadline: Timestamp,
88}
89
90/// Verified file-transfer result.
91#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
92pub struct VerifiedFileTransferOutcome {
93    /// Planned fingerprint.
94    pub before: FileTransferFingerprint,
95    /// Destination identity after transfer.
96    pub destination_after: TransferFileIdentity,
97    /// Bytes copied.
98    pub bytes: u64,
99    /// Backend send state.
100    pub send_state: MutationSendState,
101    /// Whether source and destination digests match.
102    pub verified: bool,
103    /// Whether destination content changed.
104    pub changed: bool,
105}
106
107/// Policy role used while inspecting a transfer path.
108#[derive(Debug, Clone, Copy, PartialEq, Eq)]
109pub enum FileTransferPathRole {
110    /// Source read policy.
111    Source,
112    /// Destination write policy.
113    Destination,
114}
115
116/// Reads file identities for transfer planning and verification.
117#[async_trait]
118pub trait FileTransferInspector: Send + Sync {
119    /// Reads one file identity, optionally returning absence.
120    async fn inspect_transfer_file(
121        &self,
122        host: &HostRecord,
123        path: &Path,
124        role: FileTransferPathRole,
125        optional: bool,
126        cancellation: &CancellationToken,
127    ) -> InfraResult<Option<TransferFileIdentity>>;
128}
129
130/// Complete transfer client used by the verified engine.
131pub trait VerifiedFileTransferClient: FileTransfer + FileTransferInspector {}
132impl<T> VerifiedFileTransferClient for T where T: FileTransfer + FileTransferInspector {}
133
134#[cfg(any(feature = "process-driver", test))]
135fn ensure_named_file((root, relative): (PathBuf, PathBuf)) -> InfraResult<(PathBuf, PathBuf)> {
136    if relative.as_os_str().is_empty() {
137        Err(InfraError::InvalidRequest {
138            domain: "file-transfer",
139            message: "transfer path must name a file beneath its configured root".into(),
140        })
141    } else {
142        Ok((root, relative))
143    }
144}
145
146#[cfg(any(feature = "process-driver", test))]
147pub(crate) fn identity_from_bytes(path: &Path, bytes: &[u8]) -> TransferFileIdentity {
148    TransferFileIdentity {
149        path: path.to_path_buf(),
150        bytes: bytes.len() as u64,
151        sha256: crate::mutation::sha256_hex(bytes),
152    }
153}
154
155pub(crate) fn receipt_identity(receipt: &TransferReceipt) -> InfraResult<(&str, &str)> {
156    let source = receipt
157        .source_sha256()
158        .ok_or_else(|| InfraError::InvalidRequest {
159            domain: "file-transfer",
160            message: "transfer receipt is missing source digest".into(),
161        })?;
162    let destination = receipt
163        .destination_sha256()
164        .ok_or_else(|| InfraError::InvalidRequest {
165            domain: "file-transfer",
166            message: "transfer receipt is missing destination digest".into(),
167        })?;
168    Ok((source, destination))
169}
170
171#[cfg(test)]
172#[path = "file_transfer_tests.rs"]
173mod tests;
soma_infra/file_transfer.rs

soma_infra/
file_transfer.rs
