harmos_storage/
file_production.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::{ErasedCatalog, FileBatch, FilePhase, FileProgress, StagedFile, StorageManager};
22use crate::FileReadBatch;
23use crate::{Error, Result};
24
25/* === Constants === */
26
27/* === Macros === */
28
29/* === Type Aliases === */
30
31/* === Structs === */
32
33pub struct ProductionFollower {
34    context: crate::execution::Context,
35    manager: StorageManager<ErasedCatalog>,
36    file: StagedFile,
37    offset: u64,
38    progress: FileProgress,
39    changed: watch::Receiver<u64>,
40}
41
42/* === Enums === */
43
44/* === Traits === */
45
46/* === Implementations === */
47
48impl<C> StorageManager<C> {
49    #[tracing::instrument(name = "storage.file_production.follow_production", level = "debug", skip_all)]
50    pub async fn follow_production(&self, file: &StagedFile, offset: u64) -> Result<ProductionFollower> {
51        self.open_production_follower(file, offset).await
52    }
53
54    pub(crate) async fn open_production_follower(&self, file: &StagedFile, offset: u64) -> Result<ProductionFollower> {
55        let context = crate::execution::Context::current();
56        let span = context.sync(|| tracing::debug_span!("storage.file.follow_production", outcome = "cancelled"));
57        context.result(span, async {
58        self.inner.primary.history().check(&file.history)?;
59
60        let changed = self.inner.changed.subscribe();
61        let progress = self.production_progress(file).await?;
62        if offset > progress.length {
63            tracing::error!(file = %file.id, offset, available = progress.length, reason = "offset exceeds retained bytes", "file following rejected");
64            return Err(Error::InvalidInput { field: "file offset", reason: "offset exceeds retained bytes" });
65        }
66        Ok(ProductionFollower {
67            context: crate::execution::Context::current(),
68            manager: self.erased(),
69            file: file.clone(),
70            offset,
71            progress,
72            changed,
73
74
75        })
76
77        }).await
78    }
79
80    #[tracing::instrument(name = "storage.file_production.file_progress", level = "debug", skip_all)]
81    pub async fn file_progress(&self, file: &StagedFile) -> Result<FileProgress> {
82        let context = crate::execution::Context::current();
83        let span = context.sync(|| tracing::debug_span!("storage.file.file_progress", outcome = "cancelled"));
84        context.result(span, async { self.production_progress(file).await }).await
85    }
86
87    pub(crate) async fn production_progress(&self, file: &StagedFile) -> Result<FileProgress> {
88        self.inner.primary.history().check(&file.history)?;
89        let _lifecycle = tracing::Instrument::instrument(self.inner.lifecycle.read(), tracing::trace_span!("storage.lock.wait")).await;
90        {
91            let state = self.inner.state.lock().unwrap();
92            if state.status.closing || !state.ready {
93                return Err(Error::Closing { domain: "storage file reader" });
94            }
95            if let Some(progress) = state.files.values().find_map(|active| active.production(file)) {
96                return progress;
97            }
98        }
99
100        self.inner.bounded("primary.file_progress", self.inner.primary.file_progress(file)).await
101    }
102}
103
104impl ProductionFollower {
105    pub fn offset(&self) -> u64 {
106        self.offset
107    }
108
109    pub fn progress(&self) -> FileProgress {
110        self.progress
111    }
112
113    pub async fn next_batch(&mut self, max_bytes: NonZeroUsize) -> Result<Option<FileReadBatch>> {
114        let context = self.context.clone();
115        let span = context.sync(|| tracing::trace_span!("storage.reader.page", outcome = "cancelled"));
116        let future = async {
117            loop {
118                self.changed.borrow_and_update();
119                let progress = self.manager.production_progress(&self.file).await?;
120                if self.offset > progress.length {
121                    return Err(Error::InvalidInput {
122                        field: "file offset",
123                        reason: "retained bytes moved behind reader offset",
124                    });
125                }
126                self.progress = progress;
127                if self.offset < progress.length {
128                    let ceiling = self.manager.inner.settings.response.batch_bytes.get();
129                    let remaining = (progress.length - self.offset).min(max_bytes.get() as u64);
130                    let maximum = crate::file_batch::batch_limit(ceiling, self.offset, remaining, false)?;
131                    let size = maximum.get();
132
133                    let _lifecycle =
134                        tracing::Instrument::instrument(self.manager.inner.lifecycle.read(), tracing::trace_span!("storage.lock.wait"))
135                            .await;
136                    if self.manager.inner.state.lock().unwrap().status.closing {
137                        return Err(Error::Closing { domain: "storage file reader" });
138                    }
139                    let bytes = self
140                        .manager
141                        .inner
142                        .bounded("primary.read_production", self.manager.inner.primary.read_production(&self.file, self.offset, maximum))
143                        .await?;
144                    if bytes.is_empty() || bytes.len() > size || bytes.capacity() > size {
145                        return Err(Error::InvalidInput {
146                            field: "file batch",
147                            reason: "backend returned an empty or oversized retained prefix",
148                        });
149                    }
150                    let offset = self.offset;
151                    self.offset += bytes.len() as u64;
152                    return Ok(Some(FileReadBatch::new(FileBatch { offset, bytes })));
153                }
154                if progress.phase != FilePhase::Producing {
155                    return Ok(None);
156                }
157                self.changed.changed().await.map_err(|_| Error::Closing { domain: "storage file reader" })?;
158            }
159        };
160        context.result(span, future).await
161    }
162}
163
164impl Drop for ProductionFollower {
165    fn drop(&mut self) {
166        tracing::debug!(file = %self.file.id, offset = self.offset, reason = "viewer disconnected", "file following stopped");
167    }
168}
169
170/* === Public Functions === */
171
172/* === Private Functions === */
173
174/* === Tests === */
harmos_storage/file_production.rs

harmos_storage/
file_production.rs
