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

Issue:
Segmentation fault flaky test `test_add_to_snapshot_specific_version[version_store_factory]`
### Describe the bug

We have a flaky test which crashes. Example [here](https://github.com/man-group/ArcticDB/actions/runs/10113958498/job/27971903610?pr=1714)

Pasting call stack below:
```
test_snapshot.py::test_add_to_snapshot_specific_version[version_store_factory] Fatal Python error: Segmentation fault

Thread 0x00007f16612ad740 (most recent call first):
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/arcticdb/version_store/_store.py", line 2199 in add_to_snapshot
  File "/tmp/parallel_test/tests/integration/arcticdb/version_store/test_snapshot.py", line 312 in test_add_to_snapshot_specific_version
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/python.py", line 183 in pytest_pyfunc_call
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_callers.py", line 39 in _multicall
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_manager.py", line 80 in _hookexec
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_hooks.py", line 265 in __call__
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/python.py", line 1641 in runtest
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/runner.py", line 162 in pytest_runtest_call
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_callers.py", line 39 in _multicall
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_manager.py", line 80 in _hookexec
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_hooks.py", line 265 in __call__
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/runner.py", line 255 in <lambda>
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/runner.py", line 311 in from_call
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/runner.py", line 255 in call_runtest_hook
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/runner.py", line 215 in call_and_report
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/runner.py", line 126 in runtestprotocol
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/runner.py", line 109 in pytest_runtest_protocol
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_callers.py", line 39 in _multicall
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_manager.py", line 80 in _hookexec
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_hooks.py", line 265 in __call__
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/main.py", line 348 in pytest_runtestloop
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_callers.py", line 39 in _multicall
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_manager.py", line 80 in _hookexec
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_hooks.py", line 265 in __call__
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/main.py", line 323 in _main
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/main.py", line 269 in wrap_session
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/main.py", line 316 in pytest_cmdline_main
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_callers.py", line 39 in _multicall
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_manager.py", line 80 in _hookexec
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pluggy/_hooks.py", line 265 in __call__
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/config/__init__.py", line 163 in main
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/_pytest/config/__init__.py", line 185 in console_main
  File "/opt/python/cp36-cp36m/lib/python3.6/site-packages/pytest/__main__.py", line 5 in <module>
  File "/opt/python/cp36-cp36m/lib/python3.6/runpy.py", line 85 in _run_code
  File "/opt/python/cp36-cp36m/lib/python3.6/runpy.py", line 193 in _run_module_as_main
*** Segmentation fault
Register dump:

 RAX: 0000000000000000   RBX: 0000000000000000   RCX: 00007f1660c864fb
 RDX: 000000000000000b   RSI: 0000000000000130   RDI: 0000000000000106
 RBP: 00007f1616ffd4f0   R8 : 00007f1616ffcd61   R9 : 00007f1616ffcd50
 R10: 0000000000000008   R11: 0000000000000206   R12: 00007f1616ffd520
 R13: 0000000000000001   R14: 0000000002936e20   R15: 00007f1633cf9608
 RSP: 00007f1616ffce38

... trimming on purpose
```


### Steps/Code to Reproduce

Rerun the test many times, I've seen it fail just once

### Expected Results

Test not failing

### OS, Python Version and ArcticDB Version

It failed for linux python 3.6

### Backend storage used

_No response_

### Additional Context

_No response_


Discussion / hints:
It looks like it is not related the version map refactor. I did in fact managed to reproduce in a debugger and it fails on [this operation](https://github.com/man-group/ArcticDB/blob/bbf594f8c200231bcf50a81c64e396bb42d19cda/cpp/arcticdb/version/version_map_batch_methods.hpp#L182) in the batch version map operations. This looks like some thread safety issue with std::advance although that seems quite weird. I'll look more into it.

Fix patch:
diff --git a/cpp/arcticdb/version/version_map_batch_methods.hpp b/cpp/arcticdb/version/version_map_batch_methods.hpp
index fe1425c90c..99ae5e4678 100644
--- a/cpp/arcticdb/version/version_map_batch_methods.hpp
+++ b/cpp/arcticdb/version/version_map_batch_methods.hpp
@@ -48,10 +48,10 @@ inline std::optional<std::string> collect_futures_exceptions(auto&& futures) {
 }
 
 template <typename Inputs, typename TaskSubmitter, typename ResultHandler>
-inline void submit_tasks_for_range(const Inputs& inputs, TaskSubmitter submitter, ResultHandler result_handler) {
+inline void submit_tasks_for_range(Inputs inputs, TaskSubmitter submitter, ResultHandler result_handler) {
     const auto window_size = async::TaskScheduler::instance()->io_thread_count() * 2;
 
-    auto futures = folly::window(inputs, [&submitter, &result_handler](const auto &input) {
+    auto futures = folly::window(std::move(inputs), [&submitter, &result_handler](const auto &input) {
         return submitter(input).thenValue([&result_handler, &input](auto &&r) {
             auto result = std::forward<decltype(r)>(r);
             result_handler(input, std::move(result));
@@ -163,27 +163,6 @@ inline std::vector<folly::Future<version_store::UpdateInfo>> batch_get_latest_un
     return vector_fut;
 }
 
-template <typename MapType>
-struct MapRandomAccessWrapper {
-    const MapType& map_;
-
-    using iterator = typename MapType::iterator;
-
-    explicit MapRandomAccessWrapper(const MapType& map) :
-        map_(map) {
-    }
-
-    [[nodiscard]] size_t size() const {
-        return map_.size();
-    }
-
-    auto& operator[](size_t pos) const {
-        auto it = std::begin(map_);
-        std::advance(it, pos);
-        return *it;
-    }
-};
-
 // This version assumes that there is only one version per symbol, so no need for the state machine below
 inline std::shared_ptr<std::unordered_map<StreamId, AtomKey>> batch_get_specific_version(
     const std::shared_ptr<Store>& store,
@@ -197,8 +176,8 @@ inline std::shared_ptr<std::unordered_map<StreamId, AtomKey>> batch_get_specific
     auto tombstoned_vers = std::make_shared<std::vector<std::pair<StreamId, AtomKey>>>();
     auto tombstoned_vers_mutex = std::make_shared<std::mutex>();
 
-    MapRandomAccessWrapper wrapper{sym_versions};
-    submit_tasks_for_range(wrapper, [store, version_map](auto& sym_version) {
+    auto tasks_input = std::vector(sym_versions.begin(), sym_versions.end());
+    submit_tasks_for_range(std::move(tasks_input), [store, version_map](auto& sym_version) {
             LoadStrategy load_strategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(sym_version.second)};
             return async::submit_io_task(CheckReloadTask{store, version_map, sym_version.first, load_strategy});
         },
@@ -243,14 +222,14 @@ using VersionVectorType = std::vector<VersionId>;
 inline std::shared_ptr<std::unordered_map<std::pair<StreamId, VersionId>, AtomKey>> batch_get_specific_versions(
         const std::shared_ptr<Store>& store,
         const std::shared_ptr<VersionMap>& version_map,
-        std::map<StreamId, VersionVectorType>& sym_versions,
+        const std::map<StreamId, VersionVectorType>& sym_versions,
         bool include_deleted = true) {
     ARCTICDB_SAMPLE(BatchGetLatestVersion, 0)
     auto output = std::make_shared<std::unordered_map<std::pair<StreamId, VersionId>, AtomKey>>();
-    MapRandomAccessWrapper wrapper{sym_versions};
     auto mutex = std::make_shared<std::mutex>();
 
-    submit_tasks_for_range(wrapper, [store, version_map](auto sym_version) {
+    auto tasks_input = std::vector(sym_versions.begin(), sym_versions.end());
+    submit_tasks_for_range(std::move(tasks_input), [store, version_map](auto sym_version) {
                 auto first_version = *std::min_element(std::begin(sym_version.second), std::end(sym_version.second));
                 LoadStrategy load_strategy{LoadType::DOWNTO, LoadObjective::UNDELETED_ONLY, static_cast<SignedVersionId>(first_version)};
                 return async::submit_io_task(CheckReloadTask{store, version_map, sym_version.first, load_strategy});


Test patch:
diff --git a/cpp/arcticdb/version/test/test_version_map_batch.cpp b/cpp/arcticdb/version/test/test_version_map_batch.cpp
index 9713741824..c18467f9c1 100644
--- a/cpp/arcticdb/version/test/test_version_map_batch.cpp
+++ b/cpp/arcticdb/version/test/test_version_map_batch.cpp
@@ -270,3 +270,21 @@ TEST_F(VersionMapBatchStore, CombinedQueries) {
         }
     }
 }
+
+TEST_F(VersionMapBatchStore, SpecificVersionsShouldCopyInput) {
+    SKIP_WIN("Exceeds LMDB map size");
+    ScopedConfig sc("VersionMap.ReloadInterval", std::numeric_limits<int64_t>::max());
+    auto store = test_store_->_test_get_store();
+    auto version_map = std::make_shared<VersionMap>();
+    std::string symbol = "symbol";
+    std::string symbol_2 = "symbol_2";
+
+    add_versions_for_stream(version_map, store, symbol, 5);
+
+    for (uint64_t i=0; i<1000; ++i) {
+        auto sym_versions = std::map<StreamId, VersionVectorType>{{symbol, {4}}};
+        batch_get_specific_versions(store, version_map, sym_versions);
+        // We add to the sym_versions a missing symbol after the batch_get to mimic the issue: https://github.com/man-group/ArcticDB/issues/1716
+        sym_versions.insert({symbol_2, {50}});
+    }
+}