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

Issue:
Flaky tests
Tracking ticket for flaky tests

- [x] #469 
- [x] #316 
- [x] [3.8 windows test_filter_categorical[lmdb_version_store]](https://github.com/man-group/ArcticDB/actions/runs/5281710778/jobs/9556935531)
- [x] [3.8 windows test_filter_categorical[lmdb_version_store_dynamic_schema]](https://github.com/man-group/ArcticDB/actions/runs/5281710778/jobs/9556935531)
- [x] [3.7 windows test_filter_categorical[lmdb_version_store_dynamic_schema]](https://github.com/man-group/ArcticDB/actions/runs/5310593194/jobs/9613617253)
    -> Through debugging, the underlying issue is #229
- [x] [linux 3.11 test_filter_numeric_isnotin_unsigned](https://github.com/man-group/ArcticDB/actions/runs/5288922390/jobs/9571531805?pr=490) -> #604
- [x] [linux 3.9 test_filter_numeric_isnotin_unsigned](https://github.com/man-group/ArcticDB/actions/runs/5422309116/jobs/9859134990)
- [x] [Windows 3.11 test_filter_numeric_isnotin_unsigned](https://github.com/man-group/ArcticDB/actions/runs/5453649236/jobs/9924117342)
- [x] [test_filtering.py::test_filter_numeric_isnotin_unsigned](https://github.com/man-group/ArcticDB/actions/runs/5313566049/jobs/9621422799)
- [x] [linux 3.11 test_hypothesis_mean_agg_dynamic](https://github.com/man-group/ArcticDB/actions/runs/5422309116/jobs/9859052796)
    -> Not trivially fixable/skippable. See #603
- [x] [Windows 3.7 test_basic_version_store.py::test_read_ts[EncodingVersion.V2]](https://github.com/man-group/ArcticDB/actions/runs/5464577790/jobs/9947764682)
    -> Fix attempt bundled with #604
- [ ] [Windows 3.8 `test_find_version`](https://github.com/man-group/ArcticDB/actions/runs/5556047996/jobs/10149716495?pr=604#step:9:633)
    -> Looks similar in nature to `test_read_ts` above
- [x] `test_append_with_cont_mem_problem ` below #985 
- [x] `test_append_with_defragmentation` below #985 
- [x] [`test_column_names_by_timestamp`](https://github.com/man-group/ArcticDB/actions/runs/5748892650/job/15583901448?pr=600#step:9:4984) (need to use `distinct_timestamps` on it)
- [x] [test_filter_numeric_isnotin_unsigned](https://github.com/man-group/ArcticDB/actions/runs/6904642008/job/18786268872)
- [x] [linux 3.11 test_hypothesis_sum_agg](https://github.com/man-group/ArcticDB/actions/runs/6863175524/job/18662722363)
- [x] #1153 
- [x] #1154 


Discussion / hints:
`test_append_with_cont_mem_problem` added about a month ago is flaky

[Code for it](https://github.com/man-group/ArcticDB/blame/3717cbb8acf90d09ccd74286bbe96454a05a925f/python/tests/unit/arcticdb/version_store/test_append.py#L650)

[Example job run](https://github.com/man-group/ArcticDB/actions/runs/5334866596/jobs/9668884882)

```
1: ================================== FAILURES ===================================
1: ______________________ test_append_with_cont_mem_problem ______________________
1: 
1: sym = 'test2023-06-21T14_56_11_375132'
1: lmdb_version_store_tiny_segment_dynamic = NativeVersionStore: Library: local.test_698_2023-06-21T14_56_11_375132, Primary Storage: lmdb_storage.
1: 
1:     def test_append_with_cont_mem_problem(sym, lmdb_version_store_tiny_segment_dynamic):
1:         set_config_int("SymbolDataCompact.SegmentCount", 1)
1:         df0 = pd.DataFrame({"0": ["01234567890123456"]}, index=[pd.Timestamp(0)])
1:         df1 = pd.DataFrame({"0": ["012345678901234567"]}, index=[pd.Timestamp(1)])
1:         df2 = pd.DataFrame({"0": ["0123456789012345678"]}, index=[pd.Timestamp(2)])
1:         df3 = pd.DataFrame({"0": ["01234567890123456789"]}, index=[pd.Timestamp(3)])
1:         df = pd.concat([df0, df1, df2, df3])
1:     
1:         for _ in range(100):
1:             lib = lmdb_version_store_tiny_segment_dynamic
1:             lib.write(sym, df0).version
1:             lib.append(sym, df1).version
1:             lib.append(sym, df2).version
1:             lib.append(sym, df3).version
1: >           lib.version_store.defragment_symbol_data(sym, None)
1: E           arcticdb_ext.storage.NoDataFoundException: test2023-06-21T14_56_11_375132
1: 
```

Fix patch:
diff --git a/cpp/arcticdb/stream/stream_utils.hpp b/cpp/arcticdb/stream/stream_utils.hpp
index bda00a502b..9372a706b4 100644
--- a/cpp/arcticdb/stream/stream_utils.hpp
+++ b/cpp/arcticdb/stream/stream_utils.hpp
@@ -403,22 +403,15 @@ storage::KeySegmentPair make_target_key(KeyType key_type,
     if (is_ref_key_class(key_type)) {
         return {RefKey{stream_id, key_type}, std::move(segment)};
     } else {
-        if (is_ref_key_class(variant_key_type(source_key))) {
-            auto return_segment = std::move(segment);
-            auto clone = return_segment;
-            auto decoded = decode_segment(std::move(clone));
-            auto index = index_type_from_descriptor(decoded.descriptor());
-            auto range = get_range_from_segment(index, decoded);
-            auto new_key = atom_key_builder().version_id(version_id).creation_ts(ClockType::nanos_since_epoch()).start_index(range.start_).end_index(
-                range.end_).build(stream_id, key_type);
-
-            return {new_key, std::move(return_segment)};
-        } else {
-            auto new_key = atom_key_builder().version_id(version_id).creation_ts(ClockType::nanos_since_epoch()).start_index(
-                to_atom(source_key).start_index()).end_index(to_atom(source_key).end_index()).build(stream_id, key_type);
-
-            return {new_key, std::move(segment)};
-        }
+        util::check(!is_ref_key_class(variant_key_type(source_key)),
+            "Cannot convert ref key {} to {}", source_key, key_type);
+        auto& atom_source_key = to_atom(source_key);
+        auto new_key = atom_key_builder().version_id(version_id).creation_ts(ClockType::nanos_since_epoch())
+            .start_index(atom_source_key.start_index()).end_index(atom_source_key.end_index())
+            .content_hash(atom_source_key.content_hash())
+            .build(stream_id, key_type);
+
+        return {new_key, std::move(segment)};
     }
 }
 
diff --git a/cpp/arcticdb/util/clock.hpp b/cpp/arcticdb/util/clock.hpp
index 9beca99f22..0ddc5481cc 100644
--- a/cpp/arcticdb/util/clock.hpp
+++ b/cpp/arcticdb/util/clock.hpp
@@ -46,8 +46,16 @@ struct LinearClock {
     }
 };
 
+struct ManualClock {
+    inline static std::atomic<entity::timestamp> time_{0};
 
-} // namespace arcticdb::util
-
+    static entity::timestamp nanos_since_epoch() {
+        return time_.load();
+    }
+    static entity::timestamp coarse_nanos_since_epoch() {
+        return time_.load();
+    }
+};
 
 
+} // namespace arcticdb::util
diff --git a/cpp/arcticdb/version/local_versioned_engine.cpp b/cpp/arcticdb/version/local_versioned_engine.cpp
index 10597ab714..7e35a3f579 100644
--- a/cpp/arcticdb/version/local_versioned_engine.cpp
+++ b/cpp/arcticdb/version/local_versioned_engine.cpp
@@ -24,11 +24,11 @@
 #include <arcticdb/python/gil_lock.hpp>
 
 namespace arcticdb::version_store {
-
+template<class ClockType>
 LocalVersionedEngine::LocalVersionedEngine(
         const std::shared_ptr<storage::Library>& library,
-        const std::optional<std::string>& license_key ARCTICDB_UNUSED) :
-    store_(std::make_shared<async::AsyncStore<util::SysClock>>(library, codec::default_lz4_codec(), encoding_version(library->config()))),
+        const ClockType&) :
+    store_(std::make_shared<async::AsyncStore<ClockType>>(library, codec::default_lz4_codec(), encoding_version(library->config()))),
     symbol_list_(std::make_shared<SymbolList>(version_map_)){
     configure(library->config());
     ARCTICDB_RUNTIME_DEBUG(log::version(), "Created versioned engine at {} for library path {}  with config {}", uintptr_t(this),
@@ -44,6 +44,9 @@ LocalVersionedEngine::LocalVersionedEngine(
     }
 }
 
+template LocalVersionedEngine::LocalVersionedEngine(const std::shared_ptr<storage::Library>& library, const util::SysClock&);
+template LocalVersionedEngine::LocalVersionedEngine(const std::shared_ptr<storage::Library>& library, const util::ManualClock&);
+
 folly::Future<folly::Unit> LocalVersionedEngine::delete_unreferenced_pruned_indexes(
         const std::vector<AtomKey> &pruned_indexes,
         const AtomKey& key_to_keep
@@ -839,7 +842,7 @@ folly::Future<folly::Unit> LocalVersionedEngine::delete_trees_responsibly(
     log::version().debug("Forbidden: {} total of data keys", data_keys_not_to_be_deleted.size());
     RemoveOpts remove_opts;
     remove_opts.ignores_missing_key_ = true;
-    
+
     std::vector<entity::VariantKey> vks_column_stats;
     std::transform(keys_to_delete->begin(),
                     keys_to_delete->end(),
diff --git a/cpp/arcticdb/version/local_versioned_engine.hpp b/cpp/arcticdb/version/local_versioned_engine.hpp
index 2fec960161..f181c2d7f3 100644
--- a/cpp/arcticdb/version/local_versioned_engine.hpp
+++ b/cpp/arcticdb/version/local_versioned_engine.hpp
@@ -45,9 +45,11 @@ struct IndexKeyAndUpdateInfo{
 class LocalVersionedEngine : public VersionedEngine {
 
 public:
+    template<class ClockType = util::SysClock>
     explicit LocalVersionedEngine(
         const std::shared_ptr<storage::Library>& library,
-        const std::optional<std::string>& license_key = std::nullopt);
+        const ClockType& = util::SysClock{} // Only used to allow the template variable to be inferred
+        );
 
     virtual ~LocalVersionedEngine() = default;
 
@@ -377,6 +379,12 @@ class LocalVersionedEngine : public VersionedEngine {
     void _test_set_store(std::shared_ptr<Store> store);
     std::shared_ptr<VersionMap> _test_get_version_map();
 
+    /** Get the time used by the Store (e.g. that would be used in the AtomKey).
+        For testing purposes only. */
+    entity::timestamp get_store_current_timestamp_for_tests() {
+        return store()->current_timestamp();
+    }
+
 protected:
     VersionedItem compact_incomplete_dynamic(
             const StreamId& stream_id,
diff --git a/cpp/arcticdb/version/python_bindings.cpp b/cpp/arcticdb/version/python_bindings.cpp
index d5816636d8..e17b66682f 100644
--- a/cpp/arcticdb/version/python_bindings.cpp
+++ b/cpp/arcticdb/version/python_bindings.cpp
@@ -361,7 +361,9 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
             .def_readwrite("row_filter",&UpdateQuery::row_filter);
 
     py::class_<PythonVersionStore>(version, "PythonVersionStore")
-        .def(py::init<std::shared_ptr<storage::Library>, std::optional<std::string>>(),
+        .def(py::init([](const std::shared_ptr<storage::Library>& library, std::optional<std::string>) {
+                return PythonVersionStore(library);
+             }),
              py::arg("library"),
              py::arg("license_key") = std::nullopt)
         .def("write_partitioned_dataframe",
@@ -649,8 +651,18 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
         .def("latest_timestamp",
              &PythonVersionStore::latest_timestamp,
              "Returns latest timestamp of a symbol")
+        .def("get_store_current_timestamp_for_tests",
+             &PythonVersionStore::get_store_current_timestamp_for_tests,
+             "For testing purposes only")
         ;
 
+    py::class_<ManualClockVersionStore, PythonVersionStore>(version, "ManualClockVersionStore")
+        .def(py::init<const std::shared_ptr<storage::Library>&>())
+        .def_property_static("time",
+            []() { return util::ManualClock::time_.load(); },
+            [](const py::class_<ManualClockVersionStore>&, entity::timestamp ts) { util::ManualClock::time_ = ts; })
+         ;
+
     py::class_<LocalVersionedEngine>(version, "VersionedEngine")
       .def(py::init<std::shared_ptr<storage::Library>>())
       .def("read_versioned_dataframe",
diff --git a/cpp/arcticdb/version/symbol_list.cpp b/cpp/arcticdb/version/symbol_list.cpp
index b64cc10953..2430441675 100644
--- a/cpp/arcticdb/version/symbol_list.cpp
+++ b/cpp/arcticdb/version/symbol_list.cpp
@@ -107,8 +107,13 @@ bool SymbolList::can_update_symbol_list(const std::shared_ptr<Store>& store,
             });
 
         seg.end_row();
-        store->write_sync(KeyType::SYMBOL_LIST, 0, StreamId{action}, IndexValue{symbol}, IndexValue{symbol},
+        try {
+            store->write_sync(KeyType::SYMBOL_LIST, 0, StreamId{ action }, IndexValue{ symbol }, IndexValue{ symbol },
                 std::move(seg));
+        } catch (const DuplicateKeyException& e [[unused]]) {
+            // Both version and content hash are fixed, so collision is possible
+            ARCTICDB_DEBUG(log::storage(), "Symbol list DuplicateKeyException: {}", e.what());
+        }
     }
 
     SymbolList::CollectionType SymbolList::load_from_version_keys(const std::shared_ptr<Store>& store) {
diff --git a/cpp/arcticdb/version/version_store_api.cpp b/cpp/arcticdb/version/version_store_api.cpp
index 8cc36a1cff..4d4d1ab435 100644
--- a/cpp/arcticdb/version/version_store_api.cpp
+++ b/cpp/arcticdb/version/version_store_api.cpp
@@ -38,9 +38,8 @@ using namespace arcticdb::entity;
 namespace as = arcticdb::stream;
 using namespace arcticdb::storage;
 
-PythonVersionStore::PythonVersionStore(const std::shared_ptr<storage::Library>& library, const std::optional<std::string>& license_key) :
-    LocalVersionedEngine(library, license_key) {
-}
+template PythonVersionStore::PythonVersionStore(const std::shared_ptr<storage::Library>& library, const util::SysClock& ct);
+template PythonVersionStore::PythonVersionStore(const std::shared_ptr<storage::Library>& library, const util::ManualClock& ct);
 
 VersionedItem PythonVersionStore::write_dataframe_specific_version(
     const StreamId& stream_id,
diff --git a/cpp/arcticdb/version/version_store_api.hpp b/cpp/arcticdb/version/version_store_api.hpp
index 8fa6ac285d..14811f9e60 100644
--- a/cpp/arcticdb/version/version_store_api.hpp
+++ b/cpp/arcticdb/version/version_store_api.hpp
@@ -51,9 +51,10 @@ namespace as = arcticdb::stream;
 class PythonVersionStore : public LocalVersionedEngine {
 
   public:
-    explicit PythonVersionStore(
-        const std::shared_ptr<storage::Library>& library,
-        const std::optional<std::string>& license_key = std::nullopt);
+    template<class ClockType = util::SysClock>
+    explicit PythonVersionStore(const std::shared_ptr<storage::Library>& library, const ClockType& ct = util::SysClock{}) :
+        LocalVersionedEngine(library, ct) {
+    }
 
     VersionedItem write_dataframe_specific_version(
         const StreamId& stream_id,
@@ -335,6 +336,11 @@ class PythonVersionStore : public LocalVersionedEngine {
     void delete_snapshot_sync(const SnapshotId& snap_name, const VariantKey& snap_key);
 };
 
+struct ManualClockVersionStore : PythonVersionStore {
+    ManualClockVersionStore(const std::shared_ptr<storage::Library>& library) :
+            PythonVersionStore(library, util::ManualClock{}) {}
+};
+
 inline std::vector<std::variant<ReadResult, DataError>> frame_to_read_result(std::vector<ReadVersionOutput>&& keys_frame_and_descriptors) {
     std::vector<std::variant<ReadResult, DataError>> read_results;
     read_results.reserve(keys_frame_and_descriptors.size());


Test patch:
diff --git a/python/arcticdb/util/test.py b/python/arcticdb/util/test.py
index 90a121c553..c6c816641e 100644
--- a/python/arcticdb/util/test.py
+++ b/python/arcticdb/util/test.py
@@ -13,6 +13,8 @@
 import pytest
 import string
 import random
+import time
+import attr
 from six import PY3
 from copy import deepcopy
 from functools import wraps
@@ -20,6 +22,7 @@
 
 from arcticdb.config import Defaults
 from arcticdb.log import configure, logger_by_name
+from arcticdb.version_store import NativeVersionStore
 from arcticdb.version_store._custom_normalizers import CustomNormalizer
 from arcticc.pb2.descriptors_pb2 import NormalizationMetadata
 from arcticc.pb2.logger_pb2 import LoggerConfig, LoggersConfig
@@ -490,3 +493,31 @@ def regularize_dataframe(df):
     output = output.reset_index(drop=True)
     output = output.astype("float", errors="ignore")
     return output
+
+
+@attr.s(slots=True, auto_attribs=True)
+class BeforeAfterTimestamp:
+    before: pd.Timestamp
+    after: Optional[pd.Timestamp]
+
+
+@contextmanager
+def distinct_timestamps(lib: NativeVersionStore):
+    """Ensures the timestamp used by ArcticDB operations before, during and leaving the context are all different.
+
+    Yields
+    ------
+    BeforeAfterTimestamp
+    """
+    get_ts = lib.version_store.get_store_current_timestamp_for_tests
+    before = get_ts()
+    while get_ts() == before:
+        time.sleep(0.000001)  # 1us - The worst resolution in our clock implementations
+    out = BeforeAfterTimestamp(pd.Timestamp(before, unit="ns"), None)
+    try:
+        yield out
+    finally:
+        right_after = get_ts()
+        while get_ts() == right_after:
+            time.sleep(0.000001)
+        out.after = pd.Timestamp(get_ts(), unit="ns")
diff --git a/python/tests/integration/arcticdb/version_store/test_basic_version_store.py b/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
index 6bf069cf86..482b7b1cfe 100644
--- a/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
+++ b/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
@@ -29,14 +29,13 @@
     StreamDescriptorMismatch,
     UserInputException,
 )
-from arcticdb_ext.storage import NoDataFoundException
 from arcticdb.flattener import Flattener
 from arcticdb.version_store import NativeVersionStore
 from arcticdb.version_store._custom_normalizers import CustomNormalizer, register_normalizer
 from arcticdb.version_store._store import UNSUPPORTED_S3_CHARS, MAX_SYMBOL_SIZE, VersionedItem
 from arcticdb_ext.exceptions import _ArcticLegacyCompatibilityException
 from arcticdb_ext.storage import KeyType, NoDataFoundException
-from arcticdb_ext.version_store import NoSuchVersionException, StreamDescriptorMismatch
+from arcticdb_ext.version_store import NoSuchVersionException, StreamDescriptorMismatch, ManualClockVersionStore
 from arcticc.pb2.descriptors_pb2 import NormalizationMetadata  # Importing from arcticdb dynamically loads arcticc.pb2
 from arcticdb.util.test import (
     sample_dataframe,
@@ -45,10 +44,11 @@
     assert_frame_equal,
     assert_series_equal,
     config_context,
+    distinct_timestamps,
 )
 from arcticdb_ext.tools import AZURE_SUPPORT
 from tests.util.date import DateRange
-from arcticdb_ext import set_config_int, unset_config_int
+
 
 if sys.platform == "linux":
     SMOKE_TEST_VERSION_STORES = [
@@ -785,17 +785,17 @@ def test_is_pickled_by_snapshot(lmdb_version_store):
 def test_is_pickled_by_timestamp(lmdb_version_store):
     symbol = "test"
     will_be_pickled = [1, 2, 3]
-    lmdb_version_store.write(symbol, will_be_pickled)
-    time_after_first_write = pd.Timestamp.utcnow()
-    time.sleep(0.1)
+    with distinct_timestamps(lmdb_version_store) as first_write_timestamps:
+        lmdb_version_store.write(symbol, will_be_pickled)
 
     not_pickled = pd.DataFrame({"a": np.arange(3)})
-    lmdb_version_store.write(symbol, not_pickled)
+    with distinct_timestamps(lmdb_version_store):
+        lmdb_version_store.write(symbol, not_pickled)
 
     with pytest.raises(NoDataFoundException):
         lmdb_version_store.read(symbol, pd.Timestamp(0))
     assert lmdb_version_store.is_symbol_pickled(symbol) is False
-    assert lmdb_version_store.is_symbol_pickled(symbol, time_after_first_write) is True
+    assert lmdb_version_store.is_symbol_pickled(symbol, first_write_timestamps.after) is True
     assert lmdb_version_store.is_symbol_pickled(symbol, pd.Timestamp(np.iinfo(np.int64).max)) is False
 
 
@@ -878,17 +878,13 @@ def test_list_versions_with_snapshots(lmdb_version_store):
 
 
 def test_read_ts(lmdb_version_store):
-    lmdb_version_store.write("a", 1)  # v0
-    time.sleep(0.001)  # In case utcnow() has a lower precision and returning a timestamp before the write (#496)
-    time_after_first_write = pd.Timestamp.utcnow()
-
-    assert lmdb_version_store.read("a", as_of=time_after_first_write).version == 0
-    time.sleep(0.1)
-    lmdb_version_store.write("a", 2)  # v1
-    time.sleep(0.11)
-
-    lmdb_version_store.write("a", 3)  # v2
-    time.sleep(0.1)
+    with distinct_timestamps(lmdb_version_store) as first_write_timestamps:
+        lmdb_version_store.write("a", 1)  # v0
+    assert lmdb_version_store.read("a", as_of=first_write_timestamps.after).version == 0
+    with distinct_timestamps(lmdb_version_store):
+        lmdb_version_store.write("a", 2)  # v1
+    with distinct_timestamps(lmdb_version_store):
+        lmdb_version_store.write("a", 3)  # v2
     lmdb_version_store.write("a", 4)  # v3
     lmdb_version_store.snapshot("snap3")
     versions = lmdb_version_store.list_versions()
@@ -912,7 +908,7 @@ def test_read_ts(lmdb_version_store):
     assert vitem.version == 3
     assert vitem.data == 4
 
-    vitem = lmdb_version_store.read("a", as_of=time_after_first_write)
+    vitem = lmdb_version_store.read("a", as_of=first_write_timestamps.after)
     assert vitem.version == 0
     assert vitem.data == 1
 
@@ -1738,6 +1734,8 @@ def test_dynamic_schema_similar_index_column_dataframe_multiple_col(lmdb_version
 
 def test_restore_version(version_store_factory):
     lmdb_version_store = version_store_factory(col_per_group=2, row_per_segment=2)
+    # Triggers bug https://github.com/man-group/ArcticDB/issues/469 by freezing time
+    lmdb_version_store.version_store = ManualClockVersionStore(lmdb_version_store._library)
     symbol = "test_restore_version"
     df1 = get_sample_dataframe(20, 4)
     df1.index = pd.DatetimeIndex([pd.Timestamp.now()] * len(df1))
diff --git a/python/tests/integration/arcticdb/version_store/test_metadata_support.py b/python/tests/integration/arcticdb/version_store/test_metadata_support.py
index 3298cd0da0..16af94661c 100644
--- a/python/tests/integration/arcticdb/version_store/test_metadata_support.py
+++ b/python/tests/integration/arcticdb/version_store/test_metadata_support.py
@@ -5,16 +5,14 @@
 
 As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
 """
-import time
 import numpy as np
 from pandas import DataFrame, Timestamp
 import pytest
-import sys
 
 from arcticdb.version_store import NativeVersionStore, VersionedItem
 from arcticdb.exceptions import ArcticNativeNotYetImplemented
 from arcticdb_ext.storage import NoDataFoundException
-from arcticdb.util.test import assert_frame_equal
+from arcticdb.util.test import assert_frame_equal, distinct_timestamps
 
 
 # In the following lines, the naming convention is
@@ -85,22 +83,21 @@ def test_read_metadata_by_timestamp(lmdb_version_store):
     symbol = "test_symbol"
 
     metadata_v0 = {"something": 1}
-    lmdb_version_store.write(symbol, 1, metadata=metadata_v0)  # v0
-    time_after_first_write = Timestamp.utcnow()
-    time.sleep(0.1)
+    with distinct_timestamps(lmdb_version_store) as first_write_timestamps:
+        lmdb_version_store.write(symbol, 1, metadata=metadata_v0)  # v0
 
     with pytest.raises(NoDataFoundException):
         lmdb_version_store.read(symbol, as_of=Timestamp(0))
 
-    assert lmdb_version_store.read_metadata(symbol, as_of=time_after_first_write).metadata == metadata_v0
+    assert lmdb_version_store.read_metadata(symbol, as_of=first_write_timestamps.after).metadata == metadata_v0
 
     metadata_v1 = {"something more": 2}
-    lmdb_version_store.write(symbol, 2, metadata=metadata_v1)  # v1
-    time.sleep(0.11)
+    with distinct_timestamps(lmdb_version_store):
+        lmdb_version_store.write(symbol, 2, metadata=metadata_v1)  # v1
 
     metadata_v2 = {"something else": 3}
-    lmdb_version_store.write(symbol, 3, metadata=metadata_v2)  # v2
-    time.sleep(0.1)
+    with distinct_timestamps(lmdb_version_store):
+        lmdb_version_store.write(symbol, 3, metadata=metadata_v2)  # v2
 
     metadata_v3 = {"nothing": 4}
     lmdb_version_store.write(symbol, 4, metadata=metadata_v3)  # v3
@@ -109,7 +106,7 @@ def test_read_metadata_by_timestamp(lmdb_version_store):
     assert len(versions) == 4
     sorted_versions_for_a = sorted([v for v in versions if v["symbol"] == symbol], key=lambda x: x["version"])
 
-    assert lmdb_version_store.read_metadata(symbol, as_of=time_after_first_write).metadata == metadata_v0
+    assert lmdb_version_store.read_metadata(symbol, as_of=first_write_timestamps.after).metadata == metadata_v0
 
     ts_for_v1 = sorted_versions_for_a[1]["date"]
     assert lmdb_version_store.read_metadata(symbol, as_of=ts_for_v1).metadata == metadata_v1
diff --git a/python/tests/integration/arcticdb/version_store/test_snapshot.py b/python/tests/integration/arcticdb/version_store/test_snapshot.py
index 9fc124bf98..c8335848dc 100644
--- a/python/tests/integration/arcticdb/version_store/test_snapshot.py
+++ b/python/tests/integration/arcticdb/version_store/test_snapshot.py
@@ -7,7 +7,9 @@
 """
 import pytest
 import numpy as np
+
 from arcticdb_ext.exceptions import InternalException
+from arcticdb.util.test import distinct_timestamps
 
 
 def test_basic_snapshot_flow(lmdb_version_store):
@@ -237,55 +239,31 @@ def test_pruned_symbol_in_symbol_read_version(lmdb_version_store_tombstone_and_p
     assert lib.read("a", as_of="snap").data == 1
 
 
-import pandas as pd
-
-
-def test_read_symbol_with_ts_in_snapshot(lmdb_version_store, sym):
-    lib = lmdb_version_store
-    lib.write(sym, 0)
-    lib.write(sym, 1)
-    time_after_second_write = pd.Timestamp.utcnow()
-    lib.snapshot("snap")
-    # After this write only version 1 exists via the snapshot
-    lib.write(sym, 2, prune_previous_version=True)
-    time_after_third_write = pd.Timestamp.utcnow()
-
-    assert lib.read(sym).data == 2
-    versions = lib.list_versions()
-    assert len(versions) == 2  # deleted for version 1
-
-    assert lib.read(sym, as_of=1).data == 1
-    assert lib.read(sym, as_of=time_after_second_write).version == 1
-    assert lib.read(sym, as_of=time_after_second_write).data == 1
-
-    lib.snapshot("snap1")
-    lib.delete_version(sym, 2)
-    assert lib.read(sym, as_of=2).data == 2  # still in snapshot
-    assert lib.read(sym, as_of=time_after_third_write).version == 2
-
-
-def test_read_symbol_with_ts_in_snapshot_with_pruning(lmdb_version_store_tombstone_and_pruning, sym):
-    lib = lmdb_version_store_tombstone_and_pruning
+@pytest.mark.parametrize(
+    "store", ["lmdb_version_store_v1", "lmdb_version_store_v2", "lmdb_version_store_tombstone_and_pruning"]
+)
+def test_read_symbol_with_ts_in_snapshot(store, request, sym):
+    lib = request.getfixturevalue(store)
     lib.write(sym, 0)
-    lib.write(sym, 1)
-    time_after_second_write = pd.Timestamp.utcnow()
+    with distinct_timestamps(lib) as second_write_timestamps:
+        lib.write(sym, 1)
     lib.snapshot("snap")
     # After this write only version 1 exists via the snapshot
-    lib.write(sym, 2, prune_previous_version=True)
-    time_after_third_write = pd.Timestamp.utcnow()
+    with distinct_timestamps(lib) as third_write_timestamps:
+        lib.write(sym, 2, prune_previous_version=True)
 
     assert lib.read(sym).data == 2
     versions = lib.list_versions()
     assert len(versions) == 2  # deleted for version 1
 
     assert lib.read(sym, as_of=1).data == 1
-    assert lib.read(sym, as_of=time_after_second_write).version == 1
-    assert lib.read(sym, as_of=time_after_second_write).data == 1
+    assert lib.read(sym, as_of=second_write_timestamps.after).version == 1
+    assert lib.read(sym, as_of=second_write_timestamps.after).data == 1
 
     lib.snapshot("snap1")
     lib.delete_version(sym, 2)
     assert lib.read(sym, as_of=2).data == 2  # still in snapshot
-    assert lib.read(sym, as_of=time_after_third_write).version == 2
+    assert lib.read(sym, as_of=third_write_timestamps.after).version == 2
 
 
 def test_snapshot_random_versions_to_fail(lmdb_version_store_tombstone_and_pruning, sym):
diff --git a/python/tests/unit/arcticdb/version_store/test_deletion.py b/python/tests/unit/arcticdb/version_store/test_deletion.py
index d17b600ac6..daa5a7a7c7 100644
--- a/python/tests/unit/arcticdb/version_store/test_deletion.py
+++ b/python/tests/unit/arcticdb/version_store/test_deletion.py
@@ -9,11 +9,11 @@
 import numpy as np
 import pandas as pd
 import pytest
-import time
 
 from arcticdb.util.test import assert_frame_equal
 from arcticdb_ext.exceptions import InternalException
 from arcticdb_ext.storage import KeyType, NoDataFoundException
+from arcticdb_ext.version_store import ManualClockVersionStore
 from arcticdb.version_store._normalization import NPDDataFrame
 from arcticdb.util.test import sample_dataframe
 
@@ -56,53 +56,40 @@ def test_delete_version_with_update(version_store_factory, pos, sym):
         assert_frame_equal(lmdb_version_store.read(symbol, 0).data, original_df)
 
 
-@pytest.mark.skip(reason="Flaky test, needs investigation.")
 def test_delete_by_timestamp(lmdb_version_store, sym):
     symbol = sym
-    time_initial = pd.Timestamp.utcnow()
-    time.sleep(0.1)
+    now = lmdb_version_store.version_store.get_store_current_timestamp_for_tests()
+    lmdb_version_store.version_store = ManualClockVersionStore(lmdb_version_store._library)
+    minute_in_ns = 60 * int(1e9)
 
+    ManualClockVersionStore.time = now - 5 * minute_in_ns
     lmdb_version_store.write(symbol, 1)  # v0
-    time_after_v0 = pd.Timestamp.utcnow()
-    time.sleep(0.1)
 
+    ManualClockVersionStore.time = now - 4 * minute_in_ns
     lmdb_version_store.write(symbol, 2)  # v1
-    time_after_v1 = pd.Timestamp.utcnow()
-    time.sleep(0.1)
 
+    ManualClockVersionStore.time = now - 3 * minute_in_ns
     lmdb_version_store.write(symbol, 3)  # v2
-    time_after_v2 = pd.Timestamp.utcnow()
-    time.sleep(0.1)
 
+    ManualClockVersionStore.time = now - 2 * minute_in_ns
     lmdb_version_store.write(symbol, 4)  # v3
-    time_after_v3 = pd.Timestamp.utcnow()
-    time.sleep(0.1)
 
+    ManualClockVersionStore.time = now - 1 * minute_in_ns
     lmdb_version_store.write(symbol, 5)  # v4
-    time_after_v4 = pd.Timestamp.utcnow()
 
-    lmdb_version_store._prune_previous_versions(
-        symbol, keep_mins=(pd.Timestamp.utcnow() - time_initial).microseconds / 60.0 / 1000000
-    )
+    lmdb_version_store._prune_previous_versions(symbol, keep_mins=5.5)
     assert len(lmdb_version_store.list_versions(symbol)) == 5
 
-    lmdb_version_store._prune_previous_versions(
-        symbol, keep_mins=(pd.Timestamp.utcnow() - time_after_v0).microseconds / 60.0 / 1000000
-    )
+    lmdb_version_store._prune_previous_versions(symbol, keep_mins=4.5)
     assert len(lmdb_version_store.list_versions(symbol)) == 4
 
-    lmdb_version_store._prune_previous_versions(
-        symbol, keep_mins=(pd.Timestamp.utcnow() - time_after_v3).microseconds / 60.0 / 1000000, keep_version=2
-    )
+    lmdb_version_store._prune_previous_versions(symbol, keep_mins=1.5, keep_version=2)
     assert len(lmdb_version_store.list_versions(symbol)) == 2
 
-    lmdb_version_store._prune_previous_versions(
-        symbol, keep_mins=(pd.Timestamp.utcnow() - time_after_v4).microseconds / 60.0 / 1000000
-    )
+    lmdb_version_store._prune_previous_versions(symbol, keep_mins=1.5)
     assert len(lmdb_version_store.list_versions(symbol)) == 1
 
 
-@pytest.mark.skip
 def test_clear_lmdb(lmdb_version_store, sym):
     symbol = sym
     lmdb_version_store.version_store.clear()