harmos_storage/
file_follow.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::num::NonZeroUsize;
18// Third Party Imports
19use tokio::sync::watch;
20// Local Imports
21use super::capacity::{Held, Pool, Refusal, Reservation};
22use super::{ErasedCatalog, FileBatch, FilePhase, FileProgress, StagedFile, StorageManager};
23use crate::{Error, MemorySize, Result};
24
25/* === Constants === */
26
27/* === Macros === */
28
29/* === Type Aliases === */
30
31/* === Structs === */
32
33pub struct FileFollower {
34    manager: StorageManager<ErasedCatalog>,
35    file: StagedFile,
36    offset: u64,
37    progress: FileProgress,
38    changed: watch::Receiver<u64>,
39    held: Held,
40    _metadata: Reservation,
41}
42
43/* === Enums === */
44
45/* === Traits === */
46
47/* === Implementations === */
48
49impl<C> StorageManager<C> {
50    pub async fn follow_file(&self, file: &StagedFile, offset: u64) -> Result<FileFollower> {
51        self.inner.primary.history().check(&file.history)?;
52        let held = Held::default();
53        let metadata =
54            Reservation::wait(&self.inner, Pool::Read, &held, file.memory_size().saturating_add(std::mem::size_of::<FileFollower>()), None)
55                .await
56                .map_err(Refusal::into_error)?;
57        let changed = self.inner.changed.subscribe();
58        let progress = self.file_progress(file).await?;
59        if offset > progress.length {
60            tracing::error!(file = %file.id, offset, available = progress.length, reason = "offset exceeds retained bytes", "file following rejected");
61            return Err(Error::InvalidInput { field: "file offset", reason: "offset exceeds retained bytes" });
62        }
63        Ok(FileFollower { manager: self.erased(), file: file.clone(), offset, progress, changed, held, _metadata: metadata })
64    }
65
66    pub async fn file_progress(&self, file: &StagedFile) -> Result<FileProgress> {
67        self.inner.primary.history().check(&file.history)?;
68        let _lifecycle = self.inner.lifecycle.read().await;
69        {
70            let state = self.inner.state.lock().unwrap();
71            if state.status.closing || !state.ready {
72                return Err(Error::Closing { domain: "storage file reader" });
73            }
74            if let Some(progress) = state.files.values().find_map(|active| active.production(file)) {
75                return progress;
76            }
77        }
78        self.inner.bounded(self.inner.primary.file_progress(file)).await
79    }
80}
81
82impl FileFollower {
83    pub fn offset(&self) -> u64 {
84        self.offset
85    }
86
87    pub fn progress(&self) -> FileProgress {
88        self.progress
89    }
90
91    pub async fn next_batch(&mut self, max_bytes: NonZeroUsize) -> Result<Option<FileBatch>> {
92        loop {
93            self.changed.borrow_and_update();
94            let progress = self.manager.file_progress(&self.file).await?;
95            if self.offset > progress.length {
96                return Err(Error::InvalidInput { field: "file offset", reason: "retained bytes moved behind reader offset" });
97            }
98            self.progress = progress;
99            if self.offset < progress.length {
100                let size = max_bytes
101                    .get()
102                    .min(self.manager.inner.settings.response.batch_bytes.get())
103                    .min(usize::try_from(progress.length - self.offset).unwrap_or(usize::MAX));
104                let maximum = NonZeroUsize::new(size).expect("nonempty file prefix");
105                let _transfer =
106                    Reservation::wait(&self.manager.inner, Pool::Read, &self.held, size, None).await.map_err(Refusal::into_error)?;
107                let _lifecycle = self.manager.inner.lifecycle.read().await;
108                if self.manager.inner.state.lock().unwrap().status.closing {
109                    return Err(Error::Closing { domain: "storage file reader" });
110                }
111                let bytes =
112                    self.manager.inner.bounded(self.manager.inner.primary.read_production(&self.file, self.offset, maximum)).await?;
113                if bytes.is_empty() || bytes.len() > size || bytes.capacity() > size {
114                    return Err(Error::InvalidInput {
115                        field: "file batch",
116                        reason: "backend returned an empty or oversized retained prefix",
117                    });
118                }
119                let offset = self.offset;
120                self.offset += bytes.len() as u64;
121                return Ok(Some(FileBatch { offset, bytes }));
122            }
123            if progress.phase != FilePhase::Producing {
124                return Ok(None);
125            }
126            self.changed.changed().await.map_err(|_| Error::Closing { domain: "storage file reader" })?;
127        }
128    }
129}
130
131impl Drop for FileFollower {
132    fn drop(&mut self) {
133        tracing::debug!(file = %self.file.id, offset = self.offset, reason = "viewer disconnected", "file following stopped");
134    }
135}
136
137/* === Public Functions === */
138
139/* === Private Functions === */
140
141/* === Tests === */
harmos_storage/file_follow.rs

harmos_storage/
file_follow.rs
