harmos_storage/
file_follow.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Follow Immutable Storage Or A Local Path With Explicit Identity Changes.
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::path::PathBuf;
20use std::sync::Arc;
21
22// Third Party Imports
23
24// Local Imports
25use crate::local::LocalReader;
26use crate::manager::Inner;
27use crate::{ByteRange, Error, FileBatch, FileCompletion, FileRead, FileReadEvent, FileReader, Result, StorageManager};
28use harmos_domain::model::storage::files::{FileEvent, FileSource, FollowStart};
29
30/* === Constants === */
31
32/* === Macros === */
33
34/* === Type Aliases === */
35
36/* === Structs === */
37
38#[derive(Clone, Debug)]
39pub struct FileFollow {
40    pub source: FileSource,
41    pub start: crate::FollowStart,
42}
43
44pub struct FileFollower {
45    context: crate::execution::Context,
46    inner: Arc<Inner>,
47    source: Source,
48
49    delivered: u64,
50    finished: bool,
51}
52
53struct Local {
54    path: PathBuf,
55    reader: Option<LocalReader>,
56    missing: bool,
57}
58
59/* === Enums === */
60
61enum Source {
62    Stored(FileReader),
63    Local(Local),
64}
65
66enum Step {
67    Data(FileBatch),
68    Signal(FileEvent),
69    Wait,
70    Done,
71}
72
73/* === Traits === */
74
75/* === Implementations === */
76
77impl<C> StorageManager<C> {
78    #[tracing::instrument(name = "storage.file_follow.follow_file", level = "debug", skip_all)]
79    pub async fn follow_file(&self, request: FileFollow) -> Result<FileFollower> {
80        self.open_follower(request).await
81    }
82
83    pub(crate) async fn open_follower(&self, request: FileFollow) -> Result<FileFollower> {
84        let context = crate::execution::Context::current();
85        let span = context.sync(|| tracing::debug_span!("storage.file.follow_file", outcome = "cancelled"));
86        context
87            .result(span, async {
88                let source = match request.source {
89                    FileSource::Local(path) => FileSource::Local(std::path::absolute(path).map_err(crate::local::io_error)?),
90                    stored => stored,
91                };
92
93                let source = match source {
94                    FileSource::Stored(file) => {
95                        let offset = start(request.start, file.length);
96                        let reader = self
97                            .open_file(FileRead {
98                                source: FileSource::Stored(file),
99                                range: ByteRange { offset, limit: None },
100                            })
101                            .await?;
102                        Source::Stored(reader)
103                    }
104                    FileSource::Local(path) => {
105                        let mut reader = LocalReader::open(&self.inner, path.clone(), ByteRange { offset: 0, limit: None }).await?;
106                        reader.start(start(request.start, reader.length()))?;
107                        Source::Local(Local {
108                            path,
109                            reader: Some(reader),
110                            missing: false,
111                        })
112                    }
113                };
114                Ok(FileFollower {
115                    context: crate::execution::Context::current(),
116                    inner: Arc::clone(&self.inner),
117                    source,
118
119                    delivered: 0,
120                    finished: false,
121                })
122            })
123            .await
124    }
125}
126
127impl FileFollower {
128    pub async fn next(&mut self) -> Result<Option<FileReadEvent>> {
129        let context = self.context.clone();
130        let span = context.sync(|| tracing::trace_span!("storage.reader.page", outcome = "cancelled"));
131        let future = async {
132            loop {
133                if self.finished {
134                    return Ok(None);
135                }
136                if self.inner.state.lock().unwrap().status.closing {
137                    return Err(Error::Closing { domain: "file follower" });
138                }
139                let step = match &mut self.source {
140                    Source::Stored(reader) => match reader.next_raw(true).await? {
141                        Some(batch) => Step::Data(batch),
142                        None => Step::Done,
143                    },
144                    Source::Local(local) => local.next(&self.inner).await?,
145                };
146                match step {
147                    Step::Data(batch) => {
148                        self.delivered += batch.bytes.len() as u64;
149                        return Ok(Some(FileReadEvent::new(FileEvent::Data(batch))));
150                    }
151                    Step::Signal(event) => {
152                        return Ok(Some(FileReadEvent::new(event)));
153                    }
154                    Step::Done => {
155                        self.finished = true;
156                        return Ok(None);
157                    }
158                    Step::Wait => {
159                        let mut changed = self.inner.changed.subscribe();
160                        tokio::select! {
161                            _ = tokio::time::sleep(self.inner.settings.filesystem.poll_interval) => {},
162                            _ = changed.changed() => {},
163                        }
164                    }
165                }
166            }
167        };
168        context.result(span, future).await
169    }
170
171    pub fn completion(&self) -> Result<FileCompletion> {
172        if !self.finished {
173            return Err(Error::NotReady { domain: "file follow completion" });
174        }
175        Ok(FileCompletion { bytes: self.delivered })
176    }
177}
178
179impl Local {
180    async fn next(&mut self, inner: &Arc<Inner>) -> Result<Step> {
181        if self.reader.is_none() {
182            self.reader = Some(LocalReader::open(inner, self.path.clone(), ByteRange { offset: 0, limit: None }).await?);
183        }
184        let reader = self.reader.as_ref().expect("local follower retains its current file");
185        let Some((same, length)) = reader.inspect().await? else {
186            if self.missing {
187                return Ok(Step::Wait);
188            }
189            self.missing = true;
190            return Ok(Step::Signal(FileEvent::Replaced));
191        };
192        if !same {
193            self.reader.take();
194            self.reader = Some(LocalReader::open(inner, self.path.clone(), ByteRange { offset: 0, limit: None }).await?);
195            let signaled = std::mem::replace(&mut self.missing, false);
196            return Ok(if signaled { Step::Wait } else { Step::Signal(FileEvent::Replaced) });
197        }
198        let reader = self.reader.as_mut().expect("local follower retains its current file");
199        self.missing = false;
200        if length < reader.offset() {
201            self.reader.take();
202            self.reader = Some(LocalReader::open(inner, self.path.clone(), ByteRange { offset: 0, limit: None }).await?);
203            return Ok(Step::Signal(FileEvent::Truncated { length }));
204        }
205        reader.retarget(length);
206        match reader.next(true).await? {
207            Some(batch) => Ok(Step::Data(batch)),
208            None => Ok(Step::Wait),
209        }
210    }
211}
212
213/* === Public Functions === */
214
215/* === Private Functions === */
216
217fn start(start: FollowStart, length: u64) -> u64 {
218    match start {
219        FollowStart::Beginning => 0,
220        FollowStart::End => length,
221        FollowStart::At(offset) => offset,
222    }
223}
224
225/* === Tests === */
harmos_storage/file_follow.rs

harmos_storage/
file_follow.rs
