soma_fleet/
transfer.rs

1use std::path::{Path, PathBuf};
2
3use serde::{Deserialize, Serialize};
4use soma_ops::Timestamp;
5
6use crate::{HostId, RequestError, request::validate_absolute_path};
7
8const MAX_TRANSFER_BYTES: u64 = 1024 * 1024 * 1024 * 1024;
9
10/// Descriptor-confined transfer request between managed hosts.
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
12pub struct TransferRequest {
13    source_host: HostId,
14    source_path: PathBuf,
15    destination_host: HostId,
16    destination_path: PathBuf,
17    max_bytes: u64,
18    deadline: Timestamp,
19    #[serde(default, skip_serializing_if = "Option::is_none")]
20    expected_source_sha256: Option<String>,
21}
22
23impl TransferRequest {
24    /// Creates a validated bounded transfer request.
25    pub fn new(
26        source_host: HostId,
27        source_path: impl Into<PathBuf>,
28        destination_host: HostId,
29        destination_path: impl Into<PathBuf>,
30        max_bytes: u64,
31        deadline: Timestamp,
32    ) -> Result<Self, RequestError> {
33        if max_bytes == 0 || max_bytes > MAX_TRANSFER_BYTES {
34            return Err(RequestError::InvalidTransferLimit {
35                bytes: max_bytes,
36                max: MAX_TRANSFER_BYTES,
37            });
38        }
39        Ok(Self {
40            source_host,
41            source_path: validate_absolute_path(source_path.into())?,
42            destination_host,
43            destination_path: validate_absolute_path(destination_path.into())?,
44            max_bytes,
45            deadline,
46            expected_source_sha256: None,
47        })
48    }
49
50    /// Binds the pre-checked source SHA-256 that the bytes read for send must match.
51    ///
52    /// When set, a transfer implementation must hash the bytes it actually reads
53    /// for send and reject the transfer before any destination write if the
54    /// digest has drifted from this pre-checked value.
55    #[must_use]
56    pub fn with_expected_source_sha256(mut self, sha256: impl Into<String>) -> Self {
57        self.expected_source_sha256 = Some(sha256.into());
58        self
59    }
60
61    /// Returns the pre-checked source SHA-256 bound at plan time, when present.
62    #[must_use]
63    pub fn expected_source_sha256(&self) -> Option<&str> {
64        self.expected_source_sha256.as_deref()
65    }
66
67    /// Rejects a request whose deadline has elapsed.
68    pub fn validate_at(&self, now: Timestamp) -> Result<(), RequestError> {
69        if self.deadline <= now {
70            Err(RequestError::DeadlineElapsed)
71        } else {
72            Ok(())
73        }
74    }
75
76    /// Returns the source host.
77    #[must_use]
78    pub fn source_host(&self) -> &HostId {
79        &self.source_host
80    }
81
82    /// Returns the source absolute path.
83    #[must_use]
84    pub fn source_path(&self) -> &Path {
85        &self.source_path
86    }
87
88    /// Returns the destination host.
89    #[must_use]
90    pub fn destination_host(&self) -> &HostId {
91        &self.destination_host
92    }
93
94    /// Returns the destination absolute path.
95    #[must_use]
96    pub fn destination_path(&self) -> &Path {
97        &self.destination_path
98    }
99
100    /// Returns the maximum accepted byte count.
101    #[must_use]
102    pub const fn max_bytes(&self) -> u64 {
103        self.max_bytes
104    }
105
106    /// Returns the transfer deadline.
107    #[must_use]
108    pub const fn deadline(&self) -> Timestamp {
109        self.deadline
110    }
111}
112
113/// Verified transfer completion metadata.
114#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
115pub struct TransferReceipt {
116    bytes: u64,
117    source_sha256: Option<String>,
118    destination_sha256: Option<String>,
119}
120
121impl TransferReceipt {
122    /// Creates a receipt without content digests.
123    #[must_use]
124    pub const fn new(bytes: u64) -> Self {
125        Self {
126            bytes,
127            source_sha256: None,
128            destination_sha256: None,
129        }
130    }
131
132    /// Adds verified source and destination SHA-256 digests.
133    pub fn with_digests(
134        mut self,
135        source: impl Into<String>,
136        destination: impl Into<String>,
137    ) -> Result<Self, RequestError> {
138        let source = source.into();
139        let destination = destination.into();
140        validate_sha256(&source)?;
141        validate_sha256(&destination)?;
142        self.source_sha256 = Some(source);
143        self.destination_sha256 = Some(destination);
144        Ok(self)
145    }
146
147    /// Returns transferred bytes.
148    #[must_use]
149    pub const fn bytes(&self) -> u64 {
150        self.bytes
151    }
152
153    /// Returns the source SHA-256 when recorded.
154    #[must_use]
155    pub fn source_sha256(&self) -> Option<&str> {
156        self.source_sha256.as_deref()
157    }
158
159    /// Returns the destination SHA-256 when recorded.
160    #[must_use]
161    pub fn destination_sha256(&self) -> Option<&str> {
162        self.destination_sha256.as_deref()
163    }
164
165    /// Returns whether source and destination digests match.
166    #[must_use]
167    pub fn verified(&self) -> bool {
168        matches!(
169            (&self.source_sha256, &self.destination_sha256),
170            (Some(source), Some(destination)) if source == destination
171        )
172    }
173}
174
175fn validate_sha256(value: &str) -> Result<(), RequestError> {
176    if value.len() == 64
177        && value
178            .bytes()
179            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
180    {
181        Ok(())
182    } else {
183        Err(RequestError::InvalidSha256)
184    }
185}
186#[cfg(test)]
187#[path = "transfer_tests.rs"]
188mod tests;
soma_fleet/transfer.rs

soma_fleet/
transfer.rs
