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::capacity::{Held, Pool, Refusal, Reservation};
29use super::manager::{ENTRY_OVERHEAD, Inner, Operation, encoded_size};
30use super::query::FileSelection;
31use super::snapshot::Snapshot;
32use super::{ByteRange, FileBatch, FileCompletion, FileReference, StorageManager};
33use crate::FileReadBatch;
34use crate::arrow::RetainedMemory;
35use crate::local::LocalReader;
36use crate::{BackendFailure, Error, MemorySize, Result};
37use harmos_domain::model::storage::files::{FileHash, FileSource};
38use harmos_domain::operations::storage::ReadFile;
39
40/* === Constants === */
41
42/* === Macros === */
43
44/* === Type Aliases === */
45
46/* === Structs === */
47
48pub struct FileReader {
49    context: crate::execution::Context,
50    source: Source,
51}
52
53enum Source {
54    Stored(Box<StoredReader>),
55    Local(LocalReader),
56}
57
58struct StoredReader {
59    inner: Arc<Inner>,
60    held: Held,
61    snapshot: Option<Snapshot>,
62    file: FileReference,
63    offset: u64,
64    end: u64,
65    bytes: u64,
66    verify: bool,
67    hash: Sha256,
68    _metadata: RetainedMemory,
69}
70
71/* === Enums === */
72
73/* === Traits === */
74
75/* === Implementations === */
76
77impl<C> StorageManager<C> {
78    #[tracing::instrument(name = "storage.file_reads.read_file", level = "debug", skip_all)]
79    pub async fn read_file(&self, request: ReadFile) -> Result<FileReader> {
80        let context = crate::execution::Context::current();
81        let span = context.sync(|| tracing::debug_span!("storage.file.read_file", outcome = "cancelled"));
82        context.result(span, async { self.read_file_with_held(request, Held::default()).await }).await
83    }
84
85    pub(crate) async fn read_file_with_held(&self, request: ReadFile, held: Held) -> Result<FileReader> {
86        let source = match request.source {
87            FileSource::Stored(file) => Source::Stored(Box::new(self.read_stored_file(file, request.range, held).await?)),
88            FileSource::Local(path) => Source::Local(LocalReader::open(&self.inner, path, request.range, held).await?),
89        };
90        Ok(FileReader { context: crate::execution::Context::current(), source })
91    }
92
93    async fn read_stored_file(&self, file: FileReference, range: ByteRange, held: Held) -> Result<StoredReader> {
94        let end = range.end(file.length)?;
95        self.inner.primary.history().check(&file.position.history)?;
96
97        // Accepted References Not Yet Saved Must Match The Queued Publication.
98        let (saved, pending) = {
99            let state = self.inner.state.lock().unwrap();
100            if !state.status.accepted.covers(&file.position)? {
101                return Err(invalid("file has not been admitted"));
102            }
103            let pending = state.queue.iter().find_map(|entry| match &entry.operation {
104                Operation::File(actual) if actual.as_slice()[0].position == file.position => Some(actual.as_slice()[0] == file),
105                _ => None,
106            });
107            (state.status.saved.clone(), pending)
108        };
109        if pending == Some(false) {
110            return Err(invalid("immutable metadata does not match a finished file"));
111        }
112        let metadata = Reservation::wait(
113            &self.inner,
114            Pool::Read,
115            &held,
116            file.memory_size().saturating_mul(3).saturating_add(std::mem::size_of::<StoredReader>() + ENTRY_OVERHEAD),
117            None,
118        )
119        .await
120        .map_err(Refusal::into_error)?
121        .retain();
122        let snapshot = self.snapshot(&saved, &held).await?;
123
124        // Saved References Must Be Visible In The Pinned Image.
125        if pending.is_none() {
126            let selection = FileSelection { scope: file.metadata.scope.clone(), item: Some(file.metadata.id.clone()) };
127            let page = ScanPage {
128                after: file.position.sequence.saturating_sub(1),
129                rows: NonZeroUsize::MIN,
130                bytes: NonZeroUsize::new(encoded_size(&file)?.max(1)).unwrap(),
131            };
132            let (stored, _) = snapshot.scan::<FileReference>(&selection, page, metadata.clone()).await?;
133            if stored.first() != Some(&file) {
134                return Err(invalid("immutable metadata does not match the stored file"));
135            }
136        }
137        Ok(StoredReader {
138            inner: Arc::clone(&self.inner),
139            held,
140            snapshot: Some(snapshot),
141            offset: range.offset,
142            verify: range.offset == 0 && end == file.length,
143            end,
144            file,
145            bytes: 0,
146            hash: Sha256::new(),
147            _metadata: metadata,
148        })
149    }
150}
151
152impl FileReader {
153    pub(crate) fn stored_hash(&self) -> Result<Option<FileHash>> {
154        self.completion()?;
155        Ok(match &self.source {
156            Source::Stored(reader) => Some(FileHash { bytes_read: reader.bytes, sha256: reader.hash.clone().finalize().into() }),
157            Source::Local(_) => None,
158        })
159    }
160
161    pub(crate) fn is_local(&self) -> bool {
162        matches!(self.source, Source::Local(_))
163    }
164
165    pub fn offset(&self) -> u64 {
166        match &self.source {
167            Source::Stored(reader) => reader.offset,
168            Source::Local(reader) => reader.offset(),
169        }
170    }
171
172    pub async fn next(&mut self) -> Result<Option<FileReadBatch>> {
173        Ok(self.next_raw(false).await?.map(|(batch, memory)| FileReadBatch::retained(batch, memory)))
174    }
175
176    pub(crate) async fn next_raw(&mut self, event: bool) -> Result<Option<(FileBatch, RetainedMemory)>> {
177        let context = self.context.clone();
178        let span = context.sync(|| tracing::trace_span!("storage.reader.page", outcome = "cancelled"));
179        let future = async {
180            match &mut self.source {
181                Source::Stored(reader) => reader.next(event).await,
182                Source::Local(reader) => reader.next(event).await,
183            }
184        };
185        context.result(span, future).await
186    }
187
188    pub(crate) async fn check_unchanged(&self) -> Result<()> {
189        match &self.source {
190            Source::Stored(_) => Ok(()),
191            Source::Local(reader) => reader.check_unchanged().await,
192        }
193    }
194
195    pub fn completion(&self) -> Result<FileCompletion> {
196        match &self.source {
197            Source::Stored(reader) => reader.completion(),
198            Source::Local(reader) => reader.completion(),
199        }
200    }
201}
202
203impl StoredReader {
204    async fn next(&mut self, event: bool) -> Result<Option<(FileBatch, RetainedMemory)>> {
205        let Some(snapshot) = &self.snapshot else {
206            return Ok(None);
207        };
208        if self.offset == self.end {
209            // An Empty Whole File Still Proves Its Digest.
210            if self.verify && self.end == 0 && <[u8; 32]>::from(self.hash.clone().finalize()) != self.file.sha256 {
211                return Err(corrupt("file digest mismatch"));
212            }
213            self.snapshot = None;
214            return Ok(None);
215        }
216
217        // The Response Envelope Shares The Batch Byte Ceiling.
218        let ceiling = self.inner.settings.response.batch_bytes.get();
219        let maximum = crate::file_batch::batch_limit(ceiling, self.offset, self.end - self.offset, event)?;
220        let charge = maximum.get().saturating_mul(2).saturating_add(std::mem::size_of::<FileBatch>() + ENTRY_OVERHEAD);
221        let transfer = Reservation::wait(&self.inner, Pool::Read, &self.held, charge, None).await.map_err(Refusal::into_error)?.retain();
222        let bytes = snapshot.read_file(&self.file, self.offset, maximum, transfer.clone()).await?;
223        if bytes.is_empty() || bytes.capacity() > maximum.get() || bytes.len() as u64 > self.end - self.offset {
224            return Err(corrupt("file batch is empty or outside the immutable length"));
225        }
226        // Verify The Whole-File Digest Before Delivering Its Final Batch.
227        let mut hash = self.hash.clone();
228        hash.update(&bytes);
229        let next = self.offset + bytes.len() as u64;
230        if self.verify && next == self.end && <[u8; 32]>::from(hash.clone().finalize()) != self.file.sha256 {
231            return Err(corrupt("file digest mismatch"));
232        }
233        let offset = self.offset;
234        self.hash = hash;
235        self.offset = next;
236        self.bytes += bytes.len() as u64;
237        Ok(Some((FileBatch { offset, bytes }, transfer)))
238    }
239
240    pub fn completion(&self) -> Result<FileCompletion> {
241        if self.offset != self.end || self.snapshot.is_some() {
242            return Err(Error::NotReady { domain: "file read completion" });
243        }
244        Ok(FileCompletion { bytes: self.bytes })
245    }
246}
247
248/* === Public Functions === */
249
250/* === Private Functions === */
251
252fn invalid(reason: &'static str) -> Error {
253    Error::InvalidInput { field: "file reference", reason }
254}
255
256fn corrupt(reason: &'static str) -> Error {
257    Error::backend("file read", BackendFailure::Corrupt, io::Error::new(io::ErrorKind::InvalidData, reason))
258}
259
260/* === Tests === */
harmos_storage/file_reads.rs

harmos_storage/
file_reads.rs
