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::{FileBatch, FileCompletion, FileReference, StorageManager};
33use crate::{BackendFailure, Error, Result};
34use harmos_domain::operations::storage::ReadFile;
35
36/* === Constants === */
37
38/* === Macros === */
39
40/* === Type Aliases === */
41
42/* === Structs === */
43
44pub struct FileReader {
45    inner: Arc<Inner>,
46    held: Held,
47    snapshot: Option<Snapshot>,
48    file: FileReference,
49    offset: u64,
50    end: u64,
51    bytes: u64,
52    verify: bool,
53    hash: Sha256,
54    _metadata: Reservation,
55}
56
57/* === Enums === */
58
59/* === Traits === */
60
61/* === Implementations === */
62
63impl<C> StorageManager<C> {
64    pub async fn read_file(&self, request: ReadFile) -> Result<FileReader> {
65        let ReadFile { file, range } = request;
66        let end = range.end(file.length)?;
67        self.inner.primary.history().check(&file.position.history)?;
68
69        // Accepted References Not Yet Saved Must Match The Queued Publication.
70        let (saved, pending) = {
71            let state = self.inner.state.lock().unwrap();
72            if !state.status.accepted.covers(&file.position)? {
73                return Err(invalid("file has not been admitted"));
74            }
75            let pending = state.queue.iter().find_map(|entry| match &entry.operation {
76                Operation::File(actual) if actual.position == file.position => Some(actual == &file),
77                _ => None,
78            });
79            (state.status.saved.clone(), pending)
80        };
81        if pending == Some(false) {
82            return Err(invalid("immutable metadata does not match a finished file"));
83        }
84        let held = Held::default();
85        let metadata =
86            Reservation::wait(&self.inner, Pool::Read, &held, encoded_size(&file)?.saturating_mul(2).saturating_add(ENTRY_OVERHEAD), None)
87                .await
88                .map_err(Refusal::into_error)?;
89        let snapshot = self.snapshot(&saved).await?;
90
91        // Saved References Must Be Visible In The Pinned Image.
92        if pending.is_none() {
93            let selection = FileSelection { scope: file.metadata.scope.clone(), item: Some(file.metadata.id.clone()) };
94            let page = ScanPage {
95                after: file.position.sequence.saturating_sub(1),
96                rows: NonZeroUsize::MIN,
97                bytes: NonZeroUsize::new(encoded_size(&file)?.max(1)).unwrap(),
98            };
99            let (stored, _) = snapshot.scan::<FileReference>(&selection, page).await?;
100            if stored.first() != Some(&file) {
101                return Err(invalid("immutable metadata does not match the stored file"));
102            }
103        }
104        Ok(FileReader {
105            inner: Arc::clone(&self.inner),
106            held,
107            snapshot: Some(snapshot),
108            offset: range.offset,
109            verify: range.offset == 0 && end == file.length,
110            end,
111            file,
112            bytes: 0,
113            hash: Sha256::new(),
114            _metadata: metadata,
115        })
116    }
117}
118
119impl FileReader {
120    pub fn reference(&self) -> &FileReference {
121        &self.file
122    }
123
124    pub fn offset(&self) -> u64 {
125        self.offset
126    }
127
128    pub async fn next(&mut self) -> Result<Option<FileBatch>> {
129        let Some(snapshot) = &self.snapshot else {
130            return Ok(None);
131        };
132        if self.offset == self.end {
133            // An Empty Whole File Still Proves Its Digest.
134            if self.verify && self.end == 0 && <[u8; 32]>::from(self.hash.clone().finalize()) != self.file.sha256 {
135                return Err(corrupt("file digest mismatch"));
136            }
137            self.snapshot = None;
138            return Ok(None);
139        }
140
141        // The Response Envelope Shares The Batch Byte Ceiling.
142        let ceiling = self.inner.settings.response.batch_bytes.get();
143        let overhead = 2 + varint_bytes(self.offset) + varint_bytes(ceiling as u64);
144        let remaining = usize::try_from(self.end - self.offset).unwrap_or(usize::MAX);
145        let maximum = ceiling
146            .checked_sub(overhead)
147            .map(|bytes| bytes.min(remaining))
148            .and_then(NonZeroUsize::new)
149            .ok_or(Error::ResourceExhausted { resource: "file response batch bytes" })?;
150        let charge = maximum.get().saturating_mul(2).saturating_add(std::mem::size_of::<FileBatch>() + ENTRY_OVERHEAD);
151        let _transfer = Reservation::wait(&self.inner, Pool::Read, &self.held, charge, None).await.map_err(Refusal::into_error)?;
152        let bytes = snapshot.read_file(&self.file, self.offset, maximum).await?;
153        if bytes.is_empty() || bytes.len() > maximum.get() || bytes.len() as u64 > self.end - self.offset {
154            return Err(corrupt("file batch is empty or outside the immutable length"));
155        }
156        // Verify The Whole-File Digest Before Delivering Its Final Batch.
157        let mut hash = self.hash.clone();
158        hash.update(&bytes);
159        let next = self.offset + bytes.len() as u64;
160        if self.verify && next == self.end && <[u8; 32]>::from(hash.clone().finalize()) != self.file.sha256 {
161            return Err(corrupt("file digest mismatch"));
162        }
163        let offset = self.offset;
164        self.hash = hash;
165        self.offset = next;
166        self.bytes += bytes.len() as u64;
167        Ok(Some(FileBatch { offset, bytes }))
168    }
169
170    pub fn completion(&self) -> Result<FileCompletion> {
171        if self.offset != self.end || self.snapshot.is_some() {
172            return Err(Error::NotReady { domain: "file read completion" });
173        }
174        Ok(FileCompletion { bytes: self.bytes })
175    }
176}
177
178/* === Public Functions === */
179
180/* === Private Functions === */
181
182fn varint_bytes(mut value: u64) -> usize {
183    let mut bytes = 1;
184    while value >= 128 {
185        value >>= 7;
186        bytes += 1;
187    }
188    bytes
189}
190
191fn invalid(reason: &'static str) -> Error {
192    Error::InvalidInput { field: "file reference", reason }
193}
194
195fn corrupt(reason: &'static str) -> Error {
196    Error::backend("file read", BackendFailure::Corrupt, io::Error::new(io::ErrorKind::InvalidData, reason))
197}
198
199/* === Tests === */
harmos_storage/file_reads.rs

harmos_storage/
file_reads.rs
