harmos_storage/
file_production.rs1use std::num::NonZeroUsize;
18use tokio::sync::watch;
20use super::capacity::{Held, Pool, Refusal, Reservation};
22use super::{ErasedCatalog, FileBatch, FilePhase, FileProgress, StagedFile, StorageManager};
23use crate::FileReadBatch;
24use crate::{Error, MemorySize, Result};
25
26pub struct ProductionFollower {
35 context: crate::execution::Context,
36 manager: StorageManager<ErasedCatalog>,
37 file: StagedFile,
38 offset: u64,
39 progress: FileProgress,
40 changed: watch::Receiver<u64>,
41 held: Held,
42 _metadata: Reservation,
43}
44
45impl<C> StorageManager<C> {
52 #[tracing::instrument(name = "storage.file_production.follow_production", level = "debug", skip_all)]
53 pub async fn follow_production(&self, file: &StagedFile, offset: u64) -> Result<ProductionFollower> {
54 let context = crate::execution::Context::current();
55 let span = context.sync(|| tracing::debug_span!("storage.file.follow_production", outcome = "cancelled"));
56 context.result(span, async {
57 self.inner.primary.history().check(&file.history)?;
58 let held = Held::default();
59 let metadata = Reservation::wait(
60 &self.inner,
61 Pool::Read,
62 &held,
63 file.memory_size().saturating_add(std::mem::size_of::<ProductionFollower>()),
64 None,
65 )
66 .await
67 .map_err(Refusal::into_error)?;
68 let changed = self.inner.changed.subscribe();
69 let progress = self.file_progress_with_held(file, &held).await?;
70 if offset > progress.length {
71 tracing::error!(file = %file.id, offset, available = progress.length, reason = "offset exceeds retained bytes", "file following rejected");
72 return Err(Error::InvalidInput { field: "file offset", reason: "offset exceeds retained bytes" });
73 }
74 Ok(ProductionFollower {
75 context: crate::execution::Context::current(),
76 manager: self.erased(),
77 file: file.clone(),
78 offset,
79 progress,
80 changed,
81 held,
82 _metadata: metadata,
83 })
84
85 }).await
86 }
87
88 #[tracing::instrument(name = "storage.file_production.file_progress", level = "debug", skip_all)]
89 pub async fn file_progress(&self, file: &StagedFile) -> Result<FileProgress> {
90 let context = crate::execution::Context::current();
91 let span = context.sync(|| tracing::debug_span!("storage.file.file_progress", outcome = "cancelled"));
92 context.result(span, async { self.file_progress_with_held(file, &Held::default()).await }).await
93 }
94
95 async fn file_progress_with_held(&self, file: &StagedFile, held: &Held) -> Result<FileProgress> {
96 self.inner.primary.history().check(&file.history)?;
97 let _lifecycle = tracing::Instrument::instrument(self.inner.lifecycle.read(), tracing::trace_span!("storage.lock.wait")).await;
98 {
99 let state = self.inner.state.lock().unwrap();
100 if state.status.closing || !state.ready {
101 return Err(Error::Closing { domain: "storage file reader" });
102 }
103 if let Some(progress) = state.files.values().find_map(|active| active.production(file)) {
104 return progress;
105 }
106 }
107 let charge = file.memory_size().saturating_add(std::mem::size_of::<FileProgress>() + crate::manager::ENTRY_OVERHEAD);
108 let memory = Reservation::wait(&self.inner, Pool::Read, held, charge, None).await.map_err(Refusal::into_error)?.retain();
109 self.inner.bounded("primary.file_progress", self.inner.primary.file_progress(file, memory)).await
110 }
111}
112
113impl ProductionFollower {
114 pub fn offset(&self) -> u64 {
115 self.offset
116 }
117
118 pub fn progress(&self) -> FileProgress {
119 self.progress
120 }
121
122 pub async fn next_batch(&mut self, max_bytes: NonZeroUsize) -> Result<Option<FileReadBatch>> {
123 let context = self.context.clone();
124 let span = context.sync(|| tracing::trace_span!("storage.reader.page", outcome = "cancelled"));
125 let future = async {
126 loop {
127 self.changed.borrow_and_update();
128 let progress = self.manager.file_progress_with_held(&self.file, &self.held).await?;
129 if self.offset > progress.length {
130 return Err(Error::InvalidInput { field: "file offset", reason: "retained bytes moved behind reader offset" });
131 }
132 self.progress = progress;
133 if self.offset < progress.length {
134 let ceiling = self.manager.inner.settings.response.batch_bytes.get();
135 let remaining = (progress.length - self.offset).min(max_bytes.get() as u64);
136 let maximum = crate::file_batch::batch_limit(ceiling, self.offset, remaining, false)?;
137 let size = maximum.get();
138 let transfer = Reservation::wait(
139 &self.manager.inner,
140 Pool::Read,
141 &self.held,
142 size.saturating_add(std::mem::size_of::<FileReadBatch>() + crate::manager::ENTRY_OVERHEAD),
143 None,
144 )
145 .await
146 .map_err(Refusal::into_error)?
147 .retain();
148 let _lifecycle =
149 tracing::Instrument::instrument(self.manager.inner.lifecycle.read(), tracing::trace_span!("storage.lock.wait"))
150 .await;
151 if self.manager.inner.state.lock().unwrap().status.closing {
152 return Err(Error::Closing { domain: "storage file reader" });
153 }
154 let bytes = self
155 .manager
156 .inner
157 .bounded(
158 "primary.read_production",
159 self.manager.inner.primary.read_production(&self.file, self.offset, maximum, transfer.clone()),
160 )
161 .await?;
162 if bytes.is_empty() || bytes.len() > size || bytes.capacity() > size {
163 return Err(Error::InvalidInput {
164 field: "file batch",
165 reason: "backend returned an empty or oversized retained prefix",
166 });
167 }
168 let offset = self.offset;
169 self.offset += bytes.len() as u64;
170 return Ok(Some(FileReadBatch::retained(FileBatch { offset, bytes }, transfer)));
171 }
172 if progress.phase != FilePhase::Producing {
173 return Ok(None);
174 }
175 self.changed.changed().await.map_err(|_| Error::Closing { domain: "storage file reader" })?;
176 }
177 };
178 context.result(span, future).await
179 }
180}
181
182impl Drop for ProductionFollower {
183 fn drop(&mut self) {
184 tracing::debug!(file = %self.file.id, offset = self.offset, reason = "viewer disconnected", "file following stopped");
185 }
186}
187
188