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

Issue:
Remove storage inlined headers into .cpp to fix IFNDR issue visible with MSBuild

Remove storage inlined headers into .cpp to fix IFNDR issue visible with MSBuild



Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index 2d50f04f05..da4070b18e 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -187,7 +187,6 @@ set(arcticdb_srcs
         entity/frame_and_descriptor.hpp
         entity/index_range.hpp
         entity/key.hpp
-        entity/key-inl.hpp
         entity/metrics.hpp
         entity/data_error.hpp
         entity/native_tensor.hpp
@@ -246,9 +245,7 @@ set(arcticdb_srcs
         storage/library_index.hpp
         storage/library_manager.hpp
         storage/azure/azure_storage.hpp
-        storage/azure/azure_storage-inl.hpp
         storage/lmdb/lmdb_storage.hpp
-        storage/lmdb/lmdb_storage-inl.hpp
         storage/memory/memory_storage.hpp
         storage/memory/memory_storage.cpp
         storage/mongo/mongo_client.hpp
@@ -256,11 +253,8 @@ set(arcticdb_srcs
         storage/mongo/mongo_storage.hpp
         storage/object_store_utils.hpp
         storage/s3/nfs_backed_storage.hpp
-        storage/s3/nfs_backed_storage-inl.hpp
         storage/s3/s3_client_accessor.hpp
-        storage/s3/s3_storage-inl.hpp
         storage/s3/s3_storage_tool.hpp
-        storage/s3/s3_storage_tool-inl.hpp
         storage/storage_factory.hpp
         storage/storage_options.hpp
         storage/storage.hpp
diff --git a/cpp/arcticdb/entity/key-inl.hpp b/cpp/arcticdb/entity/key-inl.hpp
deleted file mode 100644
index 8f581ed179..0000000000
--- a/cpp/arcticdb/entity/key-inl.hpp
+++ /dev/null
@@ -1,301 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * 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.
- */
-
-#ifndef ARCTICDB_KEY_H_
-#error "This should only be included by key.hpp"
-#endif
-
-#include <arcticdb/entity/types.hpp>
-#include <arcticdb/util/hash.hpp>
-#include <arcticdb/util/preconditions.hpp>
-#include <arcticdb/util/preprocess.hpp>
-#include <type_traits>
-#include <string_view>
-#include <memory>
-#include <algorithm>
-#include <variant>
-
-namespace arcticdb::entity {
-
-struct DefaultAtomKeyFormat {
-    static constexpr char format[] = "{}:{}:{:d}:0x{:x}@{:d}[{},{}]";
-};
-
-template<class T, class FormattingTag=DefaultAtomKeyFormat>
-struct FormattableRef {
-    const T &ref;
-
-    explicit FormattableRef(const T &k) : ref(k) {}
-
-    FormattableRef(const FormattableRef &) = delete;
-
-    FormattableRef(FormattableRef &&) = delete;
-
-};
-
-template<class T, class FormattingTag = DefaultAtomKeyFormat>
-auto formattable(const T &ref) {
-    return FormattableRef<T, FormattingTag>(ref);
-}
-
-using ContentHash = std::uint64_t;
-
- enum class KeyClass : int {
-     /*
-      * ATOM_KEY is a KeyType containing a segment with either data or other keys in it and the client needs to
-      * have the entire key in memory with all fields populated to read it out.
-      */
-     ATOM_KEY,
-     /*
-      * REF_KEY is the KeyType containing other keys in it, and can be easily read by the client using just the
-      * id() field in the key as it is unique for a given (id, KeyType) combination
-      */
-     REF_KEY,
-     UNKNOWN_CLASS
- };
-
-enum class KeyType : int {
-    /*
-     * TODO: For the Kestrel streaming stuff, not in general use atm.
-     */
-    STREAM_GROUP = 0,
-    /*
-     * TODO: For Kestrel streaming stuff, not in general use atm.
-     */
-    GENERATION = 1,
-    // The following keys are string-id-based
-    /*
-     * TABLE_DATA are the leaf nodes of our tree and contain the actual data in the segment.
-     */
-    TABLE_DATA = 2,
-    /*
-     * TABLE_INDEX contains multiple TABLE_DATA keys in the segment of the key, and sits between the VERSION
-     * and TABLE_DATA keys. *_REF, MULTI_KEY and VERSION keys point to the index key type.
-     */
-    TABLE_INDEX = 3,
-    /*
-     * At the moment, VERSION keys point to a single TABLE_INDEX key. In the segment of the key, it can contain
-     * other VERSION keys (for faster traversal of versions, via a linked list on storage) , TABLE_INDEX key (just
-     * one atm, but no reason why we can't have multiple) and TOMBSTONE Keys
-     */
-    VERSION = 4,
-    /*
-     * VERSION_JOURNAL is not used anymore, just exist for backwards compatibility reasons
-     */
-    VERSION_JOURNAL = 5,
-    /*
-     * METRICS is not longer used, was initially used to log timing and other performance metrics.
-     */
-    METRICS = 6,
-    /*
-     * Is not used anymore, was used before SNAPSHOT_REF was introduced and might still be present in older libraries.
-     * It points to TABLE_INDEX keys in the segment
-     */
-    SNAPSHOT = 7,
-    /*
-     * SYMBOL_LIST is used for creating a symbol linked list in storage to avoid iterating all symbols in operations
-     * like list symbols.
-     */
-    SYMBOL_LIST = 8,
-    /*
-     * VERSION_REF sits at the top of the tree and points to the latest index and version for each stream_id,
-     * this helps in speeding up latest_version operations and by pointing to the VERSION key type it forms the head
-     * of the linked list in storage
-     */
-    VERSION_REF = 9,
-    /*
-     * Used for storing storage informations (like last sync time)
-     */
-    STORAGE_INFO = 10,
-    /*
-     * The head of the list of incomplete appended segments
-     */
-    APPEND_REF = 11,
-    /*
-     * Is used as a index of indexes, specially for cases with recursive normalizers where a single write
-     * might lead to multiple reads, and on a read being issued by the client we want a faster way to reconstruct
-     * the original structure
-     */
-    MULTI_KEY = 12,
-    /*
-     * When present for a symbol, this helps us create a locking mechanism in the storage where the presence of
-     * this key will block other concurrent write operations from running at the same time.
-     */
-    LOCK = 13,
-    /*
-     * Is used to reference a single snapshot, and points to multiple index keys. replaces the old SNAPSHOT key type
-     */
-    SNAPSHOT_REF = 14,
-    /*
-     * TOMBSTONE KeyType WILL NEVER EXIST in the storage as a standalone key , and only lives inside the segment of
-     * VERSION KeyType to indicate a version that was pruned or explicitly deleted
-     */
-    TOMBSTONE = 15,
-    /*
-     * An incomplete appended segment
-     */
-    APPEND_DATA = 16,
-    /*
-     * Used for logging write, tombstone, and tombstone all events (to be used by high-priority replication job)
-     * See also LOG_COMPACTED
-     */
-    LOG = 17,
-    /*
-     * This is used to store references to partition values for a symbol, eg. for values a, b, c for a
-     * symbol 'sym' we will have a partition key per write for that symbol with subkeys for the dataframes
-     * containing individual values.
-     */
-    PARTITION = 18,
-    /*
-     * Used for storing offsets of input streams
-     */
-    OFFSET = 19,
-    /*
-     * Temporary - remove_me
-     */
-    BACKUP_SNAPSHOT_REF = 20,
-    /*
-     * Used for indicating that all keys lower than this version id are deleted
-     */
-    TOMBSTONE_ALL = 21,
-    /*
-     * Used for library configs
-     */
-    LIBRARY_CONFIG = 22,
-    /*
-     * Marks a SNAPSHOT_REF scheduled for removal. Unlike the TOMBSTONE[_ALL] keys for versioning, this is written to
-     * storage as a root key as opposed being stored in the segment. The segment contents of this key is the same as the
-     * SNAPSHOT_REF removed.
-     */
-    SNAPSHOT_TOMBSTONE = 23,
-    /*
-     * Contains multiple LOG keys in its segment (to be used by low-priority replication job)
-     */
-    LOG_COMPACTED = 24,
-    /*
-     * Contains column stats about the index key with the same stream ID and version number
-     */
-    COLUMN_STATS = 25,
-    UNDEFINED
-};
-
-inline std::vector<KeyType> key_types_write_precedence() {
-    // TOMBSTONE[_ALL] keys are not included because they're not written to the storage,
-    // they just exist inside version keys
-    return {
-        KeyType::LIBRARY_CONFIG,
-        KeyType::TABLE_DATA,
-        KeyType::TABLE_INDEX,
-        KeyType::MULTI_KEY,
-        KeyType::VERSION,
-        KeyType::VERSION_JOURNAL,
-        KeyType::VERSION_REF,
-        KeyType::SYMBOL_LIST,
-        KeyType::SNAPSHOT,
-        KeyType::SNAPSHOT_REF,
-        KeyType::SNAPSHOT_TOMBSTONE,
-        KeyType::APPEND_REF,
-        KeyType::APPEND_DATA,
-        KeyType::PARTITION,
-        KeyType::OFFSET
-    };
-}
-
-inline std::vector<KeyType> key_types_read_precedence() {
-    auto output = key_types_write_precedence();
-    std::reverse(std::begin(output), std::end(output));
-    return output;
-}
-
- } //namespace arcticdb::entity
-
-namespace std {
-    template <> struct hash<arcticdb::entity::KeyType >
-    {
-        size_t operator()(const arcticdb::entity::KeyType & kt) const {
-            return std::hash<int>{}(static_cast<int>(kt));
-        }
-    };
-}
-
-namespace arcticdb::entity {
-
-// N.B. some storages are case-insensitive, so these need to be
-// prefix-unique disregarding case
-const char* key_type_long_name(KeyType key_type);
-
-char key_type_short_name(KeyType key_type);
-
-enum class VariantType : char {
-    STRING_TYPE = 's',
-    NUMERIC_TYPE = 'd',
-    UNKNOWN_TYPE = 'u'
-};
-
-VariantType variant_type_from_key_type(KeyType key_type);
-
-inline bool is_index_key_type(KeyType key_type) {
-    // TODO: Change name probably.
-    return (key_type == KeyType::TABLE_INDEX) || (key_type == KeyType::MULTI_KEY);
-}
-
-bool is_string_key_type(KeyType k);
-
-bool is_ref_key_class(KeyType k);
-
-inline KeyType get_key_type_for_data_stream(const StreamId &) {
-    return KeyType::TABLE_DATA;
-}
-
-inline KeyType get_key_type_for_index_stream(const StreamId &) {
-    return KeyType::TABLE_INDEX;
-}
-
-template <typename Function>
-auto foreach_key_type(Function&& func) {
-    for(int k = 0; k < int(KeyType::UNDEFINED); ++k) {
-        func(KeyType(k));
-    }
-}
-
-template <typename Function>
-auto foreach_key_type_read_precedence(Function&& func) {
-    auto types = key_types_read_precedence();
-    for(auto type : types) {
-        func(KeyType(type));
-    }
-}
-
-template <typename Function>
-auto foreach_key_type_write_precedence(Function&& func) {
-    auto types = key_types_write_precedence();
-    for(auto type : types) {
-        func(KeyType(type));
-    }
-}
-
-
-} // namespace arcticdb::entity
-
-namespace fmt {
-
-using namespace arcticdb::entity;
-
-template<>
-struct formatter<KeyType> {
-
-    template<typename ParseContext>
-    constexpr auto parse(ParseContext &ctx) { return ctx.begin(); }
-
-    template<typename FormatContext>
-    auto format(const KeyType k, FormatContext &ctx) const {
-        return  fmt::format_to(ctx.out(), "{}", key_type_short_name(k));
-    }
-};
-
-}
-
diff --git a/cpp/arcticdb/entity/key.hpp b/cpp/arcticdb/entity/key.hpp
index 0be445ca99..043d32756f 100644
--- a/cpp/arcticdb/entity/key.hpp
+++ b/cpp/arcticdb/entity/key.hpp
@@ -8,7 +8,289 @@
 #pragma once
 
 #include <arcticdb/entity/types.hpp>
+#include <arcticdb/util/hash.hpp>
+#include <arcticdb/util/preconditions.hpp>
+#include <arcticdb/util/preprocess.hpp>
+#include <type_traits>
+#include <string_view>
+#include <memory>
+#include <algorithm>
+#include <variant>
+
+namespace arcticdb::entity {
+
+struct DefaultAtomKeyFormat {
+    static constexpr char format[] = "{}:{}:{:d}:0x{:x}@{:d}[{},{}]";
+};
+
+template<class T, class FormattingTag=DefaultAtomKeyFormat>
+struct FormattableRef {
+    const T &ref;
+
+    explicit FormattableRef(const T &k) : ref(k) {}
+
+    FormattableRef(const FormattableRef &) = delete;
+
+    FormattableRef(FormattableRef &&) = delete;
+
+};
+
+template<class T, class FormattingTag = DefaultAtomKeyFormat>
+auto formattable(const T &ref) {
+    return FormattableRef<T, FormattingTag>(ref);
+}
+
+using ContentHash = std::uint64_t;
+
+ enum class KeyClass : int {
+     /*
+      * ATOM_KEY is a KeyType containing a segment with either data or other keys in it and the client needs to
+      * have the entire key in memory with all fields populated to read it out.
+      */
+     ATOM_KEY,
+     /*
+      * REF_KEY is the KeyType containing other keys in it, and can be easily read by the client using just the
+      * id() field in the key as it is unique for a given (id, KeyType) combination
+      */
+     REF_KEY,
+     UNKNOWN_CLASS
+ };
+
+enum class KeyType : int {
+    /*
+     * TODO: For the Kestrel streaming stuff, not in general use atm.
+     */
+    STREAM_GROUP = 0,
+    /*
+     * TODO: For Kestrel streaming stuff, not in general use atm.
+     */
+    GENERATION = 1,
+    // The following keys are string-id-based
+    /*
+     * TABLE_DATA are the leaf nodes of our tree and contain the actual data in the segment.
+     */
+    TABLE_DATA = 2,
+    /*
+     * TABLE_INDEX contains multiple TABLE_DATA keys in the segment of the key, and sits between the VERSION
+     * and TABLE_DATA keys. *_REF, MULTI_KEY and VERSION keys point to the index key type.
+     */
+    TABLE_INDEX = 3,
+    /*
+     * At the moment, VERSION keys point to a single TABLE_INDEX key. In the segment of the key, it can contain
+     * other VERSION keys (for faster traversal of versions, via a linked list on storage) , TABLE_INDEX key (just
+     * one atm, but no reason why we can't have multiple) and TOMBSTONE Keys
+     */
+    VERSION = 4,
+    /*
+     * VERSION_JOURNAL is not used anymore, just exist for backwards compatibility reasons
+     */
+    VERSION_JOURNAL = 5,
+    /*
+     * METRICS is not longer used, was initially used to log timing and other performance metrics.
+     */
+    METRICS = 6,
+    /*
+     * Is not used anymore, was used before SNAPSHOT_REF was introduced and might still be present in older libraries.
+     * It points to TABLE_INDEX keys in the segment
+     */
+    SNAPSHOT = 7,
+    /*
+     * SYMBOL_LIST is used for creating a symbol linked list in storage to avoid iterating all symbols in operations
+     * like list symbols.
+     */
+    SYMBOL_LIST = 8,
+    /*
+     * VERSION_REF sits at the top of the tree and points to the latest index and version for each stream_id,
+     * this helps in speeding up latest_version operations and by pointing to the VERSION key type it forms the head
+     * of the linked list in storage
+     */
+    VERSION_REF = 9,
+    /*
+     * Used for storing storage informations (like last sync time)
+     */
+    STORAGE_INFO = 10,
+    /*
+     * The head of the list of incomplete appended segments
+     */
+    APPEND_REF = 11,
+    /*
+     * Is used as a index of indexes, specially for cases with recursive normalizers where a single write
+     * might lead to multiple reads, and on a read being issued by the client we want a faster way to reconstruct
+     * the original structure
+     */
+    MULTI_KEY = 12,
+    /*
+     * When present for a symbol, this helps us create a locking mechanism in the storage where the presence of
+     * this key will block other concurrent write operations from running at the same time.
+     */
+    LOCK = 13,
+    /*
+     * Is used to reference a single snapshot, and points to multiple index keys. replaces the old SNAPSHOT key type
+     */
+    SNAPSHOT_REF = 14,
+    /*
+     * TOMBSTONE KeyType WILL NEVER EXIST in the storage as a standalone key , and only lives inside the segment of
+     * VERSION KeyType to indicate a version that was pruned or explicitly deleted
+     */
+    TOMBSTONE = 15,
+    /*
+     * An incomplete appended segment
+     */
+    APPEND_DATA = 16,
+    /*
+     * Used for logging write, tombstone, and tombstone all events (to be used by high-priority replication job)
+     * See also LOG_COMPACTED
+     */
+    LOG = 17,
+    /*
+     * This is used to store references to partition values for a symbol, eg. for values a, b, c for a
+     * symbol 'sym' we will have a partition key per write for that symbol with subkeys for the dataframes
+     * containing individual values.
+     */
+    PARTITION = 18,
+    /*
+     * Used for storing offsets of input streams
+     */
+    OFFSET = 19,
+    /*
+     * Temporary - remove_me
+     */
+    BACKUP_SNAPSHOT_REF = 20,
+    /*
+     * Used for indicating that all keys lower than this version id are deleted
+     */
+    TOMBSTONE_ALL = 21,
+    /*
+     * Used for library configs
+     */
+    LIBRARY_CONFIG = 22,
+    /*
+     * Marks a SNAPSHOT_REF scheduled for removal. Unlike the TOMBSTONE[_ALL] keys for versioning, this is written to
+     * storage as a root key as opposed being stored in the segment. The segment contents of this key is the same as the
+     * SNAPSHOT_REF removed.
+     */
+    SNAPSHOT_TOMBSTONE = 23,
+    /*
+     * Contains multiple LOG keys in its segment (to be used by low-priority replication job)
+     */
+    LOG_COMPACTED = 24,
+    /*
+     * Contains column stats about the index key with the same stream ID and version number
+     */
+    COLUMN_STATS = 25,
+    UNDEFINED
+};
+
+inline std::vector<KeyType> key_types_write_precedence() {
+    // TOMBSTONE[_ALL] keys are not included because they're not written to the storage,
+    // they just exist inside version keys
+    return {
+        KeyType::LIBRARY_CONFIG,
+        KeyType::TABLE_DATA,
+        KeyType::TABLE_INDEX,
+        KeyType::MULTI_KEY,
+        KeyType::VERSION,
+        KeyType::VERSION_JOURNAL,
+        KeyType::VERSION_REF,
+        KeyType::SYMBOL_LIST,
+        KeyType::SNAPSHOT,
+        KeyType::SNAPSHOT_REF,
+        KeyType::SNAPSHOT_TOMBSTONE,
+        KeyType::APPEND_REF,
+        KeyType::APPEND_DATA,
+        KeyType::PARTITION,
+        KeyType::OFFSET
+    };
+}
+
+inline std::vector<KeyType> key_types_read_precedence() {
+    auto output = key_types_write_precedence();
+    std::reverse(std::begin(output), std::end(output));
+    return output;
+}
+
+} //namespace arcticdb::entity
+
+namespace std {
+    template <> struct hash<arcticdb::entity::KeyType >
+    {
+        size_t operator()(const arcticdb::entity::KeyType & kt) const {
+            return std::hash<int>{}(static_cast<int>(kt));
+        }
+    };
+}
+
+namespace arcticdb::entity {
+
+// N.B. some storages are case-insensitive, so these need to be
+// prefix-unique disregarding case
+const char* key_type_long_name(KeyType key_type);
+
+char key_type_short_name(KeyType key_type);
+
+enum class VariantType : char {
+    STRING_TYPE = 's',
+    NUMERIC_TYPE = 'd',
+    UNKNOWN_TYPE = 'u'
+};
+
+VariantType variant_type_from_key_type(KeyType key_type);
+
+inline bool is_index_key_type(KeyType key_type) {
+    // TODO: Change name probably.
+    return (key_type == KeyType::TABLE_INDEX) || (key_type == KeyType::MULTI_KEY);
+}
+
+bool is_string_key_type(KeyType k);
+
+bool is_ref_key_class(KeyType k);
+
+inline KeyType get_key_type_for_data_stream(const StreamId &) {
+    return KeyType::TABLE_DATA;
+}
+
+inline KeyType get_key_type_for_index_stream(const StreamId &) {
+    return KeyType::TABLE_INDEX;
+}
+
+template <typename Function>
+auto foreach_key_type(Function&& func) {
+    for(int k = 0; k < int(KeyType::UNDEFINED); ++k) {
+        func(KeyType(k));
+    }
+}
+
+template <typename Function>
+auto foreach_key_type_read_precedence(Function&& func) {
+    auto types = key_types_read_precedence();
+    for(auto type : types) {
+        func(KeyType(type));
+    }
+}
+
+template <typename Function>
+auto foreach_key_type_write_precedence(Function&& func) {
+    auto types = key_types_write_precedence();
+    for(auto type : types) {
+        func(KeyType(type));
+    }
+}
+
+
+} // namespace arcticdb::entity
+
+
+template<>
+struct fmt::formatter<arcticdb::entity::KeyType> {
+
+    template<typename ParseContext>
+    constexpr auto parse(ParseContext &ctx) { return ctx.begin(); }
+
+    template<typename FormatContext>
+    auto format(const arcticdb::entity::KeyType k, FormatContext &ctx) const {
+        using namespace arcticdb::entity;
+        return  fmt::format_to(ctx.out(), "{}", key_type_short_name(k));
+    }
+};
+
 
-#define ARCTICDB_KEY_H_
-#include <arcticdb/entity/key-inl.hpp>
-#undef ARCTICDB_KEY_H_
diff --git a/cpp/arcticdb/storage/azure/azure_storage-inl.hpp b/cpp/arcticdb/storage/azure/azure_storage-inl.hpp
deleted file mode 100644
index 724898a942..0000000000
--- a/cpp/arcticdb/storage/azure/azure_storage-inl.hpp
+++ /dev/null
@@ -1,326 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * 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.
- */
-
-#ifndef ARCTICDB_AZURE_STORAGE_H_
-#error "This should only be included by file_storage.cpp"
-#endif
-
-#include <arcticdb/util/preconditions.hpp>
-#include <arcticdb/util/pb_util.hpp>
-#include <arcticdb/log/log.hpp>
-#include <arcticdb/util/buffer_pool.hpp>
-
-#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
-#include <folly/gen/Base.h>
-#include <arcticdb/storage/azure/azure_storage.hpp>
-#include <arcticdb/storage/storage_utils.hpp>
-#include <arcticdb/storage/object_store_utils.hpp>
-#include <arcticdb/storage/storage_options.hpp>
-#include <arcticdb/entity/serialized_key.hpp>
-#include <arcticdb/util/exponential_backoff.hpp>
-#include <arcticdb/util/configs_map.hpp>
-#include <arcticdb/util/composite.hpp>
-
-#include <azure/core.hpp>
-#include <azure/storage/blobs.hpp>
-
-#include <boost/interprocess/streams/bufferstream.hpp>
-#include <folly/ThreadLocal.h>
-
-#undef GetMessage
-
-namespace arcticdb::storage {
-
-using namespace object_store_utils;
-
-namespace azure {
-
-namespace fg = folly::gen;
-
-namespace detail {
-
-static const size_t BATCH_SUBREQUEST_LIMIT = 256; //https://github.com/Azure/azure-sdk-for-python/blob/767facc39f2487504bcde4e627db16a79f96b297/sdk/storage/azure-storage-blob/azure/storage/blob/_container_client.py#L1608
-
-inline Azure::Core::Context get_context(unsigned int request_timeout){
-    Azure::Core::Context requestContext; //TODO: Maybe can be static but need to be careful with its shared_ptr and ContextSharedState
-    return requestContext.WithDeadline(std::chrono::system_clock::now() + std::chrono::milliseconds(request_timeout));
-}
-
-template<class KeyBucketizer>
-void do_write_impl(
-    Composite<KeySegmentPair>&& kvs,
-    const std::string& root_folder,
-    Azure::Storage::Blobs::BlobContainerClient& container_client,
-    KeyBucketizer&& bucketizer,
-    const Azure::Storage::Blobs::UploadBlockBlobFromOptions& upload_option,
-    unsigned int request_timeout) {
-        ARCTICDB_SAMPLE(AzureStorageWrite, 0)
-        auto fmt_db = [](auto&& kv) { return kv.key_type(); };
-
-        (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach(
-            [&container_client, &root_folder, b=std::move(bucketizer), &upload_option, &request_timeout] (auto&& group) {
-                auto key_type_dir = key_type_folder(root_folder, group.key());
-                ARCTICDB_TRACE(log::storage(), "Azure key_type_folder is {}", key_type_dir);
-
-                ARCTICDB_SUBSAMPLE(AzureStorageWriteValues, 0)
-                for (auto& kv : group.values()) {
-                    auto& k = kv.variant_key();
-                    auto blob_name = object_path(b.bucketize(key_type_dir, k), k);
-                    auto& seg = kv.segment();
-
-                    std::shared_ptr<Buffer> tmp;
-                    auto hdr_size = seg.segment_header_bytes_size();
-                    auto [dst, write_size] = seg.try_internal_write(tmp, hdr_size);
-                    util::check(arcticdb::Segment::FIXED_HEADER_SIZE + hdr_size + seg.buffer().bytes() <= write_size,
-                                    "Size disparity, fixed header size {} + variable header size {} + buffer size {}  >= total size {}",
-                                    arcticdb::Segment::FIXED_HEADER_SIZE,
-                                    hdr_size,
-                                    seg.buffer().bytes(),
-                                    write_size);
-
-                    ARCTICDB_SUBSAMPLE(AzureStorageUploadObject, 0)
-                    try{
-                        auto blob_client = container_client.GetBlockBlobClient(blob_name);
-                        ARCTICDB_RUNTIME_DEBUG(log::storage(), "Writing key {}: {}, with {} bytes of data",
-                                               variant_key_type(k),
-                                               variant_key_view(k),
-                                               seg.total_segment_size(hdr_size));
-                        blob_client.UploadFrom(dst, write_size, upload_option, get_context(request_timeout));
-                    }
-                    catch (const Azure::Core::RequestFailedException& e){
-                        util::raise_rte("Failed to write azure with key '{}' {} {}: {}",
-                                            k,
-                                            blob_name,
-                                            static_cast<int>(e.StatusCode),
-                                            e.ReasonPhrase);
-                    }
-
-                    ARCTICDB_RUNTIME_DEBUG(log::storage(), "Wrote key {}: {}, with {} bytes of data",
-                                        variant_key_type(k),
-                                        variant_key_view(k),
-                                        seg.total_segment_size(hdr_size));
-                }
-            }
-        );
-}
-
-template<class KeyBucketizer>
-void do_update_impl(
-    Composite<KeySegmentPair>&& kvs,
-    const std::string& root_folder,
-    Azure::Storage::Blobs::BlobContainerClient& container_client,
-    KeyBucketizer&& bucketizer,
-    const Azure::Storage::Blobs::UploadBlockBlobFromOptions& upload_option,
-    unsigned int request_timeout) {
-        // azure updates the key if it already exists
-        do_write_impl(std::move(kvs), root_folder, container_client, std::move(bucketizer), upload_option, request_timeout);
-}
-
-struct UnexpectedAzureErrorException : public std::exception {};
-
-template<class KeyBucketizer>
-void do_read_impl(Composite<VariantKey> && ks,
-    const ReadVisitor& visitor,
-    const std::string& root_folder,
-    Azure::Storage::Blobs::BlobContainerClient& container_client,
-    KeyBucketizer&& bucketizer,
-    ReadKeyOpts opts,
-    const Azure::Storage::Blobs::DownloadBlobToOptions& download_option,
-    unsigned int request_timeout) {
-        ARCTICDB_SAMPLE(AzureStorageRead, 0)
-        auto fmt_db = [](auto&& k) { return variant_key_type(k); };
-        std::vector<VariantKey> failed_reads;
-
-        (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach(
-            [&container_client, &root_folder, b=std::move(bucketizer), &visitor, &failed_reads,
-            opts=opts, &download_option, &request_timeout] (auto&& group) {
-            for (auto& k : group.values()) {
-                auto key_type_dir = key_type_folder(root_folder, variant_key_type(k));
-                auto blob_name = object_path(b.bucketize(key_type_dir, k), k);
-                try{
-                    ARCTICDB_DEBUG(log::storage(), "Reading key {}: {}", variant_key_type(k), variant_key_view(k));
-
-                    auto blob_client = container_client.GetBlockBlobClient(blob_name);
-                    auto properties = blob_client.GetProperties(Azure::Storage::Blobs::GetBlobPropertiesOptions{}, get_context(request_timeout)).Value;
-                    std::shared_ptr<Buffer> buffer = std::make_shared<Buffer>(properties.BlobSize);
-                    blob_client.DownloadTo(buffer->preamble(), buffer->available(), download_option, get_context(request_timeout));
-                    ARCTICDB_SUBSAMPLE(AzureStorageVisitSegment, 0)
-                    visitor(k, Segment::from_buffer(std::move(buffer)));
-
-                    ARCTICDB_DEBUG(log::storage(), "Read key {}: {}", variant_key_type(k), variant_key_view(k));
-                }
-                catch (const Azure::Core::RequestFailedException& e){
-                    if (!opts.dont_warn_about_missing_key) {
-                        log::storage().warn("Failed to read azure segment with key '{}' {} {}: {}",
-                                                k,
-                                                blob_name,
-                                                static_cast<int>(e.StatusCode),
-                                                e.ReasonPhrase);
-                    }
-                    failed_reads.push_back(k);
-                }
-            }
-        });
-        if(!failed_reads.empty())
-            throw KeyNotFoundException(Composite<VariantKey>{std::move(failed_reads)});
-}
-
-template<class KeyBucketizer>
-void do_remove_impl(Composite<VariantKey>&& ks,
-    const std::string& root_folder,
-    Azure::Storage::Blobs::BlobContainerClient& container_client,
-    KeyBucketizer&& bucketizer,
-    unsigned int request_timeout) {
-        ARCTICDB_SUBSAMPLE(AzureStorageDeleteBatch, 0)
-        auto fmt_db = [](auto&& k) { return variant_key_type(k); };
-        static const size_t delete_object_limit =
-            std::min(BATCH_SUBREQUEST_LIMIT, static_cast<size_t>(ConfigsMap::instance()->get_int("AzureStorage.DeleteBatchSize", BATCH_SUBREQUEST_LIMIT)));
-        
-        auto batch = container_client.CreateBatch();
-        unsigned int batch_counter = 0u; 
-        auto submit_batch = [&container_client, &request_timeout, &batch_counter](auto &batch) {
-            ARCTICDB_SUBSAMPLE(AzureStorageDeleteObjects, 0)
-            try{
-                container_client.SubmitBatch(batch, Azure::Storage::Blobs::SubmitBlobBatchOptions(), get_context(request_timeout));//To align with s3 behaviour, deleting non-exist objects is not an error, so not handling response
-            }
-            catch (const Azure::Core::RequestFailedException& e){
-                util::raise_rte("Failed to create azure segment batch remove request {}: {}",
-                                    static_cast<int>(e.StatusCode),
-                                    e.ReasonPhrase);
-            }
-            batch = container_client.CreateBatch();
-            batch_counter = 0u;
-        };
-        
-        (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach(
-            [&root_folder, b=std::move(bucketizer), &batch, delete_object_limit=delete_object_limit, &batch_counter, &submit_batch] (auto&& group) {//bypass incorrect 'set but no used" error for delete_object_limit
-                auto key_type_dir = key_type_folder(root_folder, group.key());
-                for (auto k : folly::enumerate(group.values())) {
-                    auto blob_name = object_path(b.bucketize(key_type_dir, *k), *k);
-                    ARCTICDB_RUNTIME_DEBUG(log::storage(), "Removing azure object with key {}", blob_name);
-                    batch.DeleteBlob(blob_name);
-                    if (++batch_counter == delete_object_limit) {
-                        ARCTICDB_RUNTIME_DEBUG(log::storage(), "Submitting DeleteBlob batch");
-                        submit_batch(batch);
-                        ARCTICDB_RUNTIME_DEBUG(log::storage(), "Submitted DeleteBlob batch");
-                    }
-                }
-            }
-        );
-        if (batch_counter) {
-            ARCTICDB_RUNTIME_DEBUG(log::storage(), "Submitting DeleteBlob batch");
-            submit_batch(batch);
-            ARCTICDB_RUNTIME_DEBUG(log::storage(), "Submitted DeleteBlob batch");
-        }
-}
-
-inline auto default_prefix_handler() {
-    return [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor& key_descriptor, KeyType) {
-        return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
-    };
-}
-
-template<class KeyBucketizer, class PrefixHandler>
-void do_iterate_type_impl(KeyType key_type,
-    const IterateTypeVisitor& visitor,
-    const std::string& root_folder,
-    Azure::Storage::Blobs::BlobContainerClient& container_client,
-    KeyBucketizer&& bucketizer,
-    PrefixHandler&& prefix_handler = default_prefix_handler(),
-    const std::string& prefix = std::string{}) {
-        ARCTICDB_SAMPLE(AzureStorageIterateType, 0)
-        auto key_type_dir = key_type_folder(root_folder, key_type);
-
-        KeyDescriptor key_descriptor(prefix,
-            is_ref_key_class(key_type) ? IndexDescriptor::UNKNOWN : IndexDescriptor::TIMESTAMP, FormatType::TOKENIZED);
-        auto key_prefix = prefix_handler(prefix, key_type_dir, key_descriptor, key_type);
-        ARCTICDB_RUNTIME_DEBUG(log::storage(), "Searching for objects with prefix {}", key_prefix);
-        const auto root_folder_size = key_type_dir.size() + 1 + bucketizer.bucketize_length(key_type);
-
-        Azure::Storage::Blobs::ListBlobsOptions options;
-        options.Prefix = key_prefix;
-        try{
-            for (auto page = container_client.ListBlobs(options); page.HasPage(); page.MoveToNextPage()) {
-                for (const auto& blob : page.Blobs) {
-                    auto key = blob.Name.substr(root_folder_size);
-                    ARCTICDB_TRACE(log::version(), "Got object_list: {}, key: {}", blob.Name, key);
-                    auto k = variant_key_from_bytes(
-                                reinterpret_cast<uint8_t*>(key.data()),
-                                key.size(),
-                                key_type);
-                    ARCTICDB_DEBUG(log::storage(), "Iterating key {}: {}", variant_key_type(k), variant_key_view(k));
-                    ARCTICDB_SUBSAMPLE(AzureStorageVisitKey, 0)
-                    visitor(std::move(k));
-                    ARCTICDB_SUBSAMPLE(AzureStorageCursorNext, 0)
-                }
-            }
-        }
-        catch (const Azure::Core::RequestFailedException& e){
-            log::storage().warn("Failed to iterate azure blobs '{}' {}: {}",
-                                key_type,
-                                static_cast<int>(e.StatusCode),
-                                e.ReasonPhrase);
-        }
-}
-
-template<class KeyBucketizer>
-bool do_key_exists_impl(
-    const VariantKey& key,
-    const std::string& root_folder,
-    Azure::Storage::Blobs::BlobContainerClient& container_client,
-    KeyBucketizer&& bucketizer) {
-        auto key_type_dir = key_type_folder(root_folder, variant_key_type(key));
-        auto blob_name = object_path(bucketizer.bucketize(key_type_dir, key), key);
-        try{
-            auto blob_client = container_client.GetBlockBlobClient(blob_name);
-            auto properties = blob_client.GetProperties().Value;
-
-            if (properties.ETag.HasValue())
-                return true;
-        }
-        catch (const Azure::Core::RequestFailedException& e){
-            log::storage().debug("Failed to check azure key '{}' {} {}: {}",
-                                key,
-                                blob_name,
-                                static_cast<int>(e.StatusCode),
-                                e.ReasonPhrase);
-        }
-        return false;
-}
-} //namespace detail
-
-inline void AzureStorage::do_write(Composite<KeySegmentPair>&& kvs) {
-    detail::do_write_impl(std::move(kvs), root_folder_, container_client_, FlatBucketizer{}, upload_option_, request_timeout_);
-}
-
-inline void AzureStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts) {
-    detail::do_update_impl(std::move(kvs), root_folder_, container_client_, FlatBucketizer{}, upload_option_, request_timeout_);
-}
-
-inline void AzureStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts opts) {
-    detail::do_read_impl(std::move(ks), visitor, root_folder_, container_client_, FlatBucketizer{}, opts, download_option_, request_timeout_);
-}
-
-inline void AzureStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
-    detail::do_remove_impl(std::move(ks), root_folder_, container_client_, FlatBucketizer{}, request_timeout_);
-}
-
-inline void AzureStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string &prefix) {
-    auto prefix_handler = [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor key_descriptor, KeyType) {
-        return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
-    };
-
-    detail::do_iterate_type_impl(key_type, std::move(visitor), root_folder_, container_client_, FlatBucketizer{}, std::move(prefix_handler), prefix);
-}
-
-inline bool AzureStorage::do_key_exists(const VariantKey& key) {
-    return detail::do_key_exists_impl(key, root_folder_, container_client_, FlatBucketizer{});
-}
-
-} // namespace azure
-
-} // namespace arcticdb::storage
diff --git a/cpp/arcticdb/storage/azure/azure_storage.cpp b/cpp/arcticdb/storage/azure/azure_storage.cpp
index daf2dfa175..c2e958ddae 100644
--- a/cpp/arcticdb/storage/azure/azure_storage.cpp
+++ b/cpp/arcticdb/storage/azure/azure_storage.cpp
@@ -5,13 +5,328 @@
  * 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.
  */
 
-#define ARCTICDB_AZURE_STORAGE_H_
-#include <arcticdb/storage/azure/azure_storage-inl.hpp>
-
+#include <arcticdb/storage/azure/azure_storage.hpp>
 
 #include <arcticdb/log/log.hpp>
 #include <azure/core/http/curl_transport.hpp>
 
+
+#include <arcticdb/util/preconditions.hpp>
+#include <arcticdb/util/pb_util.hpp>
+#include <arcticdb/log/log.hpp>
+#include <arcticdb/util/buffer_pool.hpp>
+
+#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
+#include <folly/gen/Base.h>
+#include <arcticdb/storage/storage_utils.hpp>
+#include <arcticdb/storage/object_store_utils.hpp>
+#include <arcticdb/storage/storage_options.hpp>
+#include <arcticdb/entity/serialized_key.hpp>
+#include <arcticdb/util/exponential_backoff.hpp>
+#include <arcticdb/util/configs_map.hpp>
+#include <arcticdb/util/composite.hpp>
+
+#include <azure/core.hpp>
+#include <azure/storage/blobs.hpp>
+
+#include <boost/interprocess/streams/bufferstream.hpp>
+#include <folly/ThreadLocal.h>
+
+#undef GetMessage
+
+namespace arcticdb::storage {
+
+using namespace object_store_utils;
+
+namespace azure {
+
+namespace fg = folly::gen;
+
+namespace detail {
+
+static const size_t BATCH_SUBREQUEST_LIMIT = 256; //https://github.com/Azure/azure-sdk-for-python/blob/767facc39f2487504bcde4e627db16a79f96b297/sdk/storage/azure-storage-blob/azure/storage/blob/_container_client.py#L1608
+
+Azure::Core::Context get_context(unsigned int request_timeout){
+    Azure::Core::Context requestContext; //TODO: Maybe can be static but need to be careful with its shared_ptr and ContextSharedState
+    return requestContext.WithDeadline(std::chrono::system_clock::now() + std::chrono::milliseconds(request_timeout));
+}
+
+template<class KeyBucketizer>
+void do_write_impl(
+    Composite<KeySegmentPair>&& kvs,
+    const std::string& root_folder,
+    Azure::Storage::Blobs::BlobContainerClient& container_client,
+    KeyBucketizer&& bucketizer,
+    const Azure::Storage::Blobs::UploadBlockBlobFromOptions& upload_option,
+    unsigned int request_timeout) {
+        ARCTICDB_SAMPLE(AzureStorageWrite, 0)
+        auto fmt_db = [](auto&& kv) { return kv.key_type(); };
+
+        (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach(
+            [&container_client, &root_folder, b=std::move(bucketizer), &upload_option, &request_timeout] (auto&& group) {
+                auto key_type_dir = key_type_folder(root_folder, group.key());
+                ARCTICDB_TRACE(log::storage(), "Azure key_type_folder is {}", key_type_dir);
+
+                ARCTICDB_SUBSAMPLE(AzureStorageWriteValues, 0)
+                for (auto& kv : group.values()) {
+                    auto& k = kv.variant_key();
+                    auto blob_name = object_path(b.bucketize(key_type_dir, k), k);
+                    auto& seg = kv.segment();
+
+                    std::shared_ptr<Buffer> tmp;
+                    auto hdr_size = seg.segment_header_bytes_size();
+                    auto [dst, write_size] = seg.try_internal_write(tmp, hdr_size);
+                    util::check(arcticdb::Segment::FIXED_HEADER_SIZE + hdr_size + seg.buffer().bytes() <= write_size,
+                                    "Size disparity, fixed header size {} + variable header size {} + buffer size {}  >= total size {}",
+                                    arcticdb::Segment::FIXED_HEADER_SIZE,
+                                    hdr_size,
+                                    seg.buffer().bytes(),
+                                    write_size);
+
+                    ARCTICDB_SUBSAMPLE(AzureStorageUploadObject, 0)
+                    try{
+                        auto blob_client = container_client.GetBlockBlobClient(blob_name);
+                        ARCTICDB_RUNTIME_DEBUG(log::storage(), "Writing key {}: {}, with {} bytes of data",
+                                               variant_key_type(k),
+                                               variant_key_view(k),
+                                               seg.total_segment_size(hdr_size));
+                        blob_client.UploadFrom(dst, write_size, upload_option, get_context(request_timeout));
+                    }
+                    catch (const Azure::Core::RequestFailedException& e){
+                        util::raise_rte("Failed to write azure with key '{}' {} {}: {}",
+                                            k,
+                                            blob_name,
+                                            static_cast<int>(e.StatusCode),
+                                            e.ReasonPhrase);
+                    }
+
+                    ARCTICDB_RUNTIME_DEBUG(log::storage(), "Wrote key {}: {}, with {} bytes of data",
+                                        variant_key_type(k),
+                                        variant_key_view(k),
+                                        seg.total_segment_size(hdr_size));
+                }
+            }
+        );
+}
+
+template<class KeyBucketizer>
+void do_update_impl(
+    Composite<KeySegmentPair>&& kvs,
+    const std::string& root_folder,
+    Azure::Storage::Blobs::BlobContainerClient& container_client,
+    KeyBucketizer&& bucketizer,
+    const Azure::Storage::Blobs::UploadBlockBlobFromOptions& upload_option,
+    unsigned int request_timeout) {
+        // azure updates the key if it already exists
+        do_write_impl(std::move(kvs), root_folder, container_client, std::move(bucketizer), upload_option, request_timeout);
+}
+
+struct UnexpectedAzureErrorException : public std::exception {};
+
+template<class KeyBucketizer>
+void do_read_impl(Composite<VariantKey> && ks,
+    const ReadVisitor& visitor,
+    const std::string& root_folder,
+    Azure::Storage::Blobs::BlobContainerClient& container_client,
+    KeyBucketizer&& bucketizer,
+    ReadKeyOpts opts,
+    const Azure::Storage::Blobs::DownloadBlobToOptions& download_option,
+    unsigned int request_timeout) {
+        ARCTICDB_SAMPLE(AzureStorageRead, 0)
+        auto fmt_db = [](auto&& k) { return variant_key_type(k); };
+        std::vector<VariantKey> failed_reads;
+
+        (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach(
+            [&container_client, &root_folder, b=std::move(bucketizer), &visitor, &failed_reads,
+            opts=opts, &download_option, &request_timeout] (auto&& group) {
+            for (auto& k : group.values()) {
+                auto key_type_dir = key_type_folder(root_folder, variant_key_type(k));
+                auto blob_name = object_path(b.bucketize(key_type_dir, k), k);
+                try{
+                    ARCTICDB_DEBUG(log::storage(), "Reading key {}: {}", variant_key_type(k), variant_key_view(k));
+
+                    auto blob_client = container_client.GetBlockBlobClient(blob_name);
+                    auto properties = blob_client.GetProperties(Azure::Storage::Blobs::GetBlobPropertiesOptions{}, get_context(request_timeout)).Value;
+                    std::shared_ptr<Buffer> buffer = std::make_shared<Buffer>(properties.BlobSize);
+                    blob_client.DownloadTo(buffer->preamble(), buffer->available(), download_option, get_context(request_timeout));
+                    ARCTICDB_SUBSAMPLE(AzureStorageVisitSegment, 0)
+                    visitor(k, Segment::from_buffer(std::move(buffer)));
+
+                    ARCTICDB_DEBUG(log::storage(), "Read key {}: {}", variant_key_type(k), variant_key_view(k));
+                }
+                catch (const Azure::Core::RequestFailedException& e){
+                    if (!opts.dont_warn_about_missing_key) {
+                        log::storage().warn("Failed to read azure segment with key '{}' {} {}: {}",
+                                                k,
+                                                blob_name,
+                                                static_cast<int>(e.StatusCode),
+                                                e.ReasonPhrase);
+                    }
+                    failed_reads.push_back(k);
+                }
+            }
+        });
+        if(!failed_reads.empty())
+            throw KeyNotFoundException(Composite<VariantKey>{std::move(failed_reads)});
+}
+
+template<class KeyBucketizer>
+void do_remove_impl(Composite<VariantKey>&& ks,
+    const std::string& root_folder,
+    Azure::Storage::Blobs::BlobContainerClient& container_client,
+    KeyBucketizer&& bucketizer,
+    unsigned int request_timeout) {
+        ARCTICDB_SUBSAMPLE(AzureStorageDeleteBatch, 0)
+        auto fmt_db = [](auto&& k) { return variant_key_type(k); };
+        static const size_t delete_object_limit =
+            std::min(BATCH_SUBREQUEST_LIMIT, static_cast<size_t>(ConfigsMap::instance()->get_int("AzureStorage.DeleteBatchSize", BATCH_SUBREQUEST_LIMIT)));
+
+        auto batch = container_client.CreateBatch();
+        unsigned int batch_counter = 0u;
+        auto submit_batch = [&container_client, &request_timeout, &batch_counter](auto &batch) {
+            ARCTICDB_SUBSAMPLE(AzureStorageDeleteObjects, 0)
+            try{
+                container_client.SubmitBatch(batch, Azure::Storage::Blobs::SubmitBlobBatchOptions(), get_context(request_timeout));//To align with s3 behaviour, deleting non-exist objects is not an error, so not handling response
+            }
+            catch (const Azure::Core::RequestFailedException& e){
+                util::raise_rte("Failed to create azure segment batch remove request {}: {}",
+                                    static_cast<int>(e.StatusCode),
+                                    e.ReasonPhrase);
+            }
+            batch = container_client.CreateBatch();
+            batch_counter = 0u;
+        };
+
+        (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach(
+            [&root_folder, b=std::move(bucketizer), &batch, delete_object_limit=delete_object_limit, &batch_counter, &submit_batch] (auto&& group) {//bypass incorrect 'set but no used" error for delete_object_limit
+                auto key_type_dir = key_type_folder(root_folder, group.key());
+                for (auto k : folly::enumerate(group.values())) {
+                    auto blob_name = object_path(b.bucketize(key_type_dir, *k), *k);
+                    ARCTICDB_RUNTIME_DEBUG(log::storage(), "Removing azure object with key {}", blob_name);
+                    batch.DeleteBlob(blob_name);
+                    if (++batch_counter == delete_object_limit) {
+                        ARCTICDB_RUNTIME_DEBUG(log::storage(), "Submitting DeleteBlob batch");
+                        submit_batch(batch);
+                        ARCTICDB_RUNTIME_DEBUG(log::storage(), "Submitted DeleteBlob batch");
+                    }
+                }
+            }
+        );
+        if (batch_counter) {
+            ARCTICDB_RUNTIME_DEBUG(log::storage(), "Submitting DeleteBlob batch");
+            submit_batch(batch);
+            ARCTICDB_RUNTIME_DEBUG(log::storage(), "Submitted DeleteBlob batch");
+        }
+}
+
+auto default_prefix_handler() {
+    return [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor& key_descriptor, KeyType) {
+        return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
+    };
+}
+
+template<class KeyBucketizer, class PrefixHandler>
+void do_iterate_type_impl(KeyType key_type,
+    const IterateTypeVisitor& visitor,
+    const std::string& root_folder,
+    Azure::Storage::Blobs::BlobContainerClient& container_client,
+    KeyBucketizer&& bucketizer,
+    PrefixHandler&& prefix_handler = default_prefix_handler(),
+    const std::string& prefix = std::string{}) {
+        ARCTICDB_SAMPLE(AzureStorageIterateType, 0)
+        auto key_type_dir = key_type_folder(root_folder, key_type);
+
+        KeyDescriptor key_descriptor(prefix,
+            is_ref_key_class(key_type) ? IndexDescriptor::UNKNOWN : IndexDescriptor::TIMESTAMP, FormatType::TOKENIZED);
+        auto key_prefix = prefix_handler(prefix, key_type_dir, key_descriptor, key_type);
+        ARCTICDB_RUNTIME_DEBUG(log::storage(), "Searching for objects with prefix {}", key_prefix);
+        const auto root_folder_size = key_type_dir.size() + 1 + bucketizer.bucketize_length(key_type);
+
+        Azure::Storage::Blobs::ListBlobsOptions options;
+        options.Prefix = key_prefix;
+        try{
+            for (auto page = container_client.ListBlobs(options); page.HasPage(); page.MoveToNextPage()) {
+                for (const auto& blob : page.Blobs) {
+                    auto key = blob.Name.substr(root_folder_size);
+                    ARCTICDB_TRACE(log::version(), "Got object_list: {}, key: {}", blob.Name, key);
+                    auto k = variant_key_from_bytes(
+                                reinterpret_cast<uint8_t*>(key.data()),
+                                key.size(),
+                                key_type);
+                    ARCTICDB_DEBUG(log::storage(), "Iterating key {}: {}", variant_key_type(k), variant_key_view(k));
+                    ARCTICDB_SUBSAMPLE(AzureStorageVisitKey, 0)
+                    visitor(std::move(k));
+                    ARCTICDB_SUBSAMPLE(AzureStorageCursorNext, 0)
+                }
+            }
+        }
+        catch (const Azure::Core::RequestFailedException& e){
+            log::storage().warn("Failed to iterate azure blobs '{}' {}: {}",
+                                key_type,
+                                static_cast<int>(e.StatusCode),
+                                e.ReasonPhrase);
+        }
+}
+
+template<class KeyBucketizer>
+bool do_key_exists_impl(
+    const VariantKey& key,
+    const std::string& root_folder,
+    Azure::Storage::Blobs::BlobContainerClient& container_client,
+    KeyBucketizer&& bucketizer) {
+        auto key_type_dir = key_type_folder(root_folder, variant_key_type(key));
+        auto blob_name = object_path(bucketizer.bucketize(key_type_dir, key), key);
+        try{
+            auto blob_client = container_client.GetBlockBlobClient(blob_name);
+            auto properties = blob_client.GetProperties().Value;
+
+            if (properties.ETag.HasValue())
+                return true;
+        }
+        catch (const Azure::Core::RequestFailedException& e){
+            log::storage().debug("Failed to check azure key '{}' {} {}: {}",
+                                key,
+                                blob_name,
+                                static_cast<int>(e.StatusCode),
+                                e.ReasonPhrase);
+        }
+        return false;
+}
+} //namespace detail
+
+void AzureStorage::do_write(Composite<KeySegmentPair>&& kvs) {
+    detail::do_write_impl(std::move(kvs), root_folder_, container_client_, FlatBucketizer{}, upload_option_, request_timeout_);
+}
+
+void AzureStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts) {
+    detail::do_update_impl(std::move(kvs), root_folder_, container_client_, FlatBucketizer{}, upload_option_, request_timeout_);
+}
+
+void AzureStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts opts) {
+    detail::do_read_impl(std::move(ks), visitor, root_folder_, container_client_, FlatBucketizer{}, opts, download_option_, request_timeout_);
+}
+
+void AzureStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
+    detail::do_remove_impl(std::move(ks), root_folder_, container_client_, FlatBucketizer{}, request_timeout_);
+}
+
+void AzureStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string &prefix) {
+    auto prefix_handler = [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor key_descriptor, KeyType) {
+        return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
+    };
+
+    detail::do_iterate_type_impl(key_type, std::move(visitor), root_folder_, container_client_, FlatBucketizer{}, std::move(prefix_handler), prefix);
+}
+
+bool AzureStorage::do_key_exists(const VariantKey& key) {
+    return detail::do_key_exists_impl(key, root_folder_, container_client_, FlatBucketizer{});
+}
+
+} // namespace azure
+
+} // namespace arcticdb::storage
+
+
 namespace arcticdb::storage::azure{
 
 using namespace Azure::Storage;
@@ -33,7 +348,7 @@ AzureStorage::AzureStorage(const LibraryPath &library_path, OpenMode mode, const
             ARCTICDB_RUNTIME_DEBUG(log::storage(), "Azure prefix found, using: {}", conf.prefix());
             auto prefix_path = LibraryPath::from_delim_path(conf.prefix(), '.');
             root_folder_ = object_store_utils::get_root_folder(prefix_path);
-        } else 
+        } else
             ARCTICDB_RUNTIME_DEBUG(log::storage(), "Azure prefix not found, will use {}", root_folder_);
         unsigned int max_connections = conf.max_connections() == 0 ? ConfigsMap::instance()->get_int("VersionStore.NumIOThreads", 16) : conf.max_connections();
         upload_option_.TransferOptions.Concurrency = max_connections;
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_storage-inl.hpp b/cpp/arcticdb/storage/lmdb/lmdb_storage-inl.hpp
deleted file mode 100644
index fb0f515d2d..0000000000
--- a/cpp/arcticdb/storage/lmdb/lmdb_storage-inl.hpp
+++ /dev/null
@@ -1,255 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * 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.
- */
-
-#ifndef ARCTICDB_LMDB_STORAGE_H_
-#error "This should only be included by lmdb_storage.cpp"
-#endif
-
-#include <arcticdb/storage/lmdb/lmdb_storage.hpp>
-#include <arcticdb/util/preconditions.hpp>
-#include <arcticdb/util/pb_util.hpp>
-#include <arcticdb/log/log.hpp>
-#include <arcticdb/entity/serialized_key.hpp>
-#include <arcticdb/storage/storage.hpp>
-#include <arcticdb/storage/storage_options.hpp>
-#include <arcticdb/storage/storage_utils.hpp>
-
-#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
-#include <folly/gen/Base.h>
-
-namespace arcticdb::storage::lmdb {
-
-namespace fg = folly::gen;
-
-inline void LmdbStorage::do_write_internal(Composite<KeySegmentPair>&& kvs, ::lmdb::txn& txn) {
-    auto fmt_db = [](auto &&kv) { return kv.key_type(); };
-
-    (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-        auto db_name = fmt::format("{}", group.key());
-
-        ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
-        ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
-
-        ARCTICDB_SUBSAMPLE(LmdbStorageWriteValues, 0)
-        for (auto &kv : group.values()) {
-            ARCTICDB_DEBUG(log::storage(), "Lmdb storage writing segment with key {}", kv.key_view());
-            auto k = to_serialized_key(kv.variant_key());
-            auto &seg = kv.segment();
-            MDB_val mdb_key;
-            mdb_key.mv_data = k.data();
-            mdb_key.mv_size = k.size();
-
-            std::size_t hdr_sz = seg.segment_header_bytes_size();
-
-            MDB_val mdb_val;
-            mdb_val.mv_size = seg.total_segment_size(hdr_sz);
-            int64_t overwrite_flag = std::holds_alternative<RefKey>(kv.variant_key()) ? 0 : MDB_NOOVERWRITE;
-            ARCTICDB_SUBSAMPLE(LmdbPut, 0)
-            int res = ::mdb_put(txn.handle(), dbi.handle(), &mdb_key, &mdb_val, MDB_RESERVE | overwrite_flag);
-            if (res == MDB_KEYEXIST) {
-                throw DuplicateKeyException(kv.variant_key());
-            } else if (res != 0) {
-                throw std::runtime_error(fmt::format("Invalid lmdb error code {} while putting key {}",
-                                                     res, kv.key_view()));
-            }
-            ARCTICDB_SUBSAMPLE(LmdbMemCpy, 0)
-            // mdb_val now points to a reserved memory area we must write to
-            seg.write_to(reinterpret_cast<std::uint8_t *>(mdb_val.mv_data), hdr_sz);
-        }
-    });
-}
-
-inline void LmdbStorage::do_write(Composite<KeySegmentPair>&& kvs) {
-    ARCTICDB_SAMPLE(LmdbStorageWrite, 0)
-    std::lock_guard<std::mutex> lock{*write_mutex_};
-    auto txn = ::lmdb::txn::begin(env()); // scoped abort on exception, so no partial writes
-    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
-    do_write_internal(std::move(kvs), txn);
-    ARCTICDB_SUBSAMPLE(LmdbStorageCommit, 0)
-    txn.commit();
-}
-
-inline void LmdbStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts opts) {
-    ARCTICDB_SAMPLE(LmdbStorageUpdate, 0)
-    std::lock_guard<std::mutex> lock{*write_mutex_};
-    auto txn = ::lmdb::txn::begin(env());
-    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
-    auto keys = kvs.transform([](const auto& kv){return kv.variant_key();});
-    // Deleting keys (no error is thrown if the keys already exist)
-    RemoveOpts remove_opts;
-    remove_opts.ignores_missing_key_ = opts.upsert_;
-    auto failed_deletes = do_remove_internal(std::move(keys), txn, remove_opts);
-    if(!failed_deletes.empty()) {
-        ARCTICDB_SUBSAMPLE(LmdbStorageCommit, 0)
-        txn.commit();
-        throw KeyNotFoundException(Composite<VariantKey>(std::move(failed_deletes)));
-    }
-    do_write_internal(std::move(kvs), txn);
-    ARCTICDB_SUBSAMPLE(LmdbStorageCommit, 0)
-    txn.commit();
-}
-
-inline void LmdbStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, storage::ReadKeyOpts) {
-    ARCTICDB_SAMPLE(LmdbStorageRead, 0)
-    auto txn = ::lmdb::txn::begin(env(), nullptr, MDB_RDONLY);
-
-    auto fmt_db = [](auto &&k) { return variant_key_type(k); };
-    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
-    std::vector<VariantKey> failed_reads;
-
-    (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-        auto db_name = fmt::format("{}", group.key());
-        ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
-        ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
-        for (auto &k : group.values()) {
-            auto stored_key = to_serialized_key(k);
-            MDB_val mdb_key{stored_key.size(), stored_key.data()};
-            MDB_val mdb_val;
-            ARCTICDB_SUBSAMPLE(LmdbStorageGet, 0)
-
-            if (::lmdb::dbi_get(txn, dbi.handle(), &mdb_key, &mdb_val)) {
-                ARCTICDB_SUBSAMPLE(LmdbStorageVisitSegment, 0)
-                visitor(k, Segment::from_bytes(reinterpret_cast<std::uint8_t *>(mdb_val.mv_data),
-                                               mdb_val.mv_size));
-
-                ARCTICDB_DEBUG(log::storage(), "Read key {}: {}, with {} bytes of data", variant_key_type(k), variant_key_view(k), mdb_val.mv_size);
-            } else {
-                ARCTICDB_DEBUG(log::storage(), "Failed to find segment for key {}",variant_key_view(k));
-                failed_reads.push_back(k);
-            }
-        }
-    });
-    if(!failed_reads.empty())
-        throw KeyNotFoundException(Composite<VariantKey>(std::move(failed_reads)));
-}
-
-inline bool LmdbStorage::do_key_exists(const VariantKey&key) {
-    ARCTICDB_SAMPLE(LmdbStorageKeyExists, 0)
-    auto txn = ::lmdb::txn::begin(env(), nullptr, MDB_RDONLY);
-    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
-
-    auto db_name = fmt::format("{}", variant_key_type(key));
-    ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
-
-    try {
-        ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
-        if (unsigned int tmp; ::mdb_dbi_flags(txn, dbi, &tmp) == EINVAL) {
-            return false;
-        }
-        auto stored_key = to_serialized_key(key);
-        MDB_val mdb_key{stored_key.size(), stored_key.data()};
-        MDB_val mdb_val;
-        return ::lmdb::dbi_get(txn, dbi.handle(), &mdb_key, &mdb_val);
-    } catch (const ::lmdb::not_found_error &ex) {
-        ARCTICDB_DEBUG(log::storage(), "Caught lmdb not found error: {}", ex.what());
-        return false;
-    }
-}
-
-inline std::vector<VariantKey> LmdbStorage::do_remove_internal(Composite<VariantKey>&& ks, ::lmdb::txn& txn, RemoveOpts opts)
-{
-    auto fmt_db = [](auto &&k) { return variant_key_type(k); };
-    std::vector<VariantKey> failed_deletes;
-
-    (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-        auto db_name = fmt::format("{}", group.key());
-        ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
-        try {
-            // If no key of this type has been written before, this can fail
-            ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
-            for (auto &k : group.values()) {
-                auto stored_key = to_serialized_key(k);
-                MDB_val mdb_key{stored_key.size(), stored_key.data()};
-                ARCTICDB_SUBSAMPLE(LmdbStorageDel, 0)
-
-                if (::lmdb::dbi_del(txn, dbi.handle(), &mdb_key)) {
-                    ARCTICDB_DEBUG(log::storage(), "Deleted segment for key {}", variant_key_view(k));
-                } else {
-                    if (!opts.ignores_missing_key_) {
-                        log::storage().warn("Failed to delete segment for key {}", variant_key_view(k));
-                        failed_deletes.push_back(k);
-                    }
-                }
-            }
-        } catch (const std::exception& e) {
-            if (!opts.ignores_missing_key_) {
-                for (auto &k : group.values()) {
-                    log::storage().warn("Failed to delete segment for key {}", variant_key_view(k) );
-                                        failed_deletes.push_back(k);
-                }
-            }
-        }
-    });
-    return failed_deletes;
-}
-
-inline void LmdbStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts opts)
-{
-    ARCTICDB_SAMPLE(LmdbStorageRemove, 0)
-    std::lock_guard<std::mutex> lock{*write_mutex_};
-    auto txn = ::lmdb::txn::begin(env());
-    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
-    auto failed_deletes = do_remove_internal(std::move(ks), txn, opts);
-    ARCTICDB_SUBSAMPLE(LmdbStorageCommit, 0)
-    txn.commit();
-
-    if(!failed_deletes.empty())
-        throw KeyNotFoundException(Composite<VariantKey>(std::move(failed_deletes)));
-}
-
-bool LmdbStorage::do_fast_delete() {
-    std::lock_guard<std::mutex> lock{*write_mutex_};
-    // bool is probably not the best return type here but it does help prevent the insane boilerplate for
-    // an additional function that checks whether this is supported (like the prefix matching)
-    auto dtxn = ::lmdb::txn::begin(env());
-    foreach_key_type([&] (KeyType key_type) {
-        if (key_type == KeyType::TOMBSTONE) {
-            // TOMBSTONE and LOCK both format to code 'x' - do not try to drop both
-            return;
-        }
-        auto db_name = fmt::format("{}", key_type);
-        ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
-        ARCTICDB_DEBUG(log::storage(), "dropping {}", db_name);
-        ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
-        ::lmdb::dbi_drop(dtxn, dbi);
-    });
-
-    dtxn.commit();
-    return true;
-}
-
-inline void LmdbStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string &prefix) {
-    ARCTICDB_SAMPLE(LmdbStorageItType, 0);
-    auto txn = ::lmdb::txn::begin(env(), nullptr, MDB_RDONLY); // scoped abort on
-    std::string type_db = fmt::format("{}", key_type);
-    ::lmdb::dbi& dbi = dbi_by_key_type_.at(type_db);
-
-    ARCTICDB_SUBSAMPLE(LmdbStorageOpenCursor, 0)
-    auto db_cursor = ::lmdb::cursor::open(txn, dbi);
-
-    MDB_val mdb_db_key;
-    ARCTICDB_SUBSAMPLE(LmdbStorageCursorFirst, 0)
-    if (!db_cursor.get(&mdb_db_key, nullptr, MDB_cursor_op::MDB_FIRST)) {
-        return;
-    }
-    auto prefix_matcher = stream_id_prefix_matcher(prefix);
-    do {
-        auto k = variant_key_from_bytes(
-            static_cast<uint8_t *>(mdb_db_key.mv_data),
-            mdb_db_key.mv_size,
-            key_type);
-
-        ARCTICDB_DEBUG(log::storage(), "Iterating key {}: {}", variant_key_type(k), variant_key_view(k));
-        if (prefix_matcher(variant_key_id(k))) {
-            ARCTICDB_SUBSAMPLE(LmdbStorageVisitKey, 0)
-            visitor(std::move(k));
-        }
-        ARCTICDB_SUBSAMPLE(LmdbStorageCursorNext, 0)
-    } while (db_cursor.get(&mdb_db_key, nullptr, MDB_cursor_op::MDB_NEXT));
-}
-
-}
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp b/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
index e395a87ca4..4c7fac9fdd 100644
--- a/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
+++ b/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
@@ -5,18 +5,258 @@
  * 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.
  */
 
-#define ARCTICDB_LMDB_STORAGE_H_ 1
-#include <arcticdb/storage/lmdb/lmdb_storage-inl.hpp>
+#include <arcticdb/storage/lmdb/lmdb_storage.hpp>
+
+#include <filesystem>
+
 #include <arcticdb/log/log.hpp>
 #include <arcticdb/entity/atom_key.hpp>
 #include <arcticdb/storage/library_path.hpp>
 #include <arcticdb/storage/open_mode.hpp>
 #include <arcticdb/util/format_bytes.hpp>
+#include <arcticdb/util/preconditions.hpp>
+#include <arcticdb/util/pb_util.hpp>
+#include <arcticdb/log/log.hpp>
+#include <arcticdb/entity/serialized_key.hpp>
+#include <arcticdb/storage/storage.hpp>
+#include <arcticdb/storage/storage_options.hpp>
+#include <arcticdb/storage/storage_utils.hpp>
 
-#include <filesystem>
+#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
+#include <folly/gen/Base.h>
 
 namespace arcticdb::storage::lmdb {
 
+namespace fg = folly::gen;
+
+void LmdbStorage::do_write_internal(Composite<KeySegmentPair>&& kvs, ::lmdb::txn& txn) {
+    auto fmt_db = [](auto &&kv) { return kv.key_type(); };
+
+    (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+        auto db_name = fmt::format("{}", group.key());
+
+        ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
+        ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
+
+        ARCTICDB_SUBSAMPLE(LmdbStorageWriteValues, 0)
+        for (auto &kv : group.values()) {
+            ARCTICDB_DEBUG(log::storage(), "Lmdb storage writing segment with key {}", kv.key_view());
+            auto k = to_serialized_key(kv.variant_key());
+            auto &seg = kv.segment();
+            MDB_val mdb_key;
+            mdb_key.mv_data = k.data();
+            mdb_key.mv_size = k.size();
+
+            std::size_t hdr_sz = seg.segment_header_bytes_size();
+
+            MDB_val mdb_val;
+            mdb_val.mv_size = seg.total_segment_size(hdr_sz);
+            int64_t overwrite_flag = std::holds_alternative<RefKey>(kv.variant_key()) ? 0 : MDB_NOOVERWRITE;
+            ARCTICDB_SUBSAMPLE(LmdbPut, 0)
+            int res = ::mdb_put(txn.handle(), dbi.handle(), &mdb_key, &mdb_val, MDB_RESERVE | overwrite_flag);
+            if (res == MDB_KEYEXIST) {
+                throw DuplicateKeyException(kv.variant_key());
+            } else if (res != 0) {
+                throw std::runtime_error(fmt::format("Invalid lmdb error code {} while putting key {}",
+                                                     res, kv.key_view()));
+            }
+            ARCTICDB_SUBSAMPLE(LmdbMemCpy, 0)
+            // mdb_val now points to a reserved memory area we must write to
+            seg.write_to(reinterpret_cast<std::uint8_t *>(mdb_val.mv_data), hdr_sz);
+        }
+    });
+}
+
+void LmdbStorage::do_write(Composite<KeySegmentPair>&& kvs) {
+    ARCTICDB_SAMPLE(LmdbStorageWrite, 0)
+    std::lock_guard<std::mutex> lock{*write_mutex_};
+    auto txn = ::lmdb::txn::begin(env()); // scoped abort on exception, so no partial writes
+    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
+    do_write_internal(std::move(kvs), txn);
+    ARCTICDB_SUBSAMPLE(LmdbStorageCommit, 0)
+    txn.commit();
+}
+
+void LmdbStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts opts) {
+    ARCTICDB_SAMPLE(LmdbStorageUpdate, 0)
+    std::lock_guard<std::mutex> lock{*write_mutex_};
+    auto txn = ::lmdb::txn::begin(env());
+    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
+    auto keys = kvs.transform([](const auto& kv){return kv.variant_key();});
+    // Deleting keys (no error is thrown if the keys already exist)
+    RemoveOpts remove_opts;
+    remove_opts.ignores_missing_key_ = opts.upsert_;
+    auto failed_deletes = do_remove_internal(std::move(keys), txn, remove_opts);
+    if(!failed_deletes.empty()) {
+        ARCTICDB_SUBSAMPLE(LmdbStorageCommit, 0)
+        txn.commit();
+        throw KeyNotFoundException(Composite<VariantKey>(std::move(failed_deletes)));
+    }
+    do_write_internal(std::move(kvs), txn);
+    ARCTICDB_SUBSAMPLE(LmdbStorageCommit, 0)
+    txn.commit();
+}
+
+void LmdbStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, storage::ReadKeyOpts) {
+    ARCTICDB_SAMPLE(LmdbStorageRead, 0)
+    auto txn = ::lmdb::txn::begin(env(), nullptr, MDB_RDONLY);
+
+    auto fmt_db = [](auto &&k) { return variant_key_type(k); };
+    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
+    std::vector<VariantKey> failed_reads;
+
+    (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+        auto db_name = fmt::format("{}", group.key());
+        ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
+        ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
+        for (auto &k : group.values()) {
+            auto stored_key = to_serialized_key(k);
+            MDB_val mdb_key{stored_key.size(), stored_key.data()};
+            MDB_val mdb_val;
+            ARCTICDB_SUBSAMPLE(LmdbStorageGet, 0)
+
+            if (::lmdb::dbi_get(txn, dbi.handle(), &mdb_key, &mdb_val)) {
+                ARCTICDB_SUBSAMPLE(LmdbStorageVisitSegment, 0)
+                visitor(k, Segment::from_bytes(reinterpret_cast<std::uint8_t *>(mdb_val.mv_data),
+                                               mdb_val.mv_size));
+
+                ARCTICDB_DEBUG(log::storage(), "Read key {}: {}, with {} bytes of data", variant_key_type(k), variant_key_view(k), mdb_val.mv_size);
+            } else {
+                ARCTICDB_DEBUG(log::storage(), "Failed to find segment for key {}",variant_key_view(k));
+                failed_reads.push_back(k);
+            }
+        }
+    });
+    if(!failed_reads.empty())
+        throw KeyNotFoundException(Composite<VariantKey>(std::move(failed_reads)));
+}
+
+bool LmdbStorage::do_key_exists(const VariantKey&key) {
+    ARCTICDB_SAMPLE(LmdbStorageKeyExists, 0)
+    auto txn = ::lmdb::txn::begin(env(), nullptr, MDB_RDONLY);
+    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
+
+    auto db_name = fmt::format("{}", variant_key_type(key));
+    ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
+
+    try {
+        ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
+        if (unsigned int tmp; ::mdb_dbi_flags(txn, dbi, &tmp) == EINVAL) {
+            return false;
+        }
+        auto stored_key = to_serialized_key(key);
+        MDB_val mdb_key{stored_key.size(), stored_key.data()};
+        MDB_val mdb_val;
+        return ::lmdb::dbi_get(txn, dbi.handle(), &mdb_key, &mdb_val);
+    } catch (const ::lmdb::not_found_error &ex) {
+        ARCTICDB_DEBUG(log::storage(), "Caught lmdb not found error: {}", ex.what());
+        return false;
+    }
+}
+
+std::vector<VariantKey> LmdbStorage::do_remove_internal(Composite<VariantKey>&& ks, ::lmdb::txn& txn, RemoveOpts opts)
+{
+    auto fmt_db = [](auto &&k) { return variant_key_type(k); };
+    std::vector<VariantKey> failed_deletes;
+
+    (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+        auto db_name = fmt::format("{}", group.key());
+        ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
+        try {
+            // If no key of this type has been written before, this can fail
+            ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
+            for (auto &k : group.values()) {
+                auto stored_key = to_serialized_key(k);
+                MDB_val mdb_key{stored_key.size(), stored_key.data()};
+                ARCTICDB_SUBSAMPLE(LmdbStorageDel, 0)
+
+                if (::lmdb::dbi_del(txn, dbi.handle(), &mdb_key)) {
+                    ARCTICDB_DEBUG(log::storage(), "Deleted segment for key {}", variant_key_view(k));
+                } else {
+                    if (!opts.ignores_missing_key_) {
+                        log::storage().warn("Failed to delete segment for key {}", variant_key_view(k));
+                        failed_deletes.push_back(k);
+                    }
+                }
+            }
+        } catch (const std::exception& e) {
+            if (!opts.ignores_missing_key_) {
+                for (auto &k : group.values()) {
+                    log::storage().warn("Failed to delete segment for key {}", variant_key_view(k) );
+                                        failed_deletes.push_back(k);
+                }
+            }
+        }
+    });
+    return failed_deletes;
+}
+
+void LmdbStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts opts)
+{
+    ARCTICDB_SAMPLE(LmdbStorageRemove, 0)
+    std::lock_guard<std::mutex> lock{*write_mutex_};
+    auto txn = ::lmdb::txn::begin(env());
+    ARCTICDB_SUBSAMPLE(LmdbStorageInTransaction, 0)
+    auto failed_deletes = do_remove_internal(std::move(ks), txn, opts);
+    ARCTICDB_SUBSAMPLE(LmdbStorageCommit, 0)
+    txn.commit();
+
+    if(!failed_deletes.empty())
+        throw KeyNotFoundException(Composite<VariantKey>(std::move(failed_deletes)));
+}
+
+bool LmdbStorage::do_fast_delete() {
+    std::lock_guard<std::mutex> lock{*write_mutex_};
+    // bool is probably not the best return type here but it does help prevent the insane boilerplate for
+    // an additional function that checks whether this is supported (like the prefix matching)
+    auto dtxn = ::lmdb::txn::begin(env());
+    foreach_key_type([&] (KeyType key_type) {
+        if (key_type == KeyType::TOMBSTONE) {
+            // TOMBSTONE and LOCK both format to code 'x' - do not try to drop both
+            return;
+        }
+        auto db_name = fmt::format("{}", key_type);
+        ARCTICDB_SUBSAMPLE(LmdbStorageOpenDb, 0)
+        ARCTICDB_DEBUG(log::storage(), "dropping {}", db_name);
+        ::lmdb::dbi& dbi = dbi_by_key_type_.at(db_name);
+        ::lmdb::dbi_drop(dtxn, dbi);
+    });
+
+    dtxn.commit();
+    return true;
+}
+
+void LmdbStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string &prefix) {
+    ARCTICDB_SAMPLE(LmdbStorageItType, 0);
+    auto txn = ::lmdb::txn::begin(env(), nullptr, MDB_RDONLY); // scoped abort on
+    std::string type_db = fmt::format("{}", key_type);
+    ::lmdb::dbi& dbi = dbi_by_key_type_.at(type_db);
+
+    ARCTICDB_SUBSAMPLE(LmdbStorageOpenCursor, 0)
+    auto db_cursor = ::lmdb::cursor::open(txn, dbi);
+
+    MDB_val mdb_db_key;
+    ARCTICDB_SUBSAMPLE(LmdbStorageCursorFirst, 0)
+    if (!db_cursor.get(&mdb_db_key, nullptr, MDB_cursor_op::MDB_FIRST)) {
+        return;
+    }
+    auto prefix_matcher = stream_id_prefix_matcher(prefix);
+    do {
+        auto k = variant_key_from_bytes(
+            static_cast<uint8_t *>(mdb_db_key.mv_data),
+            mdb_db_key.mv_size,
+            key_type);
+
+        ARCTICDB_DEBUG(log::storage(), "Iterating key {}: {}", variant_key_type(k), variant_key_view(k));
+        if (prefix_matcher(variant_key_id(k))) {
+            ARCTICDB_SUBSAMPLE(LmdbStorageVisitKey, 0)
+            visitor(std::move(k));
+        }
+        ARCTICDB_SUBSAMPLE(LmdbStorageCursorNext, 0)
+    } while (db_cursor.get(&mdb_db_key, nullptr, MDB_cursor_op::MDB_NEXT));
+}
+
+
 namespace {
 template<class T>
 T or_else(T val, T or_else_val, T def = T()) {
diff --git a/cpp/arcticdb/storage/memory/memory_storage-inl.hpp b/cpp/arcticdb/storage/memory/memory_storage-inl.hpp
deleted file mode 100644
index 085198b6d9..0000000000
--- a/cpp/arcticdb/storage/memory/memory_storage-inl.hpp
+++ /dev/null
@@ -1,141 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * 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.
- */
-
-#include <folly/gen/Base.h>
-#ifndef ARCTICDB_MEMORY_STORAGE_H_
-#error "This should only be included by memory_storage.hpp"
-#endif
-
-#include <arcticdb/util/preconditions.hpp>
-#include <arcticdb/entity/performance_tracing.hpp>
-#include <arcticdb/storage/memory/memory_storage.hpp>
-#include <arcticdb/storage/storage_options.hpp>
-#include <arcticdb/storage/storage_utils.hpp>
-
-namespace arcticdb::storage::memory {
-
-    namespace fg = folly::gen;
-
-    inline void MemoryStorage::do_write(Composite<KeySegmentPair>&& kvs) {
-        ARCTICDB_SAMPLE(MemoryStorageWrite, 0)
-
-        auto fmt_db = [](auto &&k) { return variant_key_type(k.variant_key()); };
-
-        (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-            auto& key_vec = data_[group.key()];
-
-            for (auto &kv : group.values()) {
-                util::variant_match(kv.variant_key(),
-                                    [&](const RefKey &key) {
-                                        if (auto it = key_vec.find(key); it != key_vec.end()) {
-                                            key_vec.erase(it);
-                                        }
-
-                                        key_vec.try_emplace(key, kv.segment());
-                                    },
-                                    [&](const AtomKey &key) {
-                                        if (key_vec.find(key) != key_vec.end()) {
-                                            throw DuplicateKeyException(key);
-                                        }
-
-                                        key_vec.try_emplace(key, kv.segment());
-                                    }
-                );
-            }
-        });
-    }
-
-    inline void MemoryStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts opts) {
-        ARCTICDB_SAMPLE(MemoryStorageUpdate, 0)
-
-        auto fmt_db = [](auto &&k) { return variant_key_type(k.variant_key()); };
-
-        (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-            auto& key_vec = data_[group.key()];
-
-            for (auto &kv : group.values()) {
-                auto it = key_vec.find(kv.variant_key());
-
-                util::check_rte(opts.upsert_ || it != key_vec.end(), "update called with upsert=false but key does not exist");
-
-                if(it != key_vec.end()) {
-                    key_vec.erase(it);
-                }
-                key_vec.insert(std::make_pair(kv.variant_key(), kv.segment()));
-            }
-        });
-    }
-
-    inline void MemoryStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts) {
-        ARCTICDB_SAMPLE(MemoryStorageRead, 0)
-        auto fmt_db = [](auto &&k) { return variant_key_type(k); };
-
-        (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-            auto& key_vec = data_[group.key()];
-            for (auto &k : group.values()) {
-                auto it = key_vec.find(k);
-
-                if(it != key_vec.end()) {
-                    ARCTICDB_DEBUG(log::storage(), "Read key {}: {}", variant_key_type(k), variant_key_view(k));
-                    auto seg = it->second;
-                    visitor(k, std::move(seg));
-                } else {
-                    throw KeyNotFoundException(std::move(ks));
-                }
-            }
-        });
-    }
-
-    inline bool MemoryStorage::do_key_exists(const VariantKey& key) {
-        ARCTICDB_SAMPLE(MemoryStorageKeyExists, 0)
-        const auto& key_vec = data_[variant_key_type(key)];
-        auto it = key_vec.find(key);
-        return it != key_vec.end();
-    }
-
-    inline void MemoryStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts opts)
-    {
-        ARCTICDB_SAMPLE(MemoryStorageRemove, 0)
-        auto fmt_db = [](auto &&k) { return variant_key_type(k); };
-
-        (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-            auto& key_vec = data_[group.key()];
-
-            for (auto &k : group.values()) {
-                auto it = key_vec.find(k);
-
-                if(it != key_vec.end()) {
-                    ARCTICDB_DEBUG(log::storage(), "Read key {}: {}, with {} bytes of data", variant_key_type(k), variant_key_view(k));
-                    key_vec.erase(it);
-                } else if (!opts.ignores_missing_key_) {
-                    util::raise_rte("Failed to find segment for key {}",variant_key_view(k));
-                }
-            }
-        });
-    }
-
-    bool MemoryStorage::do_fast_delete() {
-        foreach_key_type([&] (KeyType key_type) {
-            data_[key_type].clear();
-        });
-        return true;
-    }
-
-    inline void MemoryStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string& prefix) {
-        ARCTICDB_SAMPLE(MemoryStorageItType, 0)
-        auto& key_vec = data_[key_type];
-        auto prefix_matcher = stream_id_prefix_matcher(prefix);
-
-        for(auto& key_value : key_vec) {
-            auto key = key_value.first;
-
-            if (prefix_matcher(variant_key_id(key))) {
-                visitor(std::move(key));
-            }
-        }
-    }
-}
diff --git a/cpp/arcticdb/storage/memory/memory_storage.cpp b/cpp/arcticdb/storage/memory/memory_storage.cpp
index 0be2e50972..8a2a9bd953 100644
--- a/cpp/arcticdb/storage/memory/memory_storage.cpp
+++ b/cpp/arcticdb/storage/memory/memory_storage.cpp
@@ -4,15 +4,144 @@
  *
  * 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.
  */
-#define ARCTICDB_MEMORY_STORAGE_H_
-#include <arcticdb/storage/memory/memory_storage-inl.hpp>
 #include <arcticdb/storage/memory/memory_storage.hpp>
 
+
+#include <folly/gen/Base.h>
+
+#include <arcticdb/util/preconditions.hpp>
+#include <arcticdb/entity/performance_tracing.hpp>
+#include <arcticdb/storage/storage_options.hpp>
+#include <arcticdb/storage/storage_utils.hpp>
+
 namespace arcticdb::storage::memory {
+
+    namespace fg = folly::gen;
+
+    void MemoryStorage::do_write(Composite<KeySegmentPair>&& kvs) {
+        ARCTICDB_SAMPLE(MemoryStorageWrite, 0)
+
+        auto fmt_db = [](auto &&k) { return variant_key_type(k.variant_key()); };
+
+        (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+            auto& key_vec = data_[group.key()];
+
+            for (auto &kv : group.values()) {
+                util::variant_match(kv.variant_key(),
+                                    [&](const RefKey &key) {
+                                        if (auto it = key_vec.find(key); it != key_vec.end()) {
+                                            key_vec.erase(it);
+                                        }
+
+                                        key_vec.try_emplace(key, kv.segment());
+                                    },
+                                    [&](const AtomKey &key) {
+                                        if (key_vec.find(key) != key_vec.end()) {
+                                            throw DuplicateKeyException(key);
+                                        }
+
+                                        key_vec.try_emplace(key, kv.segment());
+                                    }
+                );
+            }
+        });
+    }
+
+    void MemoryStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts opts) {
+        ARCTICDB_SAMPLE(MemoryStorageUpdate, 0)
+
+        auto fmt_db = [](auto &&k) { return variant_key_type(k.variant_key()); };
+
+        (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+            auto& key_vec = data_[group.key()];
+
+            for (auto &kv : group.values()) {
+                auto it = key_vec.find(kv.variant_key());
+
+                util::check_rte(opts.upsert_ || it != key_vec.end(), "update called with upsert=false but key does not exist");
+
+                if(it != key_vec.end()) {
+                    key_vec.erase(it);
+                }
+                key_vec.insert(std::make_pair(kv.variant_key(), kv.segment()));
+            }
+        });
+    }
+
+    void MemoryStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts) {
+        ARCTICDB_SAMPLE(MemoryStorageRead, 0)
+        auto fmt_db = [](auto &&k) { return variant_key_type(k); };
+
+        (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+            auto& key_vec = data_[group.key()];
+            for (auto &k : group.values()) {
+                auto it = key_vec.find(k);
+
+                if(it != key_vec.end()) {
+                    ARCTICDB_DEBUG(log::storage(), "Read key {}: {}", variant_key_type(k), variant_key_view(k));
+                    auto seg = it->second;
+                    visitor(k, std::move(seg));
+                } else {
+                    throw KeyNotFoundException(std::move(ks));
+                }
+            }
+        });
+    }
+
+    bool MemoryStorage::do_key_exists(const VariantKey& key) {
+        ARCTICDB_SAMPLE(MemoryStorageKeyExists, 0)
+        const auto& key_vec = data_[variant_key_type(key)];
+        auto it = key_vec.find(key);
+        return it != key_vec.end();
+    }
+
+    void MemoryStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts opts)
+    {
+        ARCTICDB_SAMPLE(MemoryStorageRemove, 0)
+        auto fmt_db = [](auto &&k) { return variant_key_type(k); };
+
+        (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+            auto& key_vec = data_[group.key()];
+
+            for (auto &k : group.values()) {
+                auto it = key_vec.find(k);
+
+                if(it != key_vec.end()) {
+                    ARCTICDB_DEBUG(log::storage(), "Read key {}: {}, with {} bytes of data", variant_key_type(k), variant_key_view(k));
+                    key_vec.erase(it);
+                } else if (!opts.ignores_missing_key_) {
+                    util::raise_rte("Failed to find segment for key {}",variant_key_view(k));
+                }
+            }
+        });
+    }
+
+    bool MemoryStorage::do_fast_delete() {
+        foreach_key_type([&] (KeyType key_type) {
+            data_[key_type].clear();
+        });
+        return true;
+    }
+
+    void MemoryStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string& prefix) {
+        ARCTICDB_SAMPLE(MemoryStorageItType, 0)
+        auto& key_vec = data_[key_type];
+        auto prefix_matcher = stream_id_prefix_matcher(prefix);
+
+        for(auto& key_value : key_vec) {
+            auto key = key_value.first;
+
+            if (prefix_matcher(variant_key_id(key))) {
+                visitor(std::move(key));
+            }
+        }
+    }
+
     MemoryStorage::MemoryStorage(const LibraryPath &library_path, OpenMode mode, const Config&) :
         Storage(library_path, mode) {
         arcticdb::entity::foreach_key_type([this](KeyType&& key_type) {
             data_[key_type];
         });
     }
-} // arcticdb::storage::memory
+
+}
diff --git a/cpp/arcticdb/storage/mongo/mongo_storage-inl.hpp b/cpp/arcticdb/storage/mongo/mongo_storage-inl.hpp
deleted file mode 100644
index 9eaac1952e..0000000000
--- a/cpp/arcticdb/storage/mongo/mongo_storage-inl.hpp
+++ /dev/null
@@ -1,109 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * 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.
- */
-
-#include <folly/gen/Base.h>
-#ifndef ARCTICDB_MONGO_STORAGE_H_
-#error "This should only be included by mongo_storage.cpp"
-#endif
-
-#include <arcticdb/storage/mongo/mongo_storage.hpp>
-
-#include <folly/gen/Base.h>
-#include <arcticdb/storage/mongo/mongo_client.hpp>
-#include <arcticdb/entity/performance_tracing.hpp>
-#include <arcticdb/storage/storage_options.hpp>
-#include <arcticdb/storage/storage.hpp>
-
-namespace arcticdb::storage::mongo {
-
-inline std::string MongoStorage::collection_name(KeyType k) {
-    return (fmt::format("{}{}", prefix_, k));
-}
-
-inline void MongoStorage::do_write(Composite<KeySegmentPair>&& kvs) {
-    namespace fg = folly::gen;
-    auto fmt_db = [](auto &&kv) { return kv.key_type(); };
-
-    ARCTICDB_SAMPLE(MongoStorageWrite, 0)
-
-    (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-        for (auto &kv : group.values()) {
-            auto collection = collection_name(kv.key_type());
-            client_->write_segment(db_, collection, std::move(kv));
-        }
-    });
-}
-
-inline void MongoStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts opts) {
-    namespace fg = folly::gen;
-    auto fmt_db = [](auto &&kv) { return kv.key_type(); };
-
-    ARCTICDB_SAMPLE(MongoStorageWrite, 0)
-
-    (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-        for (auto &kv : group.values()) {
-            auto collection = collection_name(kv.key_type());
-            client_->update_segment(db_, collection, std::move(kv), opts.upsert_);
-        }
-    });
-}
-
-inline void MongoStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts) {
-    namespace fg = folly::gen;
-    auto fmt_db = [](auto &&k) { return variant_key_type(k); };
-    ARCTICDB_SAMPLE(MongoStorageRead, 0)
-    std::vector<VariantKey> failed_reads;
-
-    (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-        for (auto &k : group.values()) {
-                auto collection = collection_name(variant_key_type(k));
-                auto kv = client_->read_segment(db_, collection, k);
-                if(kv.has_segment())
-                    visitor(k, std::move(kv.segment()));
-                else
-                   failed_reads.push_back(k);
-        }
-    });
-
-    if(!failed_reads.empty())
-        throw KeyNotFoundException(Composite<VariantKey>{std::move(failed_reads)});
-}
-
-bool MongoStorage::do_fast_delete() {
-    foreach_key_type([&] (KeyType key_type) {
-        auto collection = collection_name(key_type);
-        client_->drop_collection(db_, collection);
-    });
-    return true;
-}
-
-inline void MongoStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
-    namespace fg = folly::gen;
-    auto fmt_db = [](auto &&k) { return variant_key_type(k); };
-    ARCTICDB_SAMPLE(MongoStorageRemove, 0)
-
-    (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
-        for (auto &k : group.values()) {
-            auto collection = collection_name(variant_key_type(k));
-            client_->remove_keyvalue(db_, collection, k);
-        }
-    });
-}
-
-inline void MongoStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string &prefix) {
-    auto collection = collection_name(key_type);
-    auto func = folly::Function<void(entity::VariantKey&&)>(visitor);
-    ARCTICDB_SAMPLE(MongoStorageItType, 0)
-    client_->iterate_type(db_, collection, key_type, std::move(func), prefix);
-}
-
-inline bool MongoStorage::do_key_exists(const VariantKey& key) {
-    auto collection = collection_name(variant_key_type(key));
-    return client_->key_exists(db_, collection, key);
-}
-
-} //namespace arcticdb::storage::mongo
diff --git a/cpp/arcticdb/storage/mongo/mongo_storage.cpp b/cpp/arcticdb/storage/mongo/mongo_storage.cpp
index de0cfc6ea2..8d0315a399 100644
--- a/cpp/arcticdb/storage/mongo/mongo_storage.cpp
+++ b/cpp/arcticdb/storage/mongo/mongo_storage.cpp
@@ -5,19 +5,110 @@
  * 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.
  */
 
+
 #include <arcticdb/storage/mongo/mongo_storage.hpp>
-#define ARCTICDB_MONGO_STORAGE_H_
-#include <arcticdb/storage/mongo/mongo_storage-inl.hpp>
-#include <arcticdb/storage/mongo/mongo_instance.hpp>
-#include <arcticdb/entity/index_range.hpp>
-#include <arcticdb/storage/mongo/mongo_client.hpp>
 
 #include <google/protobuf/io/zero_copy_stream_impl_lite.h>
 #include <arcticdb/util/configs_map.hpp>
 #include <fmt/format.h>
+#include <folly/gen/Base.h>
+
+#include <arcticdb/storage/mongo/mongo_instance.hpp>
+#include <arcticdb/entity/index_range.hpp>
+#include <arcticdb/storage/mongo/mongo_client.hpp>
+#include <arcticdb/entity/performance_tracing.hpp>
+#include <arcticdb/storage/storage_options.hpp>
+#include <arcticdb/storage/storage.hpp>
 
 namespace arcticdb::storage::mongo {
 
+std::string MongoStorage::collection_name(KeyType k) {
+    return (fmt::format("{}{}", prefix_, k));
+}
+
+void MongoStorage::do_write(Composite<KeySegmentPair>&& kvs) {
+    namespace fg = folly::gen;
+    auto fmt_db = [](auto &&kv) { return kv.key_type(); };
+
+    ARCTICDB_SAMPLE(MongoStorageWrite, 0)
+
+    (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+        for (auto &kv : group.values()) {
+            auto collection = collection_name(kv.key_type());
+            client_->write_segment(db_, collection, std::move(kv));
+        }
+    });
+}
+
+void MongoStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts opts) {
+    namespace fg = folly::gen;
+    auto fmt_db = [](auto &&kv) { return kv.key_type(); };
+
+    ARCTICDB_SAMPLE(MongoStorageWrite, 0)
+
+    (fg::from(kvs.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+        for (auto &kv : group.values()) {
+            auto collection = collection_name(kv.key_type());
+            client_->update_segment(db_, collection, std::move(kv), opts.upsert_);
+        }
+    });
+}
+
+void MongoStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts) {
+    namespace fg = folly::gen;
+    auto fmt_db = [](auto &&k) { return variant_key_type(k); };
+    ARCTICDB_SAMPLE(MongoStorageRead, 0)
+    std::vector<VariantKey> failed_reads;
+
+    (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+        for (auto &k : group.values()) {
+                auto collection = collection_name(variant_key_type(k));
+                auto kv = client_->read_segment(db_, collection, k);
+                if(kv.has_segment())
+                    visitor(k, std::move(kv.segment()));
+                else
+                   failed_reads.push_back(k);
+        }
+    });
+
+    if(!failed_reads.empty())
+        throw KeyNotFoundException(Composite<VariantKey>{std::move(failed_reads)});
+}
+
+bool MongoStorage::do_fast_delete() {
+    foreach_key_type([&] (KeyType key_type) {
+        auto collection = collection_name(key_type);
+        client_->drop_collection(db_, collection);
+    });
+    return true;
+}
+
+void MongoStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
+    namespace fg = folly::gen;
+    auto fmt_db = [](auto &&k) { return variant_key_type(k); };
+    ARCTICDB_SAMPLE(MongoStorageRemove, 0)
+
+    (fg::from(ks.as_range()) | fg::move | fg::groupBy(fmt_db)).foreach([&](auto &&group) {
+        for (auto &k : group.values()) {
+            auto collection = collection_name(variant_key_type(k));
+            client_->remove_keyvalue(db_, collection, k);
+        }
+    });
+}
+
+void MongoStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string &prefix) {
+    auto collection = collection_name(key_type);
+    auto func = folly::Function<void(entity::VariantKey&&)>(visitor);
+    ARCTICDB_SAMPLE(MongoStorageItType, 0)
+    client_->iterate_type(db_, collection, key_type, std::move(func), prefix);
+}
+
+bool MongoStorage::do_key_exists(const VariantKey& key) {
+    auto collection = collection_name(variant_key_type(key));
+    return client_->key_exists(db_, collection, key);
+}
+
+
 using Config = arcticdb::proto::mongo_storage::Config;
 
 MongoStorage::MongoStorage(
diff --git a/cpp/arcticdb/storage/s3/nfs_backed_storage-inl.hpp b/cpp/arcticdb/storage/s3/nfs_backed_storage-inl.hpp
deleted file mode 100644
index 17581d5789..0000000000
--- a/cpp/arcticdb/storage/s3/nfs_backed_storage-inl.hpp
+++ /dev/null
@@ -1,181 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * 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.
- */
-
-#ifndef ARCTICDB_NFS_BACKED_STORAGE_H_
-#error "This should only be included by nfs_backed_storage.cpp"
-#endif
-
-#include <arcticdb/util/simple_string_hash.hpp>
-#include <arcticdb/storage/s3/nfs_backed_storage.hpp>
-#include <arcticdb/storage/s3/s3_storage.hpp>
-
-namespace arcticdb::storage::nfs_backed {
-
-inline std::string add_suffix_char(const std::string& str) {
-    return fmt::format("{}*", str);
-}
-
-inline std::string remove_suffix_char(const std::string& str) {
-    return str.substr(0, str.size()-1);
-}
-
-template<typename MixedType, typename StringType, typename NumericType>
-MixedType encode_item(const MixedType& input, bool add_suffix) {
-    return util::variant_match(input,
-    [add_suffix] (const StringType& str) {
-        const auto encoded = util::safe_encode(str);
-        return add_suffix ? MixedType{add_suffix_char(encoded)} : MixedType{encoded};
-    },
-    [](const NumericType& id) {
-        return MixedType{id};
-    });
-}
-
-template<typename MixedType, typename StringType, typename NumericType>
-MixedType decode_item(const MixedType& input, bool remove_suffix) {
-    return util::variant_match(input,
-    [remove_suffix] (const StringType& str) {
-        return MixedType{util::safe_decode(remove_suffix ? remove_suffix_char(str) : str)};
-    },
-    [](const NumericType& id) {
-       return MixedType{id};
-    });
-}
-
-inline VariantKey encode_object_id(const VariantKey& key) {
-    return util::variant_match(key,
-        [] (const AtomKey& k) {
-            auto encoded_id = encode_item<StreamId, StringId, NumericId>(k.id(), false);
-            auto start_index = encode_item<IndexValue, StringIndex, NumericIndex>(k.start_index(), false);
-            auto end_index = encode_item<IndexValue, StringIndex, NumericIndex>(k.end_index(), false);
-            return VariantKey{atom_key_builder().version_id(k.version_id()).start_index(start_index)
-            .end_index(end_index).creation_ts(k.creation_ts()).content_hash(k.content_hash())
-            .build(encoded_id, k.type())};
-    },
-        [](const RefKey& r) {
-        auto encoded_id = encode_item<StreamId, StringId, NumericId>(r.id(), true);
-        return VariantKey{RefKey{encoded_id, r.type(), r.is_old_type()}};
-    });
-}
-
-inline uint32_t id_to_number(const StreamId& stream_id) {
-    return util::variant_match(stream_id,
-       [] (const NumericId& num_id) { return static_cast<uint32_t>(num_id); },
-       [] (const StringId& str_id) { return murmur3_32(str_id); });
-}
-
-inline uint32_t get_id_bucket(const StreamId& id) {
-    return id_to_number(id) % 1000;
-}
-
-inline uint32_t get_hash_bucket(const AtomKey& atom) {
-    return atom.content_hash() % 1000;
-}
-
-inline std::string get_root_folder(const std::string& root_folder, const RefKey& ref) {
-    const auto id_bucket = get_id_bucket(ref.id());
-    return fmt::format("{}/{:03}", root_folder, id_bucket);
-}
-
-inline std::string get_root_folder(const std::string& root_folder, const AtomKey& atom) {
-    const auto id_bucket = get_id_bucket(atom.id());
-    const auto hash_bucket = get_hash_bucket(atom);
-    return fmt::format("{}/{:03}/{:03}", root_folder, id_bucket, hash_bucket);
-}
-
-inline std::string get_root_folder(const std::string& root_folder, const VariantKey& vk) {
-    return util::variant_match(vk, [&root_folder] (const auto& k) {
-        return get_root_folder(root_folder, k);
-    });
-}
-
-struct NfsBucketizer {
-    static std::string bucketize(const std::string& root_folder, const VariantKey& vk) {
-        return get_root_folder(root_folder, vk);
-    }
-
-    static size_t bucketize_length(KeyType key_type) {
-        return is_ref_key_class(key_type) ? 4 : 8;
-    }
-};
-
-inline VariantKey unencode_object_id(const VariantKey& key) {
-
-    return util::variant_match(key,
-                        [] (const AtomKey& k) {
-                            auto decoded_id = decode_item<StreamId, StringId, NumericId>(k.id(), false);
-                            auto start_index = decode_item<IndexValue, StringIndex, NumericIndex>(k.start_index(), false);
-                            auto end_index = decode_item<IndexValue, StringIndex, NumericIndex>(k.end_index(), false);
-                            return VariantKey{atom_key_builder().version_id(k.version_id()).start_index(start_index)
-                                .end_index(end_index).creation_ts(k.creation_ts()).content_hash(k.content_hash())
-                                .build(decoded_id, k.type())};
-                        },
-                        [](const RefKey& r) {
-                            auto decoded_id = decode_item<StreamId, StringId, NumericId>(r.id(), true);
-                            return VariantKey{RefKey{decoded_id, r.type(), r.is_old_type()}};
-                        });
-}
-
-inline void NfsBackedStorage::do_write(Composite<KeySegmentPair>&& kvs) {
-    auto enc = kvs.transform([] (auto&& key_seg) {
-        return KeySegmentPair{encode_object_id(key_seg.variant_key()), std::move(key_seg.segment())};
-    });
-    s3::detail::do_write_impl(std::move(enc), root_folder_, bucket_name_, s3_client_, NfsBucketizer{});
-}
-
-inline void NfsBackedStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts) {
-    auto enc = kvs.transform([] (auto&& key_seg) {
-        return KeySegmentPair{encode_object_id(key_seg.variant_key()), std::move(key_seg.segment())};
-    });
-    s3::detail::do_update_impl(std::move(enc), root_folder_, bucket_name_, s3_client_, NfsBucketizer{});
-}
-
-inline void NfsBackedStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts opts) {
-    auto func = [visitor] (const VariantKey& k, Segment&& seg) mutable {
-        visitor(unencode_object_id(k), std::move(seg));
-    };
-
-    auto enc = ks.transform([] (auto&& key) {
-        return encode_object_id(key);
-    });
-
-    s3::detail::do_read_impl(std::move(enc), func, root_folder_, bucket_name_, s3_client_, NfsBucketizer{}, opts);
-}
-
-inline void NfsBackedStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
-    auto enc = ks.transform([] (auto&& key) {
-        return encode_object_id(key);
-    });
-    s3::detail::do_remove_impl(std::move(enc), root_folder_, bucket_name_, s3_client_, NfsBucketizer{});
-}
-
-inline void NfsBackedStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string& prefix) {
-    auto func = [v = std::move(visitor), prefix=prefix] (VariantKey&& k) mutable {
-        auto key = unencode_object_id(k);
-        if(prefix.empty() || variant_key_id(key) == StreamId{prefix})
-            v(std::move(key));
-    };
-
-    auto prefix_handler = [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor&, KeyType key_type) {
-        std::string new_prefix;
-        if(!prefix.empty()) {
-            uint32_t id = get_id_bucket(encode_item<StreamId, StringId, NumericId>(StringId{prefix}, is_ref_key_class(key_type)));
-            new_prefix = fmt::format("{:03}", id);
-        }
-
-        return !prefix.empty() ? fmt::format("{}/{}", key_type_dir, new_prefix) : key_type_dir;
-    };
-
-    s3::detail::do_iterate_type_impl(key_type, std::move(func), root_folder_, bucket_name_, s3_client_, NfsBucketizer{}, prefix_handler, prefix);
-}
-
-inline bool NfsBackedStorage::do_key_exists(const VariantKey& key) {
-    auto encoded_key = encode_object_id(key);
-    return s3::detail::do_key_exists_impl(encoded_key, root_folder_, bucket_name_, s3_client_, NfsBucketizer{});
-}
-
-} //namespace arcticdb::nfs_backed
\ No newline at end of file
diff --git a/cpp/arcticdb/storage/s3/nfs_backed_storage.cpp b/cpp/arcticdb/storage/s3/nfs_backed_storage.cpp
index a6bc0b1ab3..68717a15fd 100644
--- a/cpp/arcticdb/storage/s3/nfs_backed_storage.cpp
+++ b/cpp/arcticdb/storage/s3/nfs_backed_storage.cpp
@@ -6,11 +6,175 @@
  */
 
 #include <arcticdb/storage/s3/nfs_backed_storage.hpp>
-#define ARCTICDB_NFS_BACKED_STORAGE_H_
-#include <arcticdb/storage/s3/nfs_backed_storage-inl.hpp>
+
+#include <arcticdb/util/simple_string_hash.hpp>
+#include <arcticdb/storage/s3/s3_storage.hpp>
 
 namespace arcticdb::storage::nfs_backed {
 
+std::string add_suffix_char(const std::string& str) {
+    return fmt::format("{}*", str);
+}
+
+std::string remove_suffix_char(const std::string& str) {
+    return str.substr(0, str.size()-1);
+}
+
+template<typename MixedType, typename StringType, typename NumericType>
+MixedType encode_item(const MixedType& input, bool add_suffix) {
+    return util::variant_match(input,
+    [add_suffix] (const StringType& str) {
+        const auto encoded = util::safe_encode(str);
+        return add_suffix ? MixedType{add_suffix_char(encoded)} : MixedType{encoded};
+    },
+    [](const NumericType& id) {
+        return MixedType{id};
+    });
+}
+
+template<typename MixedType, typename StringType, typename NumericType>
+MixedType decode_item(const MixedType& input, bool remove_suffix) {
+    return util::variant_match(input,
+    [remove_suffix] (const StringType& str) {
+        return MixedType{util::safe_decode(remove_suffix ? remove_suffix_char(str) : str)};
+    },
+    [](const NumericType& id) {
+       return MixedType{id};
+    });
+}
+
+VariantKey encode_object_id(const VariantKey& key) {
+    return util::variant_match(key,
+        [] (const AtomKey& k) {
+            auto encoded_id = encode_item<StreamId, StringId, NumericId>(k.id(), false);
+            auto start_index = encode_item<IndexValue, StringIndex, NumericIndex>(k.start_index(), false);
+            auto end_index = encode_item<IndexValue, StringIndex, NumericIndex>(k.end_index(), false);
+            return VariantKey{atom_key_builder().version_id(k.version_id()).start_index(start_index)
+            .end_index(end_index).creation_ts(k.creation_ts()).content_hash(k.content_hash())
+            .build(encoded_id, k.type())};
+    },
+        [](const RefKey& r) {
+        auto encoded_id = encode_item<StreamId, StringId, NumericId>(r.id(), true);
+        return VariantKey{RefKey{encoded_id, r.type(), r.is_old_type()}};
+    });
+}
+
+uint32_t id_to_number(const StreamId& stream_id) {
+    return util::variant_match(stream_id,
+       [] (const NumericId& num_id) { return static_cast<uint32_t>(num_id); },
+       [] (const StringId& str_id) { return murmur3_32(str_id); });
+}
+
+uint32_t get_id_bucket(const StreamId& id) {
+    return id_to_number(id) % 1000;
+}
+
+uint32_t get_hash_bucket(const AtomKey& atom) {
+    return atom.content_hash() % 1000;
+}
+
+std::string get_root_folder(const std::string& root_folder, const RefKey& ref) {
+    const auto id_bucket = get_id_bucket(ref.id());
+    return fmt::format("{}/{:03}", root_folder, id_bucket);
+}
+
+std::string get_root_folder(const std::string& root_folder, const AtomKey& atom) {
+    const auto id_bucket = get_id_bucket(atom.id());
+    const auto hash_bucket = get_hash_bucket(atom);
+    return fmt::format("{}/{:03}/{:03}", root_folder, id_bucket, hash_bucket);
+}
+
+std::string get_root_folder(const std::string& root_folder, const VariantKey& vk) {
+    return util::variant_match(vk, [&root_folder] (const auto& k) {
+        return get_root_folder(root_folder, k);
+    });
+}
+
+
+std::string NfsBucketizer::bucketize(const std::string& root_folder, const VariantKey& vk) {
+        return get_root_folder(root_folder, vk);
+    }
+
+size_t NfsBucketizer::bucketize_length(KeyType key_type) {
+        return is_ref_key_class(key_type) ? 4 : 8;
+    }
+
+VariantKey unencode_object_id(const VariantKey& key) {
+
+    return util::variant_match(key,
+                        [] (const AtomKey& k) {
+                            auto decoded_id = decode_item<StreamId, StringId, NumericId>(k.id(), false);
+                            auto start_index = decode_item<IndexValue, StringIndex, NumericIndex>(k.start_index(), false);
+                            auto end_index = decode_item<IndexValue, StringIndex, NumericIndex>(k.end_index(), false);
+                            return VariantKey{atom_key_builder().version_id(k.version_id()).start_index(start_index)
+                                .end_index(end_index).creation_ts(k.creation_ts()).content_hash(k.content_hash())
+                                .build(decoded_id, k.type())};
+                        },
+                        [](const RefKey& r) {
+                            auto decoded_id = decode_item<StreamId, StringId, NumericId>(r.id(), true);
+                            return VariantKey{RefKey{decoded_id, r.type(), r.is_old_type()}};
+                        });
+}
+
+void NfsBackedStorage::do_write(Composite<KeySegmentPair>&& kvs) {
+    auto enc = kvs.transform([] (auto&& key_seg) {
+        return KeySegmentPair{encode_object_id(key_seg.variant_key()), std::move(key_seg.segment())};
+    });
+    s3::detail::do_write_impl(std::move(enc), root_folder_, bucket_name_, s3_client_, NfsBucketizer{});
+}
+
+void NfsBackedStorage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts) {
+    auto enc = kvs.transform([] (auto&& key_seg) {
+        return KeySegmentPair{encode_object_id(key_seg.variant_key()), std::move(key_seg.segment())};
+    });
+    s3::detail::do_update_impl(std::move(enc), root_folder_, bucket_name_, s3_client_, NfsBucketizer{});
+}
+
+void NfsBackedStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts opts) {
+    auto func = [visitor] (const VariantKey& k, Segment&& seg) mutable {
+        visitor(unencode_object_id(k), std::move(seg));
+    };
+
+    auto enc = ks.transform([] (auto&& key) {
+        return encode_object_id(key);
+    });
+
+    s3::detail::do_read_impl(std::move(enc), func, root_folder_, bucket_name_, s3_client_, NfsBucketizer{}, opts);
+}
+
+void NfsBackedStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
+    auto enc = ks.transform([] (auto&& key) {
+        return encode_object_id(key);
+    });
+    s3::detail::do_remove_impl(std::move(enc), root_folder_, bucket_name_, s3_client_, NfsBucketizer{});
+}
+
+void NfsBackedStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string& prefix) {
+    auto func = [v = std::move(visitor), prefix=prefix] (VariantKey&& k) mutable {
+        auto key = unencode_object_id(k);
+        if(prefix.empty() || variant_key_id(key) == StreamId{prefix})
+            v(std::move(key));
+    };
+
+    auto prefix_handler = [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor&, KeyType key_type) {
+        std::string new_prefix;
+        if(!prefix.empty()) {
+            uint32_t id = get_id_bucket(encode_item<StreamId, StringId, NumericId>(StringId{prefix}, is_ref_key_class(key_type)));
+            new_prefix = fmt::format("{:03}", id);
+        }
+
+        return !prefix.empty() ? fmt::format("{}/{}", key_type_dir, new_prefix) : key_type_dir;
+    };
+
+    s3::detail::do_iterate_type_impl(key_type, std::move(func), root_folder_, bucket_name_, s3_client_, NfsBucketizer{}, prefix_handler, prefix);
+}
+
+bool NfsBackedStorage::do_key_exists(const VariantKey& key) {
+    auto encoded_key = encode_object_id(key);
+    return s3::detail::do_key_exists_impl(encoded_key, root_folder_, bucket_name_, s3_client_, NfsBucketizer{});
+}
+
+
 NfsBackedStorage::NfsBackedStorage(const LibraryPath &library_path, OpenMode mode, const Config &conf) :
     Storage(library_path, mode),
     s3_api_(s3::S3ApiInstance::instance()),
diff --git a/cpp/arcticdb/storage/s3/nfs_backed_storage.hpp b/cpp/arcticdb/storage/s3/nfs_backed_storage.hpp
index f71a1b57af..6e9de4b027 100644
--- a/cpp/arcticdb/storage/s3/nfs_backed_storage.hpp
+++ b/cpp/arcticdb/storage/s3/nfs_backed_storage.hpp
@@ -90,4 +90,11 @@ inline arcticdb::proto::storage::VariantStorage pack_config(
     return output;
 }
 
+
+struct NfsBucketizer {
+    static std::string bucketize(const std::string& root_folder, const VariantKey& vk);
+    static size_t bucketize_length(KeyType key_type);
+};
+
+
 } //namespace arcticdb::nfs_backed
diff --git a/cpp/arcticdb/storage/s3/s3_storage-inl.hpp b/cpp/arcticdb/storage/s3/s3_storage-inl.hpp
deleted file mode 100644
index b3fe2535e0..0000000000
--- a/cpp/arcticdb/storage/s3/s3_storage-inl.hpp
+++ /dev/null
@@ -1,93 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * 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.
- */
-
-#ifndef ARCTICDB_S3_STORAGE_H_
-#error "This should only be included by s3_storage.cpp"
-#endif
-
-#include <arcticdb/util/preconditions.hpp>
-#include <arcticdb/util/pb_util.hpp>
-#include <arcticdb/log/log.hpp>
-#include <arcticdb/util/buffer_pool.hpp>
-
-#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
-#include <folly/gen/Base.h>
-#include <arcticdb/storage/s3/s3_storage.hpp>
-#include <arcticdb/storage/object_store_utils.hpp>
-#include <arcticdb/storage/storage_options.hpp>
-#include <arcticdb/storage/storage_utils.hpp>
-#include <arcticdb/entity/serialized_key.hpp>
-#include <arcticdb/util/exponential_backoff.hpp>
-#include <arcticdb/util/configs_map.hpp>
-#include <arcticdb/util/composite.hpp>
-
-#include <aws/s3/model/GetObjectRequest.h>
-#include <aws/s3/model/PutObjectRequest.h>
-#include <aws/s3/model/DeleteObjectRequest.h>
-#include <aws/s3/model/DeleteObjectsRequest.h>
-#include <aws/s3/model/ListObjectsV2Request.h>
-#include <aws/s3/model/HeadObjectRequest.h>
-#include <aws/s3/model/Object.h>
-#include <aws/s3/model/Delete.h>
-#include <aws/s3/model/ObjectIdentifier.h>
-
-#include <boost/interprocess/streams/bufferstream.hpp>
-#include <folly/ThreadLocal.h>
-
-#include <arcticdb/storage/s3/detail-inl.hpp>
-
-#undef GetMessage
-
-namespace arcticdb::storage {
-
-using namespace object_store_utils;
-
-namespace s3 {
-
-namespace fg = folly::gen;
-
-
-inline std::string S3Storage::get_key_path(const VariantKey& key) const {
-    auto b = FlatBucketizer{};
-    auto key_type_dir = key_type_folder(root_folder_, variant_key_type(key));
-    return object_path(b.bucketize(key_type_dir, key), key);
-    // FUTURE: if we can reuse `detail::do_*` below doing polymorphism instead of templating, then this method is useful
-    // to most of them
-}
-
-inline void S3Storage::do_write(Composite<KeySegmentPair>&& kvs) {
-    detail::do_write_impl(std::move(kvs), root_folder_, bucket_name_, s3_client_, FlatBucketizer{});
-}
-
-inline void S3Storage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts) {
-    detail::do_update_impl(std::move(kvs), root_folder_, bucket_name_, s3_client_, FlatBucketizer{});
-}
-
-inline void S3Storage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts opts) {
-    detail::do_read_impl(std::move(ks), visitor, root_folder_, bucket_name_, s3_client_, FlatBucketizer{}, opts);
-}
-
-inline void S3Storage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
-    detail::do_remove_impl(std::move(ks), root_folder_, bucket_name_, s3_client_, FlatBucketizer{});
-}
-
-inline void S3Storage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string& prefix) {
-    auto prefix_handler = [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor key_descriptor, KeyType) {
-        return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
-    };
-
-    detail::do_iterate_type_impl(key_type, std::move(visitor), root_folder_, bucket_name_, s3_client_, FlatBucketizer{}, std::move(prefix_handler), prefix);
-}
-
-inline bool S3Storage::do_key_exists(const VariantKey& key) {
-    return detail::do_key_exists_impl(key, root_folder_, bucket_name_, s3_client_, FlatBucketizer{});
-}
-
-
-} // namespace s3
-
-} // namespace arcticdb::storage
diff --git a/cpp/arcticdb/storage/s3/s3_storage.cpp b/cpp/arcticdb/storage/s3/s3_storage.cpp
index 4a057fdc7c..22c7702301 100644
--- a/cpp/arcticdb/storage/s3/s3_storage.cpp
+++ b/cpp/arcticdb/storage/s3/s3_storage.cpp
@@ -5,12 +5,94 @@
  * 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.
  */
 
-#define ARCTICDB_S3_STORAGE_H_
-#include <arcticdb/storage/s3/s3_storage-inl.hpp>
+#include <arcticdb/storage/s3/s3_storage.hpp>
+
+#include <locale>
+
+#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
+#include <folly/gen/Base.h>
 #include <aws/core/auth/AWSCredentialsProvider.h>
+
+#include <arcticdb/log/log.hpp>
 #include <arcticdb/storage/s3/s3_api.hpp>
+#include <arcticdb/util/preconditions.hpp>
+#include <arcticdb/util/pb_util.hpp>
 #include <arcticdb/log/log.hpp>
-#include <locale>
+#include <arcticdb/util/buffer_pool.hpp>
+#include <arcticdb/storage/object_store_utils.hpp>
+#include <arcticdb/storage/storage_options.hpp>
+#include <arcticdb/storage/storage_utils.hpp>
+#include <arcticdb/entity/serialized_key.hpp>
+#include <arcticdb/util/exponential_backoff.hpp>
+#include <arcticdb/util/configs_map.hpp>
+#include <arcticdb/util/composite.hpp>
+
+#include <aws/s3/model/GetObjectRequest.h>
+#include <aws/s3/model/PutObjectRequest.h>
+#include <aws/s3/model/DeleteObjectRequest.h>
+#include <aws/s3/model/DeleteObjectsRequest.h>
+#include <aws/s3/model/ListObjectsV2Request.h>
+#include <aws/s3/model/HeadObjectRequest.h>
+#include <aws/s3/model/Object.h>
+#include <aws/s3/model/Delete.h>
+#include <aws/s3/model/ObjectIdentifier.h>
+
+#include <boost/interprocess/streams/bufferstream.hpp>
+#include <folly/ThreadLocal.h>
+
+#include <arcticdb/storage/s3/detail-inl.hpp>
+
+#undef GetMessage
+
+namespace arcticdb::storage {
+
+using namespace object_store_utils;
+
+namespace s3 {
+
+namespace fg = folly::gen;
+
+
+std::string S3Storage::get_key_path(const VariantKey& key) const {
+    auto b = FlatBucketizer{};
+    auto key_type_dir = key_type_folder(root_folder_, variant_key_type(key));
+    return object_path(b.bucketize(key_type_dir, key), key);
+    // FUTURE: if we can reuse `detail::do_*` below doing polymorphism instead of templating, then this method is useful
+    // to most of them
+}
+
+void S3Storage::do_write(Composite<KeySegmentPair>&& kvs) {
+    detail::do_write_impl(std::move(kvs), root_folder_, bucket_name_, s3_client_, FlatBucketizer{});
+}
+
+void S3Storage::do_update(Composite<KeySegmentPair>&& kvs, UpdateOpts) {
+    detail::do_update_impl(std::move(kvs), root_folder_, bucket_name_, s3_client_, FlatBucketizer{});
+}
+
+void S3Storage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visitor, ReadKeyOpts opts) {
+    detail::do_read_impl(std::move(ks), visitor, root_folder_, bucket_name_, s3_client_, FlatBucketizer{}, opts);
+}
+
+void S3Storage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
+    detail::do_remove_impl(std::move(ks), root_folder_, bucket_name_, s3_client_, FlatBucketizer{});
+}
+
+void S3Storage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string& prefix) {
+    auto prefix_handler = [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor key_descriptor, KeyType) {
+        return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
+    };
+
+    detail::do_iterate_type_impl(key_type, std::move(visitor), root_folder_, bucket_name_, s3_client_, FlatBucketizer{}, std::move(prefix_handler), prefix);
+}
+
+bool S3Storage::do_key_exists(const VariantKey& key) {
+    return detail::do_key_exists_impl(key, root_folder_, bucket_name_, s3_client_, FlatBucketizer{});
+}
+
+
+} // namespace s3
+} // namespace arcticdb::storage
+
 
 namespace arcticdb::storage::s3 {
 
diff --git a/cpp/arcticdb/storage/s3/s3_storage_tool-inl.hpp b/cpp/arcticdb/storage/s3/s3_storage_tool-inl.hpp
deleted file mode 100644
index 236b3a8f9b..0000000000
--- a/cpp/arcticdb/storage/s3/s3_storage_tool-inl.hpp
+++ /dev/null
@@ -1,194 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * 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.
- */
-
-#ifndef ARCTICDB_S3_STORAGE_TOOL_H_
-#error "This should only be included by file_storage.hpp"
-#endif
-
-#include <aws/s3/model/ListObjectsV2Request.h>
-#include <aws/s3/model/GetObjectRequest.h>
-#include <aws/s3/model/DeleteObjectRequest.h>
-#include <aws/s3/model/HeadObjectRequest.h>
-
-namespace arcticdb::storage::s3 {
-
-template<class Visitor>
-void S3StorageTool::iterate_bucket(Visitor&& visitor, const std::string& prefix) {
-    Aws::S3::Model::ListObjectsV2Request objects_request;
-    objects_request.WithBucket(bucket_name_.c_str());
-    if (!prefix.empty())
-        objects_request.SetPrefix(prefix.c_str());
-
-    bool more;
-    do {
-        auto list_objects_outcome = s3_client_.ListObjectsV2(objects_request);
-
-        if (list_objects_outcome.IsSuccess()) {
-            const auto& object_list = list_objects_outcome.GetResult().GetContents();
-            for (auto const& s3_object : object_list) {
-                const auto& key = s3_object.GetKey();
-                visitor(key.c_str());
-            }
-            more = list_objects_outcome.GetResult().GetIsTruncated();
-            if (more)
-                objects_request.SetContinuationToken(list_objects_outcome.GetResult().GetNextContinuationToken());
-
-        }
-        else {
-            const auto& error = list_objects_outcome.GetError();
-            log::storage().error("Failed to iterate bucket '{}' {}:{}",
-                                 bucket_name_,
-                                 error.GetExceptionName().c_str(),
-                                 error.GetMessage().c_str());
-            return;
-        }
-    } while (more);
-}
-
-inline void S3StorageTool::set_object(const std::string& key, const std::string& data) {
-    Aws::S3::Model::PutObjectRequest object_request;
-    object_request.SetBucket(bucket_name_.c_str());
-    object_request.SetKey(key.c_str());
-    ARCTICDB_TRACE(log::storage(), "Set s3 key {}", object_request.GetKey().c_str());
-
-    auto body = std::make_shared<Aws::StringStream>();
-    body->write(data.data(), data.size());
-    object_request.SetBody(body);
-
-    auto put_object_outcome = s3_client_.PutObject(object_request);
-    if (!put_object_outcome.IsSuccess()) {
-        auto& error = put_object_outcome.GetError();
-        util::raise_rte("Failed to write s3 with key '{}' {}: {}",
-                        key,
-                        error.GetExceptionName().c_str(),
-                        error.GetMessage().c_str());
-    }
-    ARCTICDB_DEBUG(log::storage(), "Wrote key {} with {} bytes of data",
-                         key,
-                         data.size());
-}
-
-inline std::string S3StorageTool::get_object(const std::string& key) {
-    Aws::S3::Model::GetObjectRequest object_request;
-
-    object_request.SetBucket(bucket_name_.c_str());
-    object_request.SetKey(key.c_str());
-    auto get_object_outcome = s3_client_.GetObject(object_request);
-
-    if (get_object_outcome.IsSuccess()) {
-        auto& retrieved = get_object_outcome.GetResult().GetBody();
-        auto vec = storage::stream_to_vector(retrieved);
-        return std::string(vec.data(), vec.size());
-    }
-    else {
-        auto& error = get_object_outcome.GetError();
-        log::storage().warn("Failed to find data for key '{}' {}: {}",
-                            key,
-                            error.GetExceptionName().c_str(),
-                            error.GetMessage().c_str());
-        return std::string();
-    }
-}
-
-inline void S3StorageTool::delete_object(const std::string& key) {
-    Aws::S3::Model::DeleteObjectRequest object_request;
-    object_request.WithBucket(bucket_name_.c_str()).WithKey(key.c_str());
-
-    auto delete_object_outcome = s3_client_.DeleteObject(object_request);
-    if (delete_object_outcome.IsSuccess())
-        ARCTICDB_DEBUG(log::storage(), "Deleted object with key '{}'", key);
-    else {
-        const auto& error = delete_object_outcome.GetError();
-        log::storage().warn("Failed to delete segment with key '{}': {}",
-                            key,
-                            error.GetExceptionName().c_str(),
-                            error.GetMessage().c_str());
-    }
-}
-
-inline std::vector<std::string> S3StorageTool::list_bucket(const std::string& prefix) {
-    std::vector<std::string> objects;
-    iterate_bucket([&](const std::string& str) { objects.push_back(str); }, prefix);
-    return objects;
-}
-
-inline std::pair<size_t, size_t> S3StorageTool::get_prefix_info(const std::string& prefix) {
-    size_t total_size = 0;
-    size_t total_items = 0;
-    Aws::S3::Model::ListObjectsV2Request objects_request;
-    objects_request.WithBucket(bucket_name_.c_str());
-    if (!prefix.empty())
-        objects_request.SetPrefix(prefix.c_str());
-
-    bool more;
-    do {
-        auto list_objects_outcome = s3_client_.ListObjectsV2(objects_request);
-
-        if (list_objects_outcome.IsSuccess()) {
-            const auto& object_list = list_objects_outcome.GetResult().GetContents();
-            for (auto const& s3_object : object_list) {
-                total_size += s3_object.GetSize();
-                ++total_items;
-            }
-            more = list_objects_outcome.GetResult().GetIsTruncated();
-            if (more)
-                objects_request.SetContinuationToken(list_objects_outcome.GetResult().GetNextContinuationToken());
-
-        }
-        else {
-            const auto& error = list_objects_outcome.GetError();
-            log::storage().error("Failed to iterate bucket to get sizes'{}' {}:{}",
-                                 bucket_name_,
-                                 error.GetExceptionName().c_str(),
-                                 error.GetMessage().c_str());
-            return {0, 0};
-        }
-    } while (more);
-    return {total_size, total_items};
-}
-
-inline size_t S3StorageTool::get_file_size(const std::string& key) {
-    Aws::S3::Model::HeadObjectRequest head_request;
-    head_request.SetBucket(bucket_name_.c_str());
-    head_request.SetKey(key.c_str());
-
-    auto object = s3_client_.HeadObject(head_request);
-    if (object.IsSuccess())
-    {
-        auto file_sz = object.GetResultWithOwnership().GetContentLength();
-        ARCTICDB_TRACE(log::storage(), "Size of {}: {}", key, file_sz);
-        return file_sz;
-    }
-    else
-    {
-        log::storage().error("Head Object error:  {} - {}", object .GetError().GetExceptionName(),
-                             object .GetError().GetMessage());
-        return 0;
-    }
-}
-
-
-inline void S3StorageTool::delete_bucket(const std::string& prefix) {
-    iterate_bucket([&](const std::string& key) {
-                       Aws::S3::Model::DeleteObjectRequest object_request;
-                       object_request.WithBucket(bucket_name_.c_str()).WithKey(key.c_str());
-
-                       auto delete_object_outcome = s3_client_.DeleteObject(object_request);
-                       if (delete_object_outcome.IsSuccess())
-                           ARCTICDB_DEBUG(log::storage(), "Deleted object with key '{}'", key);
-                       else {
-                           const auto& error = delete_object_outcome.GetError();
-                           log::storage().warn("Failed to delete object with key '{}' {}:{}",
-                                               key,
-                                               error.GetExceptionName().c_str(),
-                                               error.GetMessage().c_str());
-                       }
-                   },
-                   prefix);
-}
-
-} // namespace arcticdb::storage::s3
diff --git a/cpp/arcticdb/storage/s3/s3_storage_tool.cpp b/cpp/arcticdb/storage/s3/s3_storage_tool.cpp
index cd39fcf172..7d16179505 100644
--- a/cpp/arcticdb/storage/s3/s3_storage_tool.cpp
+++ b/cpp/arcticdb/storage/s3/s3_storage_tool.cpp
@@ -7,8 +7,190 @@
 
 #include <arcticdb/storage/s3/s3_storage_tool.hpp>
 
+
+#include <aws/s3/model/ListObjectsV2Request.h>
+#include <aws/s3/model/GetObjectRequest.h>
+#include <aws/s3/model/DeleteObjectRequest.h>
+#include <aws/s3/model/HeadObjectRequest.h>
+
 namespace arcticdb::storage::s3 {
 
+template<class Visitor>
+void S3StorageTool::iterate_bucket(Visitor&& visitor, const std::string& prefix) {
+    Aws::S3::Model::ListObjectsV2Request objects_request;
+    objects_request.WithBucket(bucket_name_.c_str());
+    if (!prefix.empty())
+        objects_request.SetPrefix(prefix.c_str());
+
+    bool more;
+    do {
+        auto list_objects_outcome = s3_client_.ListObjectsV2(objects_request);
+
+        if (list_objects_outcome.IsSuccess()) {
+            const auto& object_list = list_objects_outcome.GetResult().GetContents();
+            for (auto const& s3_object : object_list) {
+                const auto& key = s3_object.GetKey();
+                visitor(key.c_str());
+            }
+            more = list_objects_outcome.GetResult().GetIsTruncated();
+            if (more)
+                objects_request.SetContinuationToken(list_objects_outcome.GetResult().GetNextContinuationToken());
+
+        }
+        else {
+            const auto& error = list_objects_outcome.GetError();
+            log::storage().error("Failed to iterate bucket '{}' {}:{}",
+                                 bucket_name_,
+                                 error.GetExceptionName().c_str(),
+                                 error.GetMessage().c_str());
+            return;
+        }
+    } while (more);
+}
+
+void S3StorageTool::set_object(const std::string& key, const std::string& data) {
+    Aws::S3::Model::PutObjectRequest object_request;
+    object_request.SetBucket(bucket_name_.c_str());
+    object_request.SetKey(key.c_str());
+    ARCTICDB_TRACE(log::storage(), "Set s3 key {}", object_request.GetKey().c_str());
+
+    auto body = std::make_shared<Aws::StringStream>();
+    body->write(data.data(), data.size());
+    object_request.SetBody(body);
+
+    auto put_object_outcome = s3_client_.PutObject(object_request);
+    if (!put_object_outcome.IsSuccess()) {
+        auto& error = put_object_outcome.GetError();
+        util::raise_rte("Failed to write s3 with key '{}' {}: {}",
+                        key,
+                        error.GetExceptionName().c_str(),
+                        error.GetMessage().c_str());
+    }
+    ARCTICDB_DEBUG(log::storage(), "Wrote key {} with {} bytes of data",
+                         key,
+                         data.size());
+}
+
+std::string S3StorageTool::get_object(const std::string& key) {
+    Aws::S3::Model::GetObjectRequest object_request;
+
+    object_request.SetBucket(bucket_name_.c_str());
+    object_request.SetKey(key.c_str());
+    auto get_object_outcome = s3_client_.GetObject(object_request);
+
+    if (get_object_outcome.IsSuccess()) {
+        auto& retrieved = get_object_outcome.GetResult().GetBody();
+        auto vec = storage::stream_to_vector(retrieved);
+        return std::string(vec.data(), vec.size());
+    }
+    else {
+        auto& error = get_object_outcome.GetError();
+        log::storage().warn("Failed to find data for key '{}' {}: {}",
+                            key,
+                            error.GetExceptionName().c_str(),
+                            error.GetMessage().c_str());
+        return std::string();
+    }
+}
+
+void S3StorageTool::delete_object(const std::string& key) {
+    Aws::S3::Model::DeleteObjectRequest object_request;
+    object_request.WithBucket(bucket_name_.c_str()).WithKey(key.c_str());
+
+    auto delete_object_outcome = s3_client_.DeleteObject(object_request);
+    if (delete_object_outcome.IsSuccess())
+        ARCTICDB_DEBUG(log::storage(), "Deleted object with key '{}'", key);
+    else {
+        const auto& error = delete_object_outcome.GetError();
+        log::storage().warn("Failed to delete segment with key '{}': {}",
+                            key,
+                            error.GetExceptionName().c_str(),
+                            error.GetMessage().c_str());
+    }
+}
+
+std::vector<std::string> S3StorageTool::list_bucket(const std::string& prefix) {
+    std::vector<std::string> objects;
+    iterate_bucket([&](const std::string& str) { objects.push_back(str); }, prefix);
+    return objects;
+}
+
+std::pair<size_t, size_t> S3StorageTool::get_prefix_info(const std::string& prefix) {
+    size_t total_size = 0;
+    size_t total_items = 0;
+    Aws::S3::Model::ListObjectsV2Request objects_request;
+    objects_request.WithBucket(bucket_name_.c_str());
+    if (!prefix.empty())
+        objects_request.SetPrefix(prefix.c_str());
+
+    bool more;
+    do {
+        auto list_objects_outcome = s3_client_.ListObjectsV2(objects_request);
+
+        if (list_objects_outcome.IsSuccess()) {
+            const auto& object_list = list_objects_outcome.GetResult().GetContents();
+            for (auto const& s3_object : object_list) {
+                total_size += s3_object.GetSize();
+                ++total_items;
+            }
+            more = list_objects_outcome.GetResult().GetIsTruncated();
+            if (more)
+                objects_request.SetContinuationToken(list_objects_outcome.GetResult().GetNextContinuationToken());
+
+        }
+        else {
+            const auto& error = list_objects_outcome.GetError();
+            log::storage().error("Failed to iterate bucket to get sizes'{}' {}:{}",
+                                 bucket_name_,
+                                 error.GetExceptionName().c_str(),
+                                 error.GetMessage().c_str());
+            return {0, 0};
+        }
+    } while (more);
+    return {total_size, total_items};
+}
+
+size_t S3StorageTool::get_file_size(const std::string& key) {
+    Aws::S3::Model::HeadObjectRequest head_request;
+    head_request.SetBucket(bucket_name_.c_str());
+    head_request.SetKey(key.c_str());
+
+    auto object = s3_client_.HeadObject(head_request);
+    if (object.IsSuccess())
+    {
+        auto file_sz = object.GetResultWithOwnership().GetContentLength();
+        ARCTICDB_TRACE(log::storage(), "Size of {}: {}", key, file_sz);
+        return file_sz;
+    }
+    else
+    {
+        log::storage().error("Head Object error:  {} - {}", object .GetError().GetExceptionName(),
+                             object .GetError().GetMessage());
+        return 0;
+    }
+}
+
+
+void S3StorageTool::delete_bucket(const std::string& prefix) {
+    iterate_bucket([&](const std::string& key) {
+                       Aws::S3::Model::DeleteObjectRequest object_request;
+                       object_request.WithBucket(bucket_name_.c_str()).WithKey(key.c_str());
+
+                       auto delete_object_outcome = s3_client_.DeleteObject(object_request);
+                       if (delete_object_outcome.IsSuccess())
+                           ARCTICDB_DEBUG(log::storage(), "Deleted object with key '{}'", key);
+                       else {
+                           const auto& error = delete_object_outcome.GetError();
+                           log::storage().warn("Failed to delete object with key '{}' {}:{}",
+                                               key,
+                                               error.GetExceptionName().c_str(),
+                                               error.GetMessage().c_str());
+                       }
+                   },
+                   prefix);
+}
+
+
 S3StorageTool::S3StorageTool(const Config &conf) :
     s3_api_(S3ApiInstance::instance()),
     s3_client_(get_aws_credentials(conf), get_s3_config(conf), Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, false),
diff --git a/cpp/arcticdb/storage/s3/s3_storage_tool.hpp b/cpp/arcticdb/storage/s3/s3_storage_tool.hpp
index d8c62e2a5d..a18a357f93 100644
--- a/cpp/arcticdb/storage/s3/s3_storage_tool.hpp
+++ b/cpp/arcticdb/storage/s3/s3_storage_tool.hpp
@@ -28,7 +28,7 @@ class S3StorageTool {
     std::string get_object(const std::string& key);
 
     size_t get_file_size(const std::string& key);
-    inline std::pair<size_t, size_t> get_prefix_info(const std::string& prefix = std::string());
+    std::pair<size_t, size_t> get_prefix_info(const std::string& prefix = std::string());
 
     void delete_object(const std::string& key);
 
@@ -39,6 +39,3 @@ class S3StorageTool {
 };
 
 } // namespace arcticdb::storage::s3
-
-#define ARCTICDB_S3_STORAGE_TOOL_H_
-#include <arcticdb/storage/s3/s3_storage_tool-inl.hpp>
\ No newline at end of file


Test patch:
diff --git a/cpp/arcticdb/util/test/test_id_transformation.cpp b/cpp/arcticdb/util/test/test_id_transformation.cpp
index 335241b96c..bb996c281d 100644
--- a/cpp/arcticdb/util/test/test_id_transformation.cpp
+++ b/cpp/arcticdb/util/test/test_id_transformation.cpp
@@ -7,8 +7,7 @@
 
 #include <gtest/gtest.h>
 
-#define ARCTICDB_NFS_BACKED_STORAGE_H_ "TODO: We should extract the stuff we want to to a separate header"
-#include <arcticdb/storage/s3/nfs_backed_storage-inl.hpp>
+#include <arcticdb/storage/s3/nfs_backed_storage.hpp>
 #include <arcticdb/entity/atom_key.hpp>
 
 TEST(KeyTransformation, Roundtrip) {