IndexedDB: communicate transaction errors and async response data more precisely (#38027)

Digging into several crashing tests revealed that committing
transactions is a fallible operation. Propagating those errors led to
exposing many new errors caused by the IDBRequest implementation
assuming that all successful responses contained a structured clone. The
end result is a bunch of new test failures that were previously hidden.

Testing: Existing test coverage is sufficient.

---------

Signed-off-by: Josh Matthews <josh@joshmatthews.net>

components/net/indexeddb/engines/heed.rs

@@ -9,7 +9,7 @@ use heed::types::*;
 use heed::{Database, Env, EnvOpenOptions};
 use log::warn;
 use net_traits::indexeddb_thread::{
-    AsyncOperation, AsyncReadOnlyOperation, AsyncReadWriteOperation, IndexedDBTxnMode,
+    AsyncOperation, AsyncReadOnlyOperation, AsyncReadWriteOperation, IdbResult, IndexedDBTxnMode,
 };
 use tokio::sync::oneshot;
 
@@ -72,7 +72,7 @@ impl KvsEngine for HeedEngine {
             .heed_env
             .create_database(&mut write_txn, Some(&*store_name.to_string()))?;
 
-        write_txn.commit().expect("Failed to commit transaction");
+        write_txn.commit()?;
 
         let key_generator = { if auto_increment { Some(0) } else { None } };
 
@@ -95,7 +95,7 @@ impl KvsEngine for HeedEngine {
             .heed_env
             .create_database(&mut write_txn, Some(&*store_name.to_string()))?;
         store.clear(&mut write_txn)?;
-        write_txn.commit().expect("Failed to commit transaction");
+        write_txn.commit()?;
 
         let mut open_stores = self.open_stores.write().unwrap();
         open_stores.retain(|key, _| key != &store_name);
@@ -130,6 +130,7 @@ impl KvsEngine for HeedEngine {
             self.read_pool.spawn(move || {
                 let env = heed_env;
                 let rtxn = env.read_txn().expect("Could not create idb store reader");
+                let mut results = vec![];
                 for request in transaction.requests {
                     match request.operation {
                         AsyncOperation::ReadOnly(AsyncReadOnlyOperation::GetItem(key)) => {
@@ -143,9 +144,10 @@ impl KvsEngine for HeedEngine {
                             let result = store.inner.get(&rtxn, &key).expect("Could not get item");
 
                             if let Some(blob) = result {
-                                let _ = request.sender.send(Some(blob.to_vec()));
+                                results
+                                    .push((request.sender, Some(IdbResult::Data(blob.to_vec()))));
                             } else {
-                                let _ = request.sender.send(None);
+                                results.push((request.sender, None));
                             }
                         },
                         AsyncOperation::ReadOnly(AsyncReadOnlyOperation::Count(key)) => {
@@ -172,13 +174,23 @@ impl KvsEngine for HeedEngine {
                     warn!("IDBTransaction's execution channel is dropped");
                 };
 
-                rtxn.commit().expect("Failed to commit transaction");
+                if let Err(e) = rtxn.commit() {
+                    warn!("Error committing transaction: {:?}", e);
+                    for (sender, _) in results {
+                        let _ = sender.send(Err(()));
+                    }
+                } else {
+                    for (sender, result) in results {
+                        let _ = sender.send(Ok(result));
+                    }
+                }
             });
         } else {
             self.write_pool.spawn(move || {
                 // Acquiring a writer will block the thread if another `readwrite` transaction is active
                 let env = heed_env;
                 let mut wtxn = env.write_txn().expect("Could not create idb store writer");
+                let mut results = vec![];
                 for request in transaction.requests {
                     match request.operation {
                         AsyncOperation::ReadWrite(AsyncReadWriteOperation::PutItem(
@@ -186,28 +198,28 @@ impl KvsEngine for HeedEngine {
                             value,
                             overwrite,
                         )) => {
-                            let key: Vec<u8> = bincode::serialize(&key).unwrap();
+                            let serialized_key: Vec<u8> = bincode::serialize(&key).unwrap();
                             let stores = stores
                                 .write()
                                 .expect("Could not acquire write lock on stores");
                             let store = stores
                                 .get(&request.store_name)
                                 .expect("Could not get store");
-                            if overwrite {
-                                let result =
-                                    store.inner.put(&mut wtxn, &key, &value).ok().and(Some(key));
-                                request.sender.send(result).unwrap();
-                            } else if store
-                                .inner
-                                .get(&wtxn, &key)
-                                .expect("Could not get item")
-                                .is_none()
+                            if overwrite ||
+                                store
+                                    .inner
+                                    .get(&wtxn, &serialized_key)
+                                    .expect("Could not get item")
+                                    .is_none()
                             {
-                                let result =
-                                    store.inner.put(&mut wtxn, &key, &value).ok().and(Some(key));
-                                let _ = request.sender.send(result);
+                                let result = store
+                                    .inner
+                                    .put(&mut wtxn, &serialized_key, &value)
+                                    .ok()
+                                    .and(Some(IdbResult::Key(key)));
+                                results.push((request.sender, result));
                             } else {
-                                let _ = request.sender.send(None);
+                                results.push((request.sender, None));
                             }
                         },
                         AsyncOperation::ReadOnly(AsyncReadOnlyOperation::GetItem(key)) => {
@@ -220,22 +232,25 @@ impl KvsEngine for HeedEngine {
                                 .expect("Could not get store");
                             let result = store.inner.get(&wtxn, &key).expect("Could not get item");
 
-                            if let Some(blob) = result {
-                                let _ = request.sender.send(Some(blob.to_vec()));
-                            } else {
-                                let _ = request.sender.send(None);
-                            }
+                            results.push((
+                                request.sender,
+                                result.map(|blob| IdbResult::Data(blob.to_vec())),
+                            ));
                         },
                         AsyncOperation::ReadWrite(AsyncReadWriteOperation::RemoveItem(key)) => {
-                            let key: Vec<u8> = bincode::serialize(&key).unwrap();
+                            let serialized_key: Vec<u8> = bincode::serialize(&key).unwrap();
                             let stores = stores
                                 .write()
                                 .expect("Could not acquire write lock on stores");
                             let store = stores
                                 .get(&request.store_name)
                                 .expect("Could not get store");
-                            let result = store.inner.delete(&mut wtxn, &key).ok().and(Some(key));
-                            let _ = request.sender.send(result);
+                            let result = store
+                                .inner
+                                .delete(&mut wtxn, &serialized_key)
+                                .ok()
+                                .and(Some(IdbResult::Key(key)));
+                            results.push((request.sender, result));
                         },
                         AsyncOperation::ReadOnly(AsyncReadOnlyOperation::Count(key)) => {
                             let _key: Vec<u8> = bincode::serialize(&key).unwrap();
@@ -250,7 +265,16 @@ impl KvsEngine for HeedEngine {
                     }
                 }
 
-                wtxn.commit().expect("Failed to commit to database");
+                if let Err(e) = wtxn.commit() {
+                    warn!("Error committing to database: {:?}", e);
+                    for (sender, _) in results {
+                        let _ = sender.send(Err(()));
+                    }
+                } else {
+                    for (sender, result) in results {
+                        let _ = sender.send(Ok(result));
+                    }
+                }
             })
         }
         rx

components/net/indexeddb/engines/mod.rs

@@ -5,7 +5,7 @@
 use std::collections::VecDeque;
 
 use ipc_channel::ipc::IpcSender;
-use net_traits::indexeddb_thread::{AsyncOperation, IndexedDBTxnMode};
+use net_traits::indexeddb_thread::{AsyncOperation, IdbResult, IndexedDBTxnMode};
 use tokio::sync::oneshot;
 
 pub use self::heed::HeedEngine;
@@ -46,7 +46,7 @@ impl std::fmt::Display for SanitizedName {
 }
 
 pub struct KvsOperation {
-    pub sender: IpcSender<Option<Vec<u8>>>,
+    pub sender: IpcSender<Result<Option<IdbResult>, ()>>,
     pub store_name: SanitizedName,
     pub operation: AsyncOperation,
 }

components/net/indexeddb/idb_thread.rs

@@ -11,7 +11,7 @@ use std::thread;
 use ipc_channel::ipc::{self, IpcError, IpcReceiver, IpcSender};
 use log::{debug, warn};
 use net_traits::indexeddb_thread::{
-    AsyncOperation, IndexedDBThreadMsg, IndexedDBTxnMode, SyncOperation,
+    AsyncOperation, IdbResult, IndexedDBThreadMsg, IndexedDBTxnMode, SyncOperation,
 };
 use servo_config::pref;
 use servo_url::origin::ImmutableOrigin;
@@ -88,7 +88,7 @@ impl<E: KvsEngine> IndexedDBEnvironment<E> {
 
     fn queue_operation(
         &mut self,
-        sender: IpcSender<Option<Vec<u8>>>,
+        sender: IpcSender<Result<Option<IdbResult>, ()>>,
         store_name: SanitizedName,
         serial_number: u64,
         mode: IndexedDBTxnMode,