ferroui_browser/storage/
writeable_stream.rs

1use crate::interop::completion_helper::{PromiseError, PromiseFuture};
2use crate::interop::{storage_helper, stream_helper, JsObject};
3use ferroui_base::logging::{LogArea, LogEventLevel, Logger};
4use ferroui_base::threading::Dispatcher;
5use std::cell::RefCell;
6use std::future::Future;
7use std::io::{self, Seek, SeekFrom, Write};
8use std::rc::Rc;
9use std::task::{Poll, Waker};
10
11/// Loose wrapper implementation of a stream on top of the File System
12/// Access API's `FileSystemWritableFileStream`.
13///
14/// The page writes only asynchronously. [`write_async`](Self::write_async),
15/// [`flush_async`](Self::flush_async) and [`close_async`](Self::close_async)
16/// are upstream's `WriteAsync`, `FlushAsync` and `DisposeAsync`. The
17/// synchronous [`Write`] the storage contracts hand out issues the same
18/// writes without waiting for them (the browser cannot block): the script
19/// stream queues them in order, and `write` and `flush` report a failure
20/// that has already settled. Dropping the stream closes it, which commits
21/// the file once the queued writes are done. Until that close settles, the
22/// storage items wait before they open the same file or read its
23/// properties, and a failed write or close is logged and reported by the
24/// next opening of that file (see [`settle_pending_closes`]).
25pub struct WriteableStream {
26    js_reference: Option<JsObject>,
27    // The storage item of the file, to which a failed close is reported.
28    item: Option<JsObject>,
29    // Unfortunately we can't read current length/position, so we need to keep it on this side
30    // only.
31    length: u64,
32    position: u64,
33    requests: Rc<RefCell<Requests>>,
34}
35
36/// The requests of a stream nobody waits for, and the first failure among
37/// them.
38#[derive(Default)]
39struct Requests {
40    pending: usize,
41    error: Option<PromiseError>,
42    wakers: Vec<Waker>,
43}
44
45fn disposed() -> io::Error {
46    io::Error::other("Cannot access a disposed object. Object name: 'WriteableStream'.")
47}
48
49impl WriteableStream {
50    /// The stream `js_reference` over the file of the storage item `item`,
51    /// which is `initial_length` bytes long when the stream opens.
52    pub(crate) fn new(js_reference: JsObject, item: Option<JsObject>, initial_length: u64) -> Self {
53        Self { js_reference: Some(js_reference), item, length: initial_length, position: 0, requests: Rc::default() }
54    }
55
56    fn js_reference(&self) -> io::Result<&JsObject> {
57        self.js_reference.as_ref().ok_or_else(disposed)
58    }
59
60    pub fn can_read(&self) -> bool {
61        false
62    }
63
64    pub fn can_seek(&self) -> bool {
65        true
66    }
67
68    pub fn can_write(&self) -> bool {
69        true
70    }
71
72    pub fn length(&self) -> u64 {
73        self.length
74    }
75
76    pub fn position(&self) -> u64 {
77        self.position
78    }
79
80    /// Moves the position to `position`.
81    pub fn set_position(&mut self, position: u64) -> io::Result<()> {
82        self.seek(SeekFrom::Start(position)).map(|_| ())
83    }
84
85    /// Resizes the file to `value` bytes.
86    pub fn set_length(&mut self, value: u64) -> io::Result<()> {
87        let js_reference = self.js_reference()?.clone();
88        self.length = value;
89
90        // See https://docs.w3cub.com/dom/filesystemwritablefilestream/truncate
91        // If the offset is smaller than the size, it remains unchanged. If the offset is larger
92        // than size, the offset is set to that size.
93        if self.position > self.length {
94            self.position = self.length;
95        }
96
97        self.observe(stream_helper::truncate(&js_reference, value));
98        Ok(())
99    }
100
101    /// Writes `buffer` at the position; resolves once the page has written
102    /// it.
103    pub async fn write_async(&mut self, buffer: &[u8]) -> io::Result<()> {
104        let task = self.write_internal(buffer)?;
105        task.await.map(|_| ()).map_err(io::Error::from)
106    }
107
108    fn write_internal(&mut self, buffer: &[u8]) -> io::Result<PromiseFuture<JsObject>> {
109        let js_reference = self.js_reference()?;
110        let task = stream_helper::write_async(js_reference, buffer);
111        self.position += buffer.len() as u64;
112        self.length = self.length.max(self.position);
113        Ok(task)
114    }
115
116    /// Resolves once every write, seek and truncation issued through the
117    /// synchronous members has settled, with the first failure among them.
118    pub async fn flush_async(&mut self) -> io::Result<()> {
119        requests_settled(self.requests.clone()).await;
120        self.take_error()
121    }
122
123    /// Closes the stream, which commits the written content to the file.
124    pub async fn close_async(mut self) -> io::Result<()> {
125        let Some(js_reference) = self.js_reference.take() else {
126            return Ok(());
127        };
128        let close = stream_helper::close_async(&js_reference).await;
129        requests_settled(self.requests.clone()).await;
130        self.take_error()?;
131        close.map(|_| ()).map_err(io::Error::from)
132    }
133
134    /// Counts a request nobody waits for and records its failure, for the
135    /// next call to report.
136    fn observe(&self, task: PromiseFuture<JsObject>) {
137        self.requests.borrow_mut().pending += 1;
138        let requests = self.requests.clone();
139        Dispatcher::ui_thread().to_task_scheduler().start_local(async move {
140            let result = task.await;
141            let wakers = {
142                let mut requests = requests.borrow_mut();
143                if let Err(failure) = result {
144                    requests.error.get_or_insert(failure);
145                }
146                requests.pending -= 1;
147                if requests.pending == 0 {
148                    std::mem::take(&mut requests.wakers)
149                } else {
150                    Vec::new()
151                }
152            };
153            wakers.into_iter().for_each(Waker::wake);
154        });
155    }
156
157    fn take_error(&self) -> io::Result<()> {
158        match self.requests.borrow_mut().error.take() {
159            Some(error) => Err(error.into()),
160            None => Ok(()),
161        }
162    }
163}
164
165/// Resolves once no request of `requests` is pending.
166fn requests_settled(requests: Rc<RefCell<Requests>>) -> impl Future<Output = ()> {
167    std::future::poll_fn(move |cx| {
168        let mut requests = requests.borrow_mut();
169        if requests.pending == 0 {
170            Poll::Ready(())
171        } else {
172            requests.wakers.push(cx.waker().clone());
173            Poll::Pending
174        }
175    })
176}
177
178impl Write for WriteableStream {
179    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
180        self.take_error()?;
181        let task = self.write_internal(buffer)?;
182        self.observe(task);
183        Ok(buffer.len())
184    }
185
186    fn flush(&mut self) -> io::Result<()> {
187        // The page queues every write and cannot be waited for here: reports a failure that has
188        // settled. `flush_async` waits.
189        self.take_error()
190    }
191}
192
193impl Seek for WriteableStream {
194    fn seek(&mut self, position: SeekFrom) -> io::Result<u64> {
195        let js_reference = self.js_reference()?.clone();
196        let position = seek_target(position, self.position, self.length)?;
197        self.observe(stream_helper::seek(&js_reference, position));
198        // Upstream returns the new position without storing it.
199        self.position = position;
200        Ok(position)
201    }
202}
203
204fn seek_target(position: SeekFrom, current: u64, length: u64) -> io::Result<u64> {
205    let position = match position {
206        SeekFrom::Current(offset) => current as i64 + offset,
207        SeekFrom::End(offset) => length as i64 + offset,
208        SeekFrom::Start(offset) => offset as i64,
209    };
210    u64::try_from(position).map_err(|_| {
211        io::Error::new(
212            io::ErrorKind::InvalidInput,
213            "An attempt was made to move the position before the beginning of the stream.",
214        )
215    })
216}
217
218impl Drop for WriteableStream {
219    fn drop(&mut self) {
220        let Some(js_reference) = self.js_reference.take() else {
221            return;
222        };
223        // Upstream's `Dispose`: close without waiting. The close stays registered for its file
224        // until it settles, and with its failure until an opening of the file reports it.
225        let close = stream_helper::close_async(&js_reference);
226        let pending = Rc::new(PendingClose { item: self.item.take(), state: RefCell::default() });
227        PENDING_CLOSES.with(|closes| closes.borrow_mut().push(pending.clone()));
228        let requests = self.requests.clone();
229        Dispatcher::ui_thread().to_task_scheduler().start_local(async move {
230            let close = close.await;
231            requests_settled(requests.clone()).await;
232            let error = requests.borrow_mut().error.take().or(close.err());
233            if let Some(error) = &error {
234                if let Some(logger) = Logger::try_get(LogEventLevel::Error, LogArea::BROWSER_PLATFORM) {
235                    logger.log(None, &format!("Writing a file failed: {error}"));
236                }
237            }
238            pending.settle(error);
239        });
240    }
241}
242
243/// The close of a dropped stream: settled once the page has committed the
244/// file or failed to.
245#[derive(Default)]
246struct CloseState {
247    settled: bool,
248    error: Option<PromiseError>,
249    wakers: Vec<Waker>,
250}
251
252struct PendingClose {
253    item: Option<JsObject>,
254    state: RefCell<CloseState>,
255}
256
257impl PendingClose {
258    fn settle(self: &Rc<Self>, error: Option<PromiseError>) {
259        let wakers = {
260            let mut state = self.state.borrow_mut();
261            state.settled = true;
262            state.error = error;
263            std::mem::take(&mut state.wakers)
264        };
265        // A close that succeeded has nothing left to report.
266        if self.state.borrow().error.is_none() {
267            PENDING_CLOSES.with(|closes| closes.borrow_mut().retain(|close| !Rc::ptr_eq(close, self)));
268        }
269        wakers.into_iter().for_each(Waker::wake);
270    }
271
272    fn settled(self: Rc<Self>) -> impl Future<Output = ()> {
273        std::future::poll_fn(move |cx| {
274            let mut state = self.state.borrow_mut();
275            if state.settled {
276                Poll::Ready(())
277            } else {
278                state.wakers.push(cx.waker().clone());
279                Poll::Pending
280            }
281        })
282    }
283
284    /// Takes the failure of the settled close, which is then reported.
285    fn take_error(self: &Rc<Self>) -> Option<PromiseError> {
286        let error = self.state.borrow_mut().error.take();
287        PENDING_CLOSES.with(|closes| closes.borrow_mut().retain(|close| !Rc::ptr_eq(close, self)));
288        error
289    }
290}
291
292ferroui_base::ferro_thread_local! {
293    static PENDING_CLOSES: RefCell<Vec<Rc<PendingClose>>> = const { RefCell::new(Vec::new()) };
294}
295
296/// Waits for the closes of the dropped streams over the file of the storage
297/// item `item` (the same entry, whichever item opened it). A file written
298/// through the synchronous [`Write`] has its content committed then.
299///
300/// With `report`, fails with the first failed write or close among them,
301/// which is then reported; without, failures are left for the next
302/// opening.
303pub(crate) async fn settle_pending_closes(item: &JsObject, report: bool) -> io::Result<()> {
304    let closes = PENDING_CLOSES.with(|closes| closes.borrow().clone());
305    let mut first_error = None;
306    for close in closes {
307        let Some(close_item) = &close.item else { continue };
308        if !is_same_entry(item, close_item).await {
309            continue;
310        }
311        close.clone().settled().await;
312        if report {
313            if let Some(error) = close.take_error() {
314                first_error.get_or_insert(error);
315            }
316        }
317    }
318    match first_error {
319        Some(error) => Err(io::Error::other(format!("Writing the file failed: {error}"))),
320        None => Ok(()),
321    }
322}
323
324async fn is_same_entry(item: &JsObject, other: &JsObject) -> bool {
325    if item == other {
326        return true;
327    }
328    storage_helper::is_same_entry(item, other).await.ok().and_then(|same| same.as_bool()).unwrap_or(false)
329}
330
331#[cfg(test)]
332mod tests {
333    // Not from upstream (upstream has no tests of the browser streams): the parts that need no
334    // page.
335    use super::*;
336    use std::task::Context;
337
338    #[test]
339    fn a_close_settles_for_its_waiters_and_keeps_only_its_failure() {
340        use std::sync::atomic::{AtomicUsize, Ordering};
341        use std::sync::Arc;
342        use std::task::Wake;
343
344        struct CountingWaker(AtomicUsize);
345        impl Wake for CountingWaker {
346            fn wake(self: Arc<Self>) {
347                self.0.fetch_add(1, Ordering::SeqCst);
348            }
349        }
350        let counter = Arc::new(CountingWaker(AtomicUsize::new(0)));
351        let waker = Waker::from(counter.clone());
352        let mut context = Context::from_waker(&waker);
353
354        let succeeded = Rc::new(PendingClose { item: None, state: RefCell::default() });
355        let failed = Rc::new(PendingClose { item: None, state: RefCell::default() });
356        PENDING_CLOSES.with(|closes| closes.borrow_mut().extend([succeeded.clone(), failed.clone()]));
357
358        let mut waiting = Box::pin(succeeded.clone().settled());
359        assert_eq!(waiting.as_mut().poll(&mut context), Poll::Pending);
360        succeeded.settle(None);
361        assert_eq!(counter.0.load(Ordering::SeqCst), 1);
362        assert_eq!(waiting.as_mut().poll(&mut context), Poll::Ready(()));
363
364        failed.settle(Some(PromiseError::new("", "The disk is full")));
365        assert_eq!(PENDING_CLOSES.with(|closes| closes.borrow().len()), 1, "only the failed close is kept");
366        assert_eq!(failed.take_error(), Some(PromiseError::new("", "The disk is full")));
367        assert_eq!(PENDING_CLOSES.with(|closes| closes.borrow().len()), 0, "a reported failure is dropped");
368    }
369
370    #[test]
371    fn requests_settled_waits_for_the_pending_requests() {
372        let requests = Rc::new(RefCell::new(Requests { pending: 1, ..Default::default() }));
373        let mut settled = Box::pin(requests_settled(requests.clone()));
374        let waker = Waker::noop();
375        let mut context = Context::from_waker(waker);
376        assert_eq!(settled.as_mut().poll(&mut context), Poll::Pending);
377        requests.borrow_mut().pending = 0;
378        assert_eq!(settled.as_mut().poll(&mut context), Poll::Ready(()));
379    }
380
381    #[test]
382    fn seek_target_is_relative_to_each_origin() {
383        assert_eq!(seek_target(SeekFrom::Start(3), 7, 10).unwrap(), 3);
384        assert_eq!(seek_target(SeekFrom::Current(2), 7, 10).unwrap(), 9);
385        assert_eq!(seek_target(SeekFrom::End(-4), 7, 10).unwrap(), 6);
386        assert_eq!(seek_target(SeekFrom::End(5), 7, 10).unwrap(), 15);
387        assert!(seek_target(SeekFrom::Current(-8), 7, 10).is_err());
388    }
389}
ferroui_browser/storage/writeable_stream.rs

ferroui_browser/storage/
writeable_stream.rs
