1use std::path::PathBuf;
20use std::sync::Arc;
21
22use crate::arrow::RetainedMemory;
26use crate::capacity::{Held, Pool, Refusal, Reservation};
27use crate::local::LocalReader;
28use crate::manager::{ENTRY_OVERHEAD, Inner};
29use crate::{ByteRange, Error, FileBatch, FileCompletion, FileReadEvent, FileReader, MemorySize, Result, StorageManager};
30use harmos_domain::model::storage::files::{FileEvent, FileSource, FollowStart};
31use harmos_domain::operations::storage::{FollowFile, ReadFile};
32
33pub struct FileFollower {
42 context: crate::execution::Context,
43 inner: Arc<Inner>,
44 source: Source,
45 held: Held,
46 delivered: u64,
47 finished: bool,
48 _description: Reservation,
49}
50
51struct Local {
52 path: PathBuf,
53 reader: Option<LocalReader>,
54 missing: bool,
55}
56
57enum Source {
60 Stored(FileReader),
61 Local(Local),
62}
63
64enum Step {
65 Data(FileBatch, RetainedMemory),
66 Signal(FileEvent),
67 Wait,
68 Done,
69}
70
71impl<C> StorageManager<C> {
76 #[tracing::instrument(name = "storage.file_follow.follow_file", level = "debug", skip_all)]
77 pub async fn follow_file(&self, request: FollowFile) -> Result<FileFollower> {
78 let context = crate::execution::Context::current();
79 let span = context.sync(|| tracing::debug_span!("storage.file.follow_file", outcome = "cancelled"));
80 context
81 .result(span, async {
82 let held = Held::default();
83 let source = match request.source {
84 FileSource::Local(path) => FileSource::Local(std::path::absolute(path).map_err(crate::local::io_error)?),
85 stored => stored,
86 };
87 let path_bytes = match &source {
88 FileSource::Local(path) => path.capacity(),
89 FileSource::Stored(_) => 0,
90 };
91 let charge = path_bytes.saturating_add(std::mem::size_of::<FileFollower>() + ENTRY_OVERHEAD);
92 let description = Reservation::wait(&self.inner, Pool::Read, &held, charge, None).await.map_err(Refusal::into_error)?;
93 let source = match source {
94 FileSource::Stored(file) => {
95 let offset = start(request.start, file.length);
96 let reader = self
97 .read_file_with_held(
98 ReadFile { source: FileSource::Stored(file), range: ByteRange { offset, limit: None } },
99 held.clone(),
100 )
101 .await?;
102 Source::Stored(reader)
103 }
104 FileSource::Local(path) => {
105 let mut reader =
106 LocalReader::open(&self.inner, path.clone(), ByteRange { offset: 0, limit: None }, held.clone()).await?;
107 reader.start(start(request.start, reader.length()))?;
108 Source::Local(Local { path, reader: Some(reader), missing: false })
109 }
110 };
111 Ok(FileFollower {
112 context: crate::execution::Context::current(),
113 inner: Arc::clone(&self.inner),
114 source,
115 held,
116 delivered: 0,
117 finished: false,
118 _description: description,
119 })
120 })
121 .await
122 }
123}
124
125impl FileFollower {
126 pub async fn next(&mut self) -> Result<Option<FileReadEvent>> {
127 let context = self.context.clone();
128 let span = context.sync(|| tracing::trace_span!("storage.reader.page", outcome = "cancelled"));
129 let future = async {
130 loop {
131 if self.finished {
132 return Ok(None);
133 }
134 if self.inner.state.lock().unwrap().status.closing {
135 return Err(Error::Closing { domain: "file follower" });
136 }
137 let step = match &mut self.source {
138 Source::Stored(reader) => match reader.next_raw(true).await? {
139 Some((batch, memory)) => Step::Data(batch, memory),
140 None => Step::Done,
141 },
142 Source::Local(local) => local.next(&self.inner, &self.held).await?,
143 };
144 match step {
145 Step::Data(batch, memory) => {
146 self.delivered += batch.bytes.len() as u64;
147 return Ok(Some(FileReadEvent::retained(FileEvent::Data(batch), memory)));
148 }
149 Step::Signal(event) => {
150 let charge = event.memory_size().saturating_add(ENTRY_OVERHEAD);
151 let memory =
152 Reservation::wait(&self.inner, Pool::Read, &self.held, charge, None).await.map_err(Refusal::into_error)?;
153 return Ok(Some(FileReadEvent::new(event, memory)));
154 }
155 Step::Done => {
156 self.finished = true;
157 return Ok(None);
158 }
159 Step::Wait => {
160 let mut changed = self.inner.changed.subscribe();
161 tokio::select! {
162 _ = tokio::time::sleep(self.inner.settings.filesystem.poll_interval) => {},
163 _ = changed.changed() => {},
164 }
165 }
166 }
167 }
168 };
169 context.result(span, future).await
170 }
171
172 pub fn completion(&self) -> Result<FileCompletion> {
173 if !self.finished {
174 return Err(Error::NotReady { domain: "file follow completion" });
175 }
176 Ok(FileCompletion { bytes: self.delivered })
177 }
178}
179
180impl Local {
181 async fn next(&mut self, inner: &Arc<Inner>, held: &Held) -> Result<Step> {
182 if self.reader.is_none() {
183 self.reader = Some(LocalReader::open(inner, self.path.clone(), ByteRange { offset: 0, limit: None }, held.clone()).await?);
184 }
185 let reader = self.reader.as_ref().expect("local follower retains its current file");
186 let Some((same, length)) = reader.inspect().await? else {
187 if self.missing {
188 return Ok(Step::Wait);
189 }
190 self.missing = true;
191 return Ok(Step::Signal(FileEvent::Replaced));
192 };
193 if !same {
194 self.reader.take();
195 self.reader = Some(LocalReader::open(inner, self.path.clone(), ByteRange { offset: 0, limit: None }, held.clone()).await?);
196 let signaled = std::mem::replace(&mut self.missing, false);
197 return Ok(if signaled { Step::Wait } else { Step::Signal(FileEvent::Replaced) });
198 }
199 let reader = self.reader.as_mut().expect("local follower retains its current file");
200 self.missing = false;
201 if length < reader.offset() {
202 self.reader.take();
203 self.reader = Some(LocalReader::open(inner, self.path.clone(), ByteRange { offset: 0, limit: None }, held.clone()).await?);
204 return Ok(Step::Signal(FileEvent::Truncated { length }));
205 }
206 reader.retarget(length);
207 match reader.next(true).await? {
208 Some((batch, memory)) => Ok(Step::Data(batch, memory)),
209 None => Ok(Step::Wait),
210 }
211 }
212}
213
214fn start(start: FollowStart, length: u64) -> u64 {
219 match start {
220 FollowStart::Beginning => 0,
221 FollowStart::End => length,
222 FollowStart::At(offset) => offset,
223 }
224}
225
226