Repository: man-group/ArcticDB (language: python)

Issue:
Write_batch deadlocks when dedup option is enabled
### Describe the bug

write_batch operation can deadlock when depup=True option is used in the library. The reason is that check_reload function contain a get() for its futures, which breaks the entire chain of futures used by write_batch. Due to timing reasons, it cannot be reproduced in the current test infrastructure. So, it only can be reproduced when we use real storages in the test.

### Steps/Code to Reproduce

def test_write_batch_dedup(library_factory):
    ac = Arctic(<URI_REAL_STORAGE>)
    ac.create_library("test_write_batch_dedup", LibraryOptions(dedup=True))
    lib = ac["test_write_batch_dedup"]
    df1 = pd.DataFrame({'A': [1, 2, 3], 'B': [1, 2, 3]})                                                                                                                                              
    df2 = pd.DataFrame({'A': [10, 20, 30], 'B': [1, 2, 3]})     
    lib.write_batch([WritePayload("TEST_sym_1", df1), WritePayload("TEST_sym_2", df2)])

### Expected Results

No deadlock

### OS, Python Version and ArcticDB Version

Python: 3.6.8 (default, Jun 20 2023, 11:53:23) 
[GCC 4.8.5 20150623 (Red Hat 4.8.5-44)]
OS: Linux-5.15.0-72-generic-x86_64-with-centos-7.9.2009-Core
ArcticDB: 1.4.1.dev0


### Backend storage used

_No response_

### Additional Context

_No response_


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/stream/stream_utils.hpp b/cpp/arcticdb/stream/stream_utils.hpp
index 1c0b3bf09f..dd0ce41b59 100644
--- a/cpp/arcticdb/stream/stream_utils.hpp
+++ b/cpp/arcticdb/stream/stream_utils.hpp
@@ -205,13 +205,13 @@ inline auto generate_segments_from_keys(
     return
         map([&read_store](auto &&key) {
             ARCTICDB_DEBUG(log::inmem(), "Getting segment for key {}: {}", key.type(), key.view());
-            return read_store.read(std::forward<decltype(key)>(key));
+            return read_store.read_sync(std::forward<decltype(key)>(key));
         })
             | window(prefetch_window)
             | move
-            | map([timeout, opts](auto &&fut_key_seg) {
+            | map([timeout, opts](auto &&key_seg) {
                 try {
-                    return std::make_optional(std::forward<decltype(fut_key_seg)>(fut_key_seg).get(timeout));
+                    return std::make_optional(std::forward<decltype(key_seg)>(key_seg));
                 } catch(storage::KeyNotFoundException& e) {
                     if (opts.ignores_missing_key_) {
                         return std::optional<ReadKeyOutput>();
@@ -229,10 +229,10 @@ inline auto generate_keys_from_segments(
     std::optional<entity::KeyType> expected_index_type = std::nullopt) {
     return folly::gen::map([expected_key_type, expected_index_type, &read_store](auto &&key_seg) {
         return folly::gen::detail::GeneratorBuilder<entity::AtomKey>() + [&](auto &&yield) {
-            std::stack<folly::Future<std::pair<entity::VariantKey, SegmentInMemory>>> key_segs;
-            key_segs.push(folly::makeFuture(std::forward<decltype(key_seg)>(key_seg)));
+            std::stack<std::pair<entity::VariantKey, SegmentInMemory>> key_segs;
+            key_segs.push(std::forward<decltype(key_seg)>(key_seg));
             while(!key_segs.empty()) {
-                auto [key, seg] = std::move(key_segs.top()).get();
+                auto [key, seg] = std::move(key_segs.top());
                 key_segs.pop();
                 for (ssize_t i = 0; i < ssize_t(seg.row_count()); ++i) {
                     auto read_key = read_key_row(seg, i);
@@ -241,7 +241,7 @@ inline auto generate_keys_from_segments(
                             "Found unsupported key type in index segment. Expected {} or (index) {}, actual {}",
                             expected_key_type, expected_index_type.value(), read_key
                         );
-                        key_segs.push(read_store.read(read_key));
+                        key_segs.push(read_store.read_sync(read_key));
                     }
                     yield(read_key);
                 }
diff --git a/cpp/arcticdb/util/key_utils.hpp b/cpp/arcticdb/util/key_utils.hpp
index 0f45de032a..64a9ab859d 100644
--- a/cpp/arcticdb/util/key_utils.hpp
+++ b/cpp/arcticdb/util/key_utils.hpp
@@ -41,12 +41,34 @@ inline void delete_keys_of_type_if(const std::shared_ptr<Store>& store, Predicat
     }
 }
 
+template<class Predicate>
+inline void delete_keys_of_type_if_sync(const std::shared_ptr<Store>& store, Predicate&& predicate, KeyType key_type, const std::string& prefix = std::string(), bool continue_on_error = false) {
+    try {
+        store->iterate_type(key_type, [predicate=std::forward<Predicate>(predicate), store=store](VariantKey &&key) {
+            if(predicate(key))
+                store->remove_key_sync(key);
+        }, prefix);
+    }
+    catch(const std::exception& ex) {
+        if(continue_on_error)
+            log::storage().warn("Caught exception {} trying to delete key, continuing", ex.what());
+        else
+            throw;
+    }
+}
+
 inline void delete_keys_of_type_for_stream(const std::shared_ptr<Store>& store, const StreamId& stream_id, KeyType key_type, bool continue_on_error = false) {
     auto prefix = std::holds_alternative<StringId>(stream_id) ? std::get<StringId>(stream_id) : std::string();
     auto match_stream_id =  [&stream_id](const VariantKey & k){ return variant_key_id(k) == stream_id; };
     delete_keys_of_type_if(store, std::move(match_stream_id), key_type, prefix, continue_on_error);
 }
 
+inline void delete_keys_of_type_for_stream_sync(const std::shared_ptr<Store>& store, const StreamId& stream_id, KeyType key_type, bool continue_on_error = false) {
+    auto prefix = std::holds_alternative<StringId>(stream_id) ? std::get<StringId>(stream_id) : std::string();
+    auto match_stream_id =  [&stream_id](const VariantKey & k){ return variant_key_id(k) == stream_id; };
+    delete_keys_of_type_if_sync(store, std::move(match_stream_id), key_type, prefix, continue_on_error);
+}
+
 inline void delete_all_keys_of_type(KeyType key_type, const std::shared_ptr<Store>& store, bool continue_on_error) {
     auto match_stream_id = [](const VariantKey &){ return true; };
     delete_keys_of_type_if(store, std::move(match_stream_id), key_type, std::string{}, continue_on_error);
diff --git a/cpp/arcticdb/version/version_map.hpp b/cpp/arcticdb/version/version_map.hpp
index 6cb531124e..12cff72b1a 100644
--- a/cpp/arcticdb/version/version_map.hpp
+++ b/cpp/arcticdb/version/version_map.hpp
@@ -493,8 +493,7 @@ class VersionMapImpl {
         if (validate_)
             entry->validate();
 
-        auto fut_journal_key = journal_single_key(store, key, entry->head_);
-        auto journal_key = to_atom(std::move(fut_journal_key).get());
+        auto journal_key = to_atom(std::move(journal_single_key(store, key, entry->head_)));
         write_to_entry(entry, key, journal_key);
         write_symbol_ref(store, key, journal_key);
     }
@@ -610,10 +609,10 @@ class VersionMapImpl {
 
     AtomKey write_entry_to_storage(std::shared_ptr<Store> store, const StreamId &stream_id, VersionId version_id,
                                    const std::shared_ptr<VersionMapEntry> &entry) {
-        folly::Future<VariantKey> journal_key_fut = folly::Future<VariantKey>::makeEmpty();
+        VariantKey journal_key;
         entry->validate_types();
 
-        IndexAggregator<RowCountIndex> version_agg(stream_id, [&store, &journal_key_fut, &version_id, &stream_id](auto &&segment) {
+        IndexAggregator<RowCountIndex> version_agg(stream_id, [&store, &journal_key, &version_id, &stream_id](auto &&segment) {
             stream::StreamSink::PartialKey pk{
                     KeyType::VERSION,
                     version_id,
@@ -621,7 +620,7 @@ class VersionMapImpl {
                     IndexValue(0),
                     IndexValue(0)};
 
-            journal_key_fut = store->write_sync(pk, std::forward<SegmentInMemory>(segment));
+            journal_key = store->write_sync(pk, std::forward<SegmentInMemory>(segment));
         });
 
         for (const auto &key : entry->keys_) {
@@ -629,9 +628,9 @@ class VersionMapImpl {
         }
 
         version_agg.commit();
-        auto journal_key =  to_atom(std::move(journal_key_fut).get());
-        write_symbol_ref(store, *entry->keys_.cbegin(), journal_key);
-        return journal_key;
+        auto journal_atom_key =  to_atom(std::move(journal_key));
+        write_symbol_ref(store, *entry->keys_.cbegin(), journal_atom_key);
+        return journal_atom_key;
     }
 
     bool has_stored_entry(std::shared_ptr<StreamSource> store, const StreamId &stream_id) const {
@@ -683,14 +682,14 @@ class VersionMapImpl {
         return entry;
     }
 
-    folly::Future<VariantKey> journal_single_key(
+    VariantKey journal_single_key(
         std::shared_ptr<StreamSink> store,
         const AtomKey &key,
         std::optional<AtomKey> prev_journal_key) {
         ARCTICDB_SAMPLE(WriteJournalEntry, 0)
         ARCTICDB_DEBUG(log::version(), "Version map writing version for key {}", key);
 
-        folly::Future<VariantKey> journal_key = folly::Future<VariantKey>::makeEmpty();
+        VariantKey journal_key;
         IndexAggregator<RowCountIndex> journal_agg(key.id(), [&store, &journal_key, &key](auto &&segment) {
             stream::StreamSink::PartialKey pk{
                 KeyType::VERSION,
@@ -877,7 +876,7 @@ class VersionMapImpl {
         if (auto old_entry = load_from_old_journal_keys(store, stream_id); !old_entry->keys_.empty()) {
             entry->keys_ = std::move(old_entry->keys_);
             entry->head_ = rewrite_old_journal_keys(store, stream_id, entry);
-            delete_keys_of_type_for_stream(store, stream_id, KeyType::VERSION_JOURNAL);
+            delete_keys_of_type_for_stream_sync(store, stream_id, KeyType::VERSION_JOURNAL);
         }
         return entry;
     }
diff --git a/cpp/arcticdb/version/version_utils.hpp b/cpp/arcticdb/version/version_utils.hpp
index 00c53df1b9..b00e14031e 100644
--- a/cpp/arcticdb/version/version_utils.hpp
+++ b/cpp/arcticdb/version/version_utils.hpp
@@ -126,7 +126,7 @@ std::shared_ptr<VersionMapEntry> build_version_map_entry_with_predicate_iteratio
                                 read_keys.push_back(key);
                                 ARCTICDB_DEBUG(log::storage(), "Version map iterating key {}", key);
                                 if (perform_read_segment_with_keys) {
-                                    auto [kv, seg] = store->read(to_atom(key)).get();
+                                    auto [kv, seg] = store->read_sync(to_atom(key));
                                     read_segment_with_keys(seg, output);
                                 }
                             },


Test patch:
diff --git a/python/tests/integration/arcticdb/test_arctic.py b/python/tests/integration/arcticdb/test_arctic.py
index 3c33f981f1..acfc354eba 100644
--- a/python/tests/integration/arcticdb/test_arctic.py
+++ b/python/tests/integration/arcticdb/test_arctic.py
@@ -782,6 +782,46 @@ def test_write_batch(library_factory):
         assert_frame_equal(read_batch_result[sym].data, original_dataframe)
 
 
+def test_write_batch_dedup(library_factory):
+    """Should be able to write different size of batch of data."""
+    lib = library_factory(LibraryOptions(rows_per_segment=10, dedup=True))
+    assert lib._nvs._lib_cfg.lib_desc.version.write_options.segment_row_size == 10
+    assert lib._nvs._lib_cfg.lib_desc.version.write_options.de_duplication == True
+    num_days = 40
+    num_symbols = 10
+    num_versions = 4
+    dt = datetime(2019, 4, 8, 0, 0, 0)
+    column_length = 4
+    num_columns = 5
+    num_rows_per_day = 1
+    read_requests = []
+    list_dataframes = {}
+    columns = random_strings_of_length(num_columns, num_columns, True)
+    df = generate_dataframe(random.sample(columns, num_columns), dt, num_days, num_rows_per_day)
+    for v in range(num_versions):
+        write_requests = []
+        for sym in range(num_symbols):
+            write_requests.append(WritePayload("symbol_" + str(sym), df, metadata="great_metadata" + str(v)))
+            read_requests.append("symbol_" + str(sym))
+            list_dataframes[sym] = df
+        write_batch_result = lib.write_batch(write_requests)
+        assert all(type(w) == PythonVersionedItem for w in write_batch_result)
+
+    read_batch_result = lib.read_batch(read_requests)
+    for sym in range(num_symbols):
+        original_dataframe = list_dataframes[sym]
+        assert read_batch_result[sym].metadata == "great_metadata" + str(num_versions - 1)
+        assert_frame_equal(read_batch_result[sym].data, original_dataframe)
+
+    num_segments = int(
+        (num_days * num_rows_per_day) / lib._nvs._lib_cfg.lib_desc.version.write_options.segment_row_size
+    )
+    for sym in range(num_symbols):
+        data_key_version = lib._nvs.read_index("symbol_" + str(sym))["version_id"]
+        for s in range(num_segments):
+            assert data_key_version[s] == 0
+
+
 def test_write_with_unpacking(arctic_library):
     """Check the syntactic sugar that lets us unpack WritePayload in `write` calls using *."""
     lib = arctic_library