fix(l1): close background threads before dropping db (#5749)

**Motivation**

The error `pthread lock: Invalid argument` has been happening rarely
with rocksdb, but #5497 manages to trigger it consistently so it should
be fixed soon.

This PR is inspired by @PivasDesant's solution (see #5661)

**Description**

Stops background DB threads before dropping the inner database object.

The store background threads do not need to be stopped since these hold
a reference to the db (preventing it from being dropped early).

#5658 contains a script to test the fix by running the rocksdb test a
thousand times

<!-- A clear and concise general description of the changes this PR
introduces -->

<!-- Link to issues: Resolves #111, Resolves #222 -->

**Checklist**

- [X] Updated `STORE_SCHEMA_VERSION` (crates/storage/lib.rs) if the PR
includes breaking changes to the `Store` requiring a re-sync.

Closes #5658

---------

Co-authored-by: Edgar <[email protected]>

Author: iovoid

crates/storage/backend/rocksdb.rs

@@ -203,6 +203,16 @@ impl RocksDBBackend {
     }
 }
 
+impl Drop for RocksDBBackend {
+    fn drop(&mut self) {
+        // When the last reference to the db is dropped, stop background threads
+        // See https://github.com/facebook/rocksdb/issues/11349
+        if let Some(db) = Arc::get_mut(&mut self.db) {
+            db.cancel_all_background_work(true);
+        }
+    }
+}
+
 impl StorageBackend for RocksDBBackend {
     fn clear_table(&self, table: &'static str) -> Result<(), StoreError> {
         let cf = self

crates/storage/store.rs

@@ -50,6 +50,7 @@ use std::{
         atomic::AtomicU64,
         mpsc::{SyncSender, TryRecvError, sync_channel},
     },
+    thread::JoinHandle,
 };
 use tracing::{debug, error, info};
 /// Number of state trie segments to fetch concurrently during state sync
@@ -145,6 +146,21 @@ pub struct Store {
     /// those changes already affect the code hash stored in the account, and only
     /// may result in this cache having useless data.
     account_code_cache: Arc<CodeCache>,
+
+    background_threads: Arc<ThreadList>,
+}
+
+#[derive(Debug, Default)]
+struct ThreadList {
+    list: Vec<JoinHandle<()>>,
+}
+
+impl Drop for ThreadList {
+    fn drop(&mut self) {
+        for handle in self.list.drain(..) {
+            let _ = handle.join();
+        }
+    }
 }
 
 pub type StorageTrieNodes = Vec<(H256, Vec<(Nibbles, Vec<u8>)>)>;
@@ -1273,7 +1289,8 @@ impl Store {
                 last_written
             }
         };
-        let store = Self {
+        let mut background_threads = Vec::new();
+        let mut store = Self {
             db_path,
             backend,
             chain_config: Default::default(),
@@ -1283,10 +1300,11 @@ impl Store {
             trie_update_worker_tx: trie_upd_tx,
             last_computed_flatkeyvalue: Arc::new(Mutex::new(last_written)),
             account_code_cache: Arc::new(CodeCache::default()),
+            background_threads: Default::default(),
         };
         let backend_clone = store.backend.clone();
         let last_computed_fkv = store.last_computed_flatkeyvalue.clone();
-        std::thread::spawn(move || {
+        background_threads.push(std::thread::spawn(move || {
             let rx = fkv_rx;
             // Wait for the first Continue to start generation
             loop {
@@ -1302,7 +1320,7 @@ impl Store {
 
             let _ = flatkeyvalue_generator(&backend_clone, &last_computed_fkv, &rx)
                 .inspect_err(|err| error!("Error while generating FlatKeyValue: {err}"));
-        });
+        }));
         let backend = store.backend.clone();
         let flatkeyvalue_control_tx = store.flatkeyvalue_control_tx.clone();
         let trie_cache = store.trie_cache.clone();
@@ -1326,7 +1344,7 @@ impl Store {
 
             - Third, it removes the (no longer needed) bottom-most diff layer from the trie layers in the same way as the first step.
         */
-        std::thread::spawn(move || {
+        background_threads.push(std::thread::spawn(move || {
             let rx = trie_upd_rx;
             loop {
                 match rx.recv() {
@@ -1346,6 +1364,9 @@ impl Store {
                     }
                 }
             }
+        }));
+        store.background_threads = Arc::new(ThreadList {
+            list: background_threads,
         });
         Ok(store)
     }