harmos_storage/
file_reads.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Finished File Reads
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::io;
20use std::num::NonZeroUsize;
21use std::sync::Arc;
22
23// Third Party Imports
24use sha2::{Digest, Sha256};
25
26// Local Imports
27use super::backend::ScanPage;
28use super::manager::{Inner, Operation, encoded_size};
29use super::query::FileSelection;
30use super::snapshot::Snapshot;
31use super::{ByteRange, FileBatch, FileCompletion, FileReference, StorageManager};
32use crate::FileReadBatch;
33use crate::local::LocalReader;
34use crate::{BackendFailure, Error, Result};
35use harmos_domain::model::storage::files::{FileHash, FileSource};
36
37/* === Constants === */
38
39/* === Macros === */
40
41/* === Type Aliases === */
42
43/* === Structs === */
44
45#[derive(Clone, Debug)]
46pub struct FileRead {
47    pub source: FileSource,
48    pub range: ByteRange,
49}
50
51pub struct FileReader {
52    context: crate::execution::Context,
53    source: Source,
54}
55
56enum Source {
57    Stored(Box<StoredReader>),
58    Local(LocalReader),
59}
60
61struct StoredReader {
62    inner: Arc<Inner>,
63
64    snapshot: Option<Snapshot>,
65    file: FileReference,
66    offset: u64,
67    end: u64,
68    bytes: u64,
69    verify: bool,
70    hash: Sha256,
71}
72
73/* === Enums === */
74
75/* === Traits === */
76
77/* === Implementations === */
78
79impl<C> StorageManager<C> {
80    #[tracing::instrument(name = "storage.file_reads.read_file", level = "debug", skip_all)]
81    pub async fn read_file(&self, request: FileRead) -> Result<FileReader> {
82        let context = crate::execution::Context::current();
83        let span = context.sync(|| tracing::debug_span!("storage.file.read_file", outcome = "cancelled"));
84        context.result(span, async { self.open_file(request).await }).await
85    }
86
87    /// Reads the supplied file object without changing the caller's cursor.
88    /// On Windows, `file` must allow a separate read open; exclusive handles are refused.
89    pub async fn read_opened_file(&self, file: std::fs::File, range: ByteRange) -> Result<FileReader> {
90        self.open_file_handle(file, range).await
91    }
92
93    pub(crate) async fn open_file_handle(&self, file: std::fs::File, range: ByteRange) -> Result<FileReader> {
94        let reader = LocalReader::opened(&self.inner, file, range).await?;
95        Ok(FileReader {
96            context: crate::execution::Context::current(),
97            source: Source::Local(reader),
98        })
99    }
100
101    pub(crate) async fn open_file(&self, request: FileRead) -> Result<FileReader> {
102        let source = match request.source {
103            FileSource::Stored(file) => Source::Stored(Box::new(self.read_stored_file(file, request.range).await?)),
104            FileSource::Local(path) => Source::Local(LocalReader::open(&self.inner, path, request.range).await?),
105        };
106        Ok(FileReader {
107            context: crate::execution::Context::current(),
108            source,
109        })
110    }
111
112    async fn read_stored_file(&self, file: FileReference, range: ByteRange) -> Result<StoredReader> {
113        let end = range.end(file.length)?;
114        self.inner.primary.history().check(&file.position.history)?;
115
116        // Accepted References Not Yet Saved Must Match The Queued Publication.
117        let (saved, pending) = {
118            let state = self.inner.state.lock().unwrap();
119            if !state.status.accepted.covers(&file.position)? {
120                return Err(invalid("file has not been admitted"));
121            }
122            let pending = state.queue.iter().find_map(|entry| match &entry.operation {
123                Operation::File(actual) if actual.as_slice()[0].position == file.position => Some(actual.as_slice()[0] == file),
124                _ => None,
125            });
126            (state.status.saved.clone(), pending)
127        };
128        if pending == Some(false) {
129            return Err(invalid("immutable metadata does not match a finished file"));
130        }
131
132        let snapshot = self.snapshot(&saved).await?;
133
134        // Saved References Must Be Visible In The Pinned Image.
135        if pending.is_none() {
136            let selection = FileSelection {
137                scope: file.metadata.scope.clone(),
138                item: Some(file.metadata.id.clone()),
139            };
140            let page = ScanPage {
141                after: file.position.sequence.saturating_sub(1),
142                rows: NonZeroUsize::MIN,
143                bytes: NonZeroUsize::new(encoded_size(&file)?.max(1)).unwrap(),
144            };
145            let (stored, _) = snapshot.scan::<FileReference>(&selection, page).await?;
146            if stored.first() != Some(&file) {
147                return Err(invalid("immutable metadata does not match the stored file"));
148            }
149        }
150        // Acquire The Backend's Immutable Read Lease Before Returning The Reader.
151        // The Byte Is Re-Read On Delivery So Digest/Offset Acknowledgment Is Unchanged.
152        snapshot.read_file(&file, range.offset, NonZeroUsize::MIN).await?;
153        Ok(StoredReader {
154            inner: Arc::clone(&self.inner),
155
156            snapshot: Some(snapshot),
157            offset: range.offset,
158            verify: range.offset == 0 && end == file.length,
159            end,
160            file,
161            bytes: 0,
162            hash: Sha256::new(),
163        })
164    }
165}
166
167impl FileReader {
168    pub(crate) fn observation(&self) -> Option<(u64, std::time::SystemTime)> {
169        match &self.source {
170            Source::Local(reader) => Some(reader.observation()),
171            Source::Stored(_) => None,
172        }
173    }
174
175    pub(crate) fn start_opened(&mut self, offset: u64) -> Result<()> {
176        match &mut self.source {
177            Source::Local(reader) => reader.start(offset),
178            Source::Stored(_) => Err(Error::InvalidInput {
179                field: "opened capture",
180                reason: "requires an opened source",
181            }),
182        }
183    }
184
185    pub(crate) async fn observation_unchanged(&self) -> Result<bool> {
186        match &self.source {
187            Source::Local(reader) => reader.observation_unchanged().await,
188            Source::Stored(_) => Ok(true),
189        }
190    }
191
192    pub(crate) fn stored_hash(&self) -> Result<Option<FileHash>> {
193        self.completion()?;
194        Ok(match &self.source {
195            Source::Stored(reader) => Some(FileHash {
196                bytes_read: reader.bytes,
197                sha256: reader.hash.clone().finalize().into(),
198            }),
199            Source::Local(_) => None,
200        })
201    }
202
203    pub(crate) fn is_local(&self) -> bool {
204        matches!(self.source, Source::Local(_))
205    }
206
207    pub fn offset(&self) -> u64 {
208        match &self.source {
209            Source::Stored(reader) => reader.offset,
210            Source::Local(reader) => reader.offset(),
211        }
212    }
213
214    pub async fn next(&mut self) -> Result<Option<FileReadBatch>> {
215        Ok(self.next_raw(false).await?.map(FileReadBatch::new))
216    }
217
218    pub(crate) async fn next_raw(&mut self, event: bool) -> Result<Option<FileBatch>> {
219        let context = self.context.clone();
220        let span = context.sync(|| tracing::trace_span!("storage.reader.page", outcome = "cancelled"));
221        let future = async {
222            match &mut self.source {
223                Source::Stored(reader) => reader.next(event).await,
224                Source::Local(reader) => reader.next(event).await,
225            }
226        };
227        context.result(span, future).await
228    }
229
230    pub(crate) async fn check_unchanged(&self) -> Result<()> {
231        match &self.source {
232            Source::Stored(_) => Ok(()),
233            Source::Local(reader) => reader.check_unchanged().await,
234        }
235    }
236
237    pub fn completion(&self) -> Result<FileCompletion> {
238        match &self.source {
239            Source::Stored(reader) => reader.completion(),
240            Source::Local(reader) => reader.completion(),
241        }
242    }
243}
244
245impl StoredReader {
246    async fn next(&mut self, event: bool) -> Result<Option<FileBatch>> {
247        let Some(snapshot) = &self.snapshot else {
248            return Ok(None);
249        };
250        if self.offset == self.end {
251            // An Empty Whole File Still Proves Its Digest.
252            if self.verify && self.end == 0 && <[u8; 32]>::from(self.hash.clone().finalize()) != self.file.sha256 {
253                return Err(corrupt("file digest mismatch"));
254            }
255            self.snapshot = None;
256            return Ok(None);
257        }
258
259        // The Response Envelope Shares The Batch Byte Ceiling.
260        let ceiling = self.inner.settings.response.batch_bytes.get();
261        let maximum = crate::file_batch::batch_limit(ceiling, self.offset, self.end - self.offset, event)?;
262
263        let bytes = snapshot.read_file(&self.file, self.offset, maximum).await?;
264        if bytes.is_empty() || bytes.capacity() > maximum.get() || bytes.len() as u64 > self.end - self.offset {
265            return Err(corrupt("file batch is empty or outside the immutable length"));
266        }
267        // Verify The Whole-File Digest Before Delivering Its Final Batch.
268        let mut hash = self.hash.clone();
269        hash.update(&bytes);
270        let next = self.offset + bytes.len() as u64;
271        if self.verify && next == self.end && <[u8; 32]>::from(hash.clone().finalize()) != self.file.sha256 {
272            return Err(corrupt("file digest mismatch"));
273        }
274        let offset = self.offset;
275        self.hash = hash;
276        self.offset = next;
277        self.bytes += bytes.len() as u64;
278        Ok(Some(FileBatch { offset, bytes }))
279    }
280
281    pub fn completion(&self) -> Result<FileCompletion> {
282        if self.offset != self.end || self.snapshot.is_some() {
283            return Err(Error::NotReady { domain: "file read completion" });
284        }
285        Ok(FileCompletion { bytes: self.bytes })
286    }
287}
288
289/* === Public Functions === */
290
291/* === Private Functions === */
292
293fn invalid(reason: &'static str) -> Error {
294    Error::InvalidInput { field: "file reference", reason }
295}
296
297fn corrupt(reason: &'static str) -> Error {
298    Error::backend("file read", BackendFailure::Corrupt, io::Error::new(io::ErrorKind::InvalidData, reason))
299}
300
301/* === Tests === */
harmos_storage/file_reads.rs

harmos_storage/
file_reads.rs
