[BUG] Lock BF writer (#2189)

## Description of changes

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
- Locks the BF writer for write operations since we need to make these
properly thread safe
 - New functionality
	 - None

## Test plan
*How are these changes tested?*
- [x] Tests pass locally with `pytest` for python, `yarn test` for js,
`cargo test` for rust

## Documentation Changes
None

Author: HammadB

rust/worker/src/blockstore/arrow/blockfile.rs

@@ -24,6 +24,7 @@ pub(crate) struct ArrowBlockfileWriter {
     block_deltas: Arc<Mutex<HashMap<Uuid, BlockDelta>>>,
     sparse_index: SparseIndex,
     id: Uuid,
+    write_mutex: Arc<tokio::sync::Mutex<()>>,
 }
 // TODO: method visibility should not be pub(crate)
 
@@ -62,6 +63,7 @@ impl ArrowBlockfileWriter {
             block_deltas: block_deltas,
             sparse_index: sparse_index,
             id,
+            write_mutex: Arc::new(tokio::sync::Mutex::new(())),
         }
     }
 
@@ -78,6 +80,7 @@ impl ArrowBlockfileWriter {
             block_deltas: block_deltas,
             sparse_index: new_sparse_index,
             id,
+            write_mutex: Arc::new(tokio::sync::Mutex::new(())),
         }
     }
 
@@ -110,6 +113,9 @@ impl ArrowBlockfileWriter {
         key: K,
         value: V,
     ) -> Result<(), Box<dyn ChromaError>> {
+        // TODO: for now the BF writer locks the entire write operation
+        let _guard = self.write_mutex.lock().await;
+
         // TODO: value must be smaller than the block size except for position lists, which are a special case
         //         // where we split the value across multiple blocks
         //         if !self.in_transaction() {
@@ -173,6 +179,7 @@ impl ArrowBlockfileWriter {
         prefix: &str,
         key: K,
     ) -> Result<(), Box<dyn ChromaError>> {
+        let _guard = self.write_mutex.lock().await;
         // Get the target block id for the key
         let search_key = CompositeKey::new(prefix.to_string(), key.clone());
         let target_block_id = self.sparse_index.get_target_block_id(&search_key);