fix: Fixed a race condition in cursors

Closes #52

Author: seanaye

src/cursor/base_cursor.rs

@@ -1,5 +1,6 @@
 use super::{CursorDirection, CursorSys};
-use crate::future::{PollUnpinned, VoidRequest};
+use crate::future::request::listeners::EventTargetResult;
+use crate::future::{PollUnpinned, Request};
 use crate::internal_utils::SystemRepr;
 use crate::primitive::{TryFromJs, TryToJs};
 use fancy_constructor::new;
@@ -27,7 +28,7 @@ pub(crate) enum CursorState {
     TryNext,
 
     /// We are currently reading the next record.
-    ReadingNext(VoidRequest),
+    ReadingNext(Request<EventTargetResult>),
 }
 
 impl BaseCursor {
@@ -68,7 +69,7 @@ impl BaseCursor {
             Ok(())
         } else {
             self.as_sys().advance(step)?;
-            self.req().await
+            self.req().await.map(|_| ())
         }
     }
 
@@ -108,8 +109,8 @@ impl BaseCursor {
         self.as_sys().direction()
     }
 
-    pub(crate) fn req(&self) -> VoidRequest {
-        VoidRequest::new(self.as_sys().req())
+    pub(crate) fn req(&self) -> Request<EventTargetResult> {
+        Request::new(self.as_sys().req())
     }
 
     pub(crate) fn poll_state<R, F>(
@@ -186,7 +187,7 @@ impl BaseCursor {
 
     fn on_req_polled<R, F>(
         &mut self,
-        poll: Poll<crate::Result<()>>,
+        poll: Poll<crate::Result<EventTargetResult>>,
         read_current: F,
     ) -> Poll<crate::Result<Option<R>>>
     where
@@ -196,15 +197,27 @@ impl BaseCursor {
             Poll::Ready(res) => {
                 self.state = CursorState::ReadCurrent;
 
-                Poll::Ready(match res {
-                    Ok(()) => {
+                let should_continue = res.map(|event_result| match event_result {
+                    EventTargetResult::Null => false,
+                    EventTargetResult::Cursor(cursor_sys) => {
+                        self.sys = cursor_sys;
+                        true
+                    }
+                    EventTargetResult::NotNull => true,
+                });
+
+                Poll::Ready(match should_continue {
+                    Ok(true) => {
                         // Firefox implementation: key gets set to undefined on cursor end
                         if self.has_key() {
                             self.read_current(read_current)
                         } else {
                             Ok(None)
                         }
                     }
+                    // Chrome implementation: the only way to know if a cursor has finished
+                    // is by reading the value from onsuccess event.target.result
+                    Ok(false) => Ok(None),
                     Err(e) => Err(e),
                 })
             }

src/future/mod.rs

@@ -13,7 +13,7 @@ mod basic;
 mod get_all;
 mod maybe_errored;
 mod open_db;
-mod request;
+pub(crate) mod request;
 mod traits;
 
 iffeat! {

src/future/request.rs

@@ -1,4 +1,4 @@
-mod listeners;
+pub(crate) mod listeners;
 mod untyped;
 
 use crate::internal_utils::SystemRepr;
@@ -54,19 +54,30 @@ impl PollUnpinned for Request {
 
     fn poll_unpinned(&mut self, cx: &mut Context) -> Poll<Self::Output> {
         match self.inner.poll_unpinned(cx) {
-            Poll::Ready(Ok(())) => Poll::Ready(self.as_sys().result().map_err(Into::into)),
+            Poll::Ready(Ok(_)) => Poll::Ready(self.as_sys().result().map_err(Into::into)),
             Poll::Pending => Poll::Pending,
             Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
         }
     }
 }
 
+#[sealed]
+#[cfg(feature = "cursors")]
+impl PollUnpinned for Request<listeners::EventTargetResult> {
+    type Output = crate::Result<listeners::EventTargetResult>;
+
+    #[inline]
+    fn poll_unpinned(&mut self, cx: &mut Context) -> Poll<Self::Output> {
+        self.inner.poll_unpinned(cx)
+    }
+}
+
 #[sealed]
 impl PollUnpinned for Request<()> {
     type Output = crate::Result<()>;
 
     #[inline]
     fn poll_unpinned(&mut self, cx: &mut Context) -> Poll<Self::Output> {
-        self.inner.poll_unpinned(cx)
+        self.inner.poll_unpinned(cx).map(|res| res.map(|_| ()))
     }
 }

src/future/request/listeners.rs

@@ -1,23 +1,58 @@
 use super::super::traits::*;
 use crate::error::UnexpectedDataError;
+use cfg_if::cfg_if;
 use std::task::{Context, Poll};
 use tokio::sync::mpsc;
 use wasm_bindgen::prelude::*;
 
-type Callback = Closure<dyn FnMut() + 'static>;
+type Callback = Closure<dyn FnMut(web_sys::Event) + 'static>;
+
+/// represents the value on an event.target.result
+pub(crate) enum EventTargetResult {
+    /// the event.target.result was null
+    Null,
+    /// the event.target.result was a [`web_sys::IdbCursor`] instance
+    #[cfg(feature = "cursors")]
+    Cursor(crate::cursor::CursorSys),
+    /// the event.target.result was not null
+    NotNull,
+}
 
 pub(super) struct Listeners {
-    rx: mpsc::Receiver<()>,
+    rx: mpsc::Receiver<EventTargetResult>,
     req: web_sys::IdbRequest,
-    _callback: Closure<dyn FnMut() + 'static>,
+    _callback: Callback,
 }
 
 impl Listeners {
     pub(super) fn new(req: web_sys::IdbRequest) -> Self {
         let (tx, rx) = mpsc::channel(1);
 
-        let callback = Callback::wrap(Box::new(move || {
-            let _ = tx.try_send(());
+        let callback = Callback::wrap(Box::new(move |e: web_sys::Event| {
+            let non_null_result = e
+                .target()
+                .map(JsValue::from)
+                // get the event.target.result
+                .and_then(|val| js_sys::Reflect::get(&val, &JsValue::from("result")).ok())
+                // make sure its not null or undefined
+                .filter(|val| !val.is_undefined() && !val.is_null());
+
+            let _ = tx.try_send(match non_null_result {
+                None => EventTargetResult::Null,
+                #[cfg_attr(not(feature = "cursors"), expect(unused_variables))]
+                Some(val) => {
+                    cfg_if! {
+                        if #[cfg(feature = "cursors")] {
+                             match val.dyn_into::<crate::cursor::CursorSys>() {
+                                Ok(cursor) => EventTargetResult::Cursor(cursor),
+                                Err(_) => EventTargetResult::NotNull,
+                            }
+                        } else {
+                            EventTargetResult::NotNull
+                        }
+                    }
+                }
+            });
         }));
 
         let as_fn = callback.as_ref().unchecked_ref();
@@ -41,12 +76,15 @@ impl Drop for Listeners {
 
 #[::sealed::sealed]
 impl PollUnpinned for Listeners {
-    type Output = crate::Result<()>;
+    type Output = crate::Result<EventTargetResult>;
 
     fn poll_unpinned(&mut self, cx: &mut Context) -> Poll<Self::Output> {
         match self.rx.poll_recv(cx) {
             Poll::Pending => Poll::Pending,
-            Poll::Ready(Some(())) => Poll::Ready(super::UntypedRequest::req_to_result(&self.req)),
+            Poll::Ready(Some(event_target)) => Poll::Ready(super::UntypedRequest::req_to_result(
+                &self.req,
+                event_target,
+            )),
             Poll::Ready(None) => Poll::Ready(Err(UnexpectedDataError::ChannelDropped.into())),
         }
     }

src/future/request/untyped.rs

@@ -1,5 +1,5 @@
-use super::super::traits::*;
 use super::Listeners;
+use super::{super::traits::*, listeners::EventTargetResult};
 use sealed::sealed;
 use std::task::{Context, Poll};
 use wasm_bindgen::prelude::*;
@@ -31,17 +31,17 @@ impl crate::internal_utils::SystemRepr for UntypedRequest {
 }
 
 impl UntypedRequest {
-    pub(super) fn req_to_result(req: &web_sys::IdbRequest) -> crate::Result<()> {
+    pub(super) fn req_to_result<T>(req: &web_sys::IdbRequest, v: T) -> crate::Result<T> {
         match req.error() {
-            Ok(None) => Ok(()),
+            Ok(None) => Ok(v),
             Ok(Some(e)) => Err(e.into()),
             Err(e) => Err(e.into()),
         }
     }
 
-    fn poll_request(req: &web_sys::IdbRequest) -> Poll<crate::Result<()>> {
+    fn poll_request<T>(req: &web_sys::IdbRequest, v: T) -> Poll<crate::Result<T>> {
         if matches!(req.ready_state(), IdbRequestReadyState::Done) {
-            Poll::Ready(Self::req_to_result(req))
+            Poll::Ready(Self::req_to_result(req, v))
         } else {
             Poll::Pending
         }
@@ -56,13 +56,13 @@ impl UntypedRequest {
 
 #[sealed]
 impl PollUnpinned for UntypedRequest {
-    type Output = crate::Result<()>;
+    type Output = crate::Result<EventTargetResult>;
 
     fn poll_unpinned(&mut self, cx: &mut Context) -> Poll<Self::Output> {
         match self {
             Self::WithListeners(listeners) => listeners.poll_unpinned(cx),
             Self::Bare(req) => {
-                if let Poll::Ready(res) = Self::poll_request(req) {
+                if let Poll::Ready(res) = Self::poll_request(req, EventTargetResult::NotNull) {
                     Poll::Ready(res)
                 } else {
                     let mut listeners = Self::create_listeners(req);

tests/tests/object_store/add_put.rs

@@ -103,7 +103,7 @@ macro_rules! common_tests {
                 .unwrap()
                 .await;
 
-            assert_eq!(res, Ok(()));
+            assert!(res.is_ok());
         }
     };
 }

tests/tests/object_store/query_source/cursor.rs

@@ -378,6 +378,44 @@ pub mod cursor {
             assert_eq!(data, expected);
         }
 
+        #[wasm_bindgen_test]
+        #[cfg(all(feature = "serde", feature = "streams", feature = "cursors"))]
+        pub async fn cursor_stream_one() {
+            use serde::{Deserialize, Serialize};
+            let db = random_db_keyval().await;
+            #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+            struct Foo {
+                a: String,
+                b: u32,
+                key: u32,
+            }
+            let data_1 = Foo {
+                a: "hello".to_string(),
+                b: 42,
+                key: 1,
+            };
+            let data_2 = Foo {
+                a: "world".to_string(),
+                b: 42,
+                key: 2,
+            };
+            // seed db
+            {
+                open_tx!(db, Readwrite > (tx, store));
+
+                store.add(data_1.clone()).serde().unwrap().await.unwrap();
+                store.add(data_2.clone()).serde().unwrap().await.unwrap();
+                drop(store);
+                tx.commit().await.unwrap();
+            }
+            open_tx!(db, Readonly > (tx, store));
+            let cursor = store.open_cursor().serde().unwrap().await.unwrap().unwrap();
+            let res: Vec<Foo> = cursor.stream_ser::<Foo>().try_collect().await.unwrap();
+            assert_eq!(res.len(), 2);
+            assert_eq!(res.first().unwrap(), &data_1);
+            assert_eq!(res.get(1).unwrap(), &data_2);
+        }
+
         #[wasm_bindgen_test]
         pub async fn with_query() {
             let db = random_db_keyval().await;