Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

12 files changed, 3433 insertions, 0 deletions

diff --git a/src/test/crimson/seastore/CMakeLists.txt b/src/test/crimson/seastore/CMakeLists.txt
new file mode 100644
index 000000000..6c21ac7c5
--- /dev/null
+++ b/src/test/crimson/seastore/CMakeLists.txt
@@ -0,0 +1,51 @@
+add_executable(unittest-transaction-manager
+  test_block.cc
+  test_transaction_manager.cc
+  ../gtest_seastar.cc)
+add_ceph_unittest(unittest-transaction-manager
+  --memory 256M --smp 1)
+target_link_libraries(
+  unittest-transaction-manager
+  ${CMAKE_DL_LIBS}
+  crimson-seastore)
+
+add_executable(unittest-btree-lba-manager
+  test_btree_lba_manager.cc
+  ../gtest_seastar.cc)
+add_ceph_unittest(unittest-btree-lba-manager
+  --memory 256M --smp 1)
+target_link_libraries(
+  unittest-btree-lba-manager
+  ${CMAKE_DL_LIBS}
+  crimson-seastore)
+
+add_executable(unittest-seastore-journal
+  test_seastore_journal.cc)
+add_ceph_test(unittest-seastore-journal
+  unittest-seastore-journal --memory 256M --smp 1)
+target_link_libraries(
+  unittest-seastore-journal
+  crimson::gtest
+  crimson-seastore)
+
+add_executable(unittest-seastore-cache
+  test_block.cc
+  test_seastore_cache.cc)
+add_ceph_test(unittest-seastore-cache
+  unittest-seastore-cache --memory 256M --smp 1)
+target_link_libraries(
+  unittest-seastore-cache
+  crimson::gtest
+  crimson-seastore)
+
+add_executable(unittest-extmap-manager
+  test_extmap_manager.cc
+  ../gtest_seastar.cc)
+add_ceph_unittest(unittest-extmap-manager
+  --memory 256M --smp 1)
+target_link_libraries(
+  unittest-extmap-manager
+  ${CMAKE_DL_LIBS}
+  crimson-seastore)
+
+add_subdirectory(onode_tree)
diff --git a/src/test/crimson/seastore/onode_tree/CMakeLists.txt b/src/test/crimson/seastore/onode_tree/CMakeLists.txt
new file mode 100644
index 000000000..0886d2fb6
--- /dev/null
+++ b/src/test/crimson/seastore/onode_tree/CMakeLists.txt
@@ -0,0 +1,15 @@
+add_executable(test-seastore-onode-tree-node
+  test_node.cc)
+add_ceph_unittest(test-seastore-onode-tree-node
+  --memory 256M --smp 1)
+target_link_libraries(test-seastore-onode-tree-node
+  crimson-seastore
+  GTest::Main)
+
+add_executable(unittest-staged-fltree
+  test_staged_fltree.cc
+  ../../gtest_seastar.cc)
+add_ceph_unittest(unittest-staged-fltree
+  --memory 256M --smp 1)
+target_link_libraries(unittest-staged-fltree
+  crimson-seastore)
diff --git a/src/test/crimson/seastore/onode_tree/test_node.cc b/src/test/crimson/seastore/onode_tree/test_node.cc
new file mode 100644
index 000000000..178f78365
--- /dev/null
+++ b/src/test/crimson/seastore/onode_tree/test_node.cc
@@ -0,0 +1,207 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <gtest/gtest.h>
+
+#include "crimson/os/seastore/onode_manager/simple-fltree/onode_node.h"
+
+using crimson::os::seastore::Onode;
+using crimson::os::seastore::OnodeRef;
+
+TEST(OnodeNode, denc)
+{
+  Onode onode{"hello"};
+  bufferlist bl;
+  ceph::encode(onode, bl);
+  bl.rebuild();
+  auto flattened = reinterpret_cast<const onode_t*>(bl.c_str());
+  auto actual_onode = flattened->decode();
+  ASSERT_EQ(*actual_onode, onode);
+}
+
+TEST(OnodeNode, lookup)
+{
+  static constexpr size_t BLOCK_SIZE = 512;
+  char buf[BLOCK_SIZE];
+  using leaf_node_0 = node_t<BLOCK_SIZE, 0, ntype_t::leaf>;
+  auto leaf = new (buf) leaf_node_0;
+  ghobject_t oid{hobject_t{object_t{"saturn"}, "", 0, 0, 0, "solar"}};
+  {
+    auto [slot, found] = leaf->lower_bound(oid);
+    ASSERT_FALSE(found);
+    ASSERT_EQ(0, slot);
+  }
+  Onode onode{"hello"};
+  bufferlist bl;
+  ceph::encode(onode, bl);
+  bl.rebuild();
+  auto flattened = reinterpret_cast<const onode_t*>(bl.c_str());
+  leaf->insert_at(0, oid, *flattened);
+  {
+    auto [slot, found] = leaf->lower_bound(oid);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(0, slot);
+    const auto& [key1, key2] = leaf->key_at(slot);
+    auto& item = leaf->item_at(key1);
+    auto actual_onode = item.decode();
+    ASSERT_EQ(*actual_onode, onode);
+  }
+}
+
+TEST(OnodeNode, grab_from_right)
+{
+  static constexpr size_t BLOCK_SIZE = 512;
+  char buf1[BLOCK_SIZE];
+  char buf2[BLOCK_SIZE];
+  using leaf_node_0 = node_t<BLOCK_SIZE, 0, ntype_t::leaf>;
+  auto leaf1 = new (buf1) leaf_node_0;
+  auto leaf2 = new (buf2) leaf_node_0;
+  auto& dummy_parent = *leaf1;
+
+  ghobject_t oid1{hobject_t{object_t{"earth"}, "", 0, 0, 0, "solar"}};
+  ghobject_t oid2{hobject_t{object_t{"jupiter"}, "", 0, 0, 0, "solar"}};
+  ghobject_t oid3{hobject_t{object_t{"saturn"}, "", 0, 0, 0, "solar"}};
+  Onode onode{"hello"};
+  bufferlist bl;
+  ceph::encode(onode, bl);
+  bl.rebuild();
+  auto flattened = reinterpret_cast<const onode_t*>(bl.c_str());
+  // so they are ordered as they should
+  leaf1->insert_at(0, oid1, *flattened);
+  ASSERT_EQ(1, leaf1->count);
+  {
+    auto [slot, found] = leaf1->lower_bound(oid1);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(0, slot);
+  }
+  {
+    leaf2->insert_at(0, oid2, *flattened);
+    auto [slot, found] = leaf2->lower_bound(oid2);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(0, slot);
+  }
+  {
+    leaf2->insert_at(1, oid3, *flattened);
+    auto [slot, found] = leaf2->lower_bound(oid3);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(1, slot);
+  }
+  ASSERT_EQ(2, leaf2->count);
+
+  // normally we let left merge right, so we just need to remove an
+  // entry in parent, let's keep this convention here
+  auto mover = make_mover(dummy_parent, *leaf2, *leaf1, 0);
+  // just grab a single item from right
+  mover.move_from(0, 1, 1);
+  auto to_delta = mover.to_delta();
+  ASSERT_EQ(to_delta.op_t::insert_back, to_delta.op);
+  leaf1->insert_back(std::move(to_delta.keys), std::move(to_delta.cells));
+
+  ASSERT_EQ(2, leaf1->count);
+  {
+    auto [slot, found] = leaf1->lower_bound(oid2);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(1, slot);
+  }
+
+  auto from_delta = mover.from_delta();
+  ASSERT_EQ(from_delta.op_t::shift_left, from_delta.op);
+  leaf2->shift_left(from_delta.n, 0);
+  ASSERT_EQ(1, leaf2->count);
+}
+
+TEST(OnodeNode, merge_right)
+{
+  static constexpr size_t BLOCK_SIZE = 512;
+  char buf1[BLOCK_SIZE];
+  char buf2[BLOCK_SIZE];
+  using leaf_node_0 = node_t<BLOCK_SIZE, 0, ntype_t::leaf>;
+  auto leaf1 = new (buf1) leaf_node_0;
+  auto leaf2 = new (buf2) leaf_node_0;
+  auto& dummy_parent = leaf1;
+
+  ghobject_t oid1{hobject_t{object_t{"earth"}, "", 0, 0, 0, "solar"}};
+  ghobject_t oid2{hobject_t{object_t{"jupiter"}, "", 0, 0, 0, "solar"}};
+  ghobject_t oid3{hobject_t{object_t{"saturn"}, "", 0, 0, 0, "solar"}};
+  Onode onode{"hello"};
+  bufferlist bl;
+  ceph::encode(onode, bl);
+  bl.rebuild();
+  auto flattened = reinterpret_cast<const onode_t*>(bl.c_str());
+  // so they are ordered as they should
+  leaf1->insert_at(0, oid1, *flattened);
+  ASSERT_EQ(1, leaf1->count);
+  {
+    auto [slot, found] = leaf1->lower_bound(oid1);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(0, slot);
+  }
+  {
+    leaf2->insert_at(0, oid2, *flattened);
+    auto [slot, found] = leaf2->lower_bound(oid2);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(0, slot);
+  }
+  {
+    leaf2->insert_at(1, oid3, *flattened);
+    auto [slot, found] = leaf2->lower_bound(oid3);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(1, slot);
+  }
+  ASSERT_EQ(2, leaf2->count);
+
+  // normally we let left merge right, so we just need to remove an
+  // entry in parent, let's keep this convention here
+  auto mover = make_mover(dummy_parent, *leaf2, *leaf1, 0);
+  // just grab a single item from right
+  mover.move_from(0, 1, 2);
+  auto to_delta = mover.to_delta();
+  ASSERT_EQ(to_delta.op_t::insert_back, to_delta.op);
+  leaf1->insert_back(std::move(to_delta.keys), std::move(to_delta.cells));
+
+  ASSERT_EQ(3, leaf1->count);
+  {
+    auto [slot, found] = leaf1->lower_bound(oid2);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(1, slot);
+  }
+  {
+    auto [slot, found] = leaf1->lower_bound(oid3);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(2, slot);
+  }
+
+  // its onode tree's responsibility to retire the node
+  auto from_delta = mover.from_delta();
+  ASSERT_EQ(from_delta.op_t::nop, from_delta.op);
+}
+
+TEST(OnodeNode, remove_basic)
+{
+  static constexpr size_t BLOCK_SIZE = 512;
+  char buf[BLOCK_SIZE];
+  using leaf_node_0 = node_t<BLOCK_SIZE, 0, ntype_t::leaf>;
+  auto leaf = new (buf) leaf_node_0;
+  ghobject_t oid{hobject_t{object_t{"saturn"}, "", 0, 0, 0, "solar"}};
+  {
+    auto [slot, found] = leaf->lower_bound(oid);
+    ASSERT_FALSE(found);
+    ASSERT_EQ(0, slot);
+  }
+  Onode onode{"hello"};
+  bufferlist bl;
+  ceph::encode(onode, bl);
+  bl.rebuild();
+  auto flattened = reinterpret_cast<const onode_t*>(bl.c_str());
+  leaf->insert_at(0, oid, *flattened);
+  {
+    auto [slot, found] = leaf->lower_bound(oid);
+    ASSERT_TRUE(found);
+    ASSERT_EQ(0, slot);
+    leaf->remove_from(slot);
+  }
+  {
+    auto [slot, found] = leaf->lower_bound(oid);
+    ASSERT_FALSE(found);
+  }
+}
diff --git a/src/test/crimson/seastore/onode_tree/test_staged_fltree.cc b/src/test/crimson/seastore/onode_tree/test_staged_fltree.cc
new file mode 100644
index 000000000..da7422bcb
--- /dev/null
+++ b/src/test/crimson/seastore/onode_tree/test_staged_fltree.cc
@@ -0,0 +1,1204 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <array>
+#include <cstring>
+#include <memory>
+#include <set>
+#include <sstream>
+#include <vector>
+
+#include "crimson/common/log.h"
+#include "crimson/os/seastore/onode_manager/staged-fltree/node.h"
+#include "crimson/os/seastore/onode_manager/staged-fltree/node_extent_manager.h"
+#include "crimson/os/seastore/onode_manager/staged-fltree/node_layout.h"
+#include "crimson/os/seastore/onode_manager/staged-fltree/tree.h"
+#include "crimson/os/seastore/onode_manager/staged-fltree/tree_utils.h"
+
+#include "test/crimson/gtest_seastar.h"
+#include "test/crimson/seastore/transaction_manager_test_state.h"
+
+using namespace crimson::os::seastore::onode;
+
+namespace {
+  constexpr bool IS_DUMMY_SYNC = false;
+
+  [[maybe_unused]] seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_test);
+  }
+
+  ghobject_t make_ghobj(
+      shard_t shard, pool_t pool, crush_hash_t crush,
+      std::string ns, std::string oid, snap_t snap, gen_t gen) {
+    return ghobject_t{shard_id_t{shard}, pool, crush, ns, oid, snap, gen};
+  }
+
+  // return a key_view_t and its underlying memory buffer.
+  // the buffer needs to be freed manually.
+  std::pair<key_view_t, void*> build_key_view(const ghobject_t& hobj) {
+    key_hobj_t key_hobj(hobj);
+    size_t key_size = sizeof(shard_pool_crush_t) + sizeof(snap_gen_t) +
+                      ns_oid_view_t::estimate_size<KeyT::HOBJ>(key_hobj);
+    void* p_mem = std::malloc(key_size);
+
+    key_view_t key_view;
+    char* p_fill = (char*)p_mem + key_size;
+
+    auto spc = shard_pool_crush_t::from_key<KeyT::HOBJ>(key_hobj);
+    p_fill -= sizeof(shard_pool_crush_t);
+    std::memcpy(p_fill, &spc, sizeof(shard_pool_crush_t));
+    key_view.set(*reinterpret_cast<const shard_pool_crush_t*>(p_fill));
+
+    auto p_ns_oid = p_fill;
+    ns_oid_view_t::test_append<KeyT::HOBJ>(key_hobj, p_fill);
+    ns_oid_view_t ns_oid_view(p_ns_oid);
+    key_view.set(ns_oid_view);
+
+    auto sg = snap_gen_t::from_key<KeyT::HOBJ>(key_hobj);
+    p_fill -= sizeof(snap_gen_t);
+    ceph_assert(p_fill == (char*)p_mem);
+    std::memcpy(p_fill, &sg, sizeof(snap_gen_t));
+    key_view.set(*reinterpret_cast<const snap_gen_t*>(p_fill));
+
+    return {key_view, p_mem};
+  }
+}
+
+struct a_basic_test_t : public seastar_test_suite_t {};
+
+TEST_F(a_basic_test_t, 1_basic_sizes)
+{
+  logger().info("\n"
+    "Bytes of struct:\n"
+    "  node_header_t: {}\n"
+    "  shard_pool_t: {}\n"
+    "  shard_pool_crush_t: {}\n"
+    "  crush_t: {}\n"
+    "  snap_gen_t: {}\n"
+    "  slot_0_t: {}\n"
+    "  slot_1_t: {}\n"
+    "  slot_3_t: {}\n"
+    "  node_fields_0_t: {}\n"
+    "  node_fields_1_t: {}\n"
+    "  node_fields_2_t: {}\n"
+    "  internal_fields_3_t: {}\n"
+    "  leaf_fields_3_t: {}\n"
+    "  internal_sub_item_t: {}",
+    sizeof(node_header_t), sizeof(shard_pool_t),
+    sizeof(shard_pool_crush_t), sizeof(crush_t), sizeof(snap_gen_t),
+    sizeof(slot_0_t), sizeof(slot_1_t), sizeof(slot_3_t),
+    sizeof(node_fields_0_t), sizeof(node_fields_1_t), sizeof(node_fields_2_t),
+    sizeof(internal_fields_3_t), sizeof(leaf_fields_3_t), sizeof(internal_sub_item_t)
+  );
+
+  auto hobj = make_ghobj(0, 0, 0, "n", "o", 0, 0);
+  key_hobj_t key(hobj);
+  auto [key_view, p_mem] = build_key_view(hobj);
+  onode_t value = {2};
+#define _STAGE_T(NodeType) node_to_stage_t<typename NodeType::node_stage_t>
+#define NXT_T(StageType)  staged<typename StageType::next_param_t>
+  laddr_packed_t i_value{0};
+  logger().info("\n"
+    "Bytes of a key-value insertion (full-string):\n"
+    "  s-p-c, 'n'-'o', s-g => onode_t(2): typically internal 41B, leaf 35B\n"
+    "  InternalNode0: {} {} {}\n"
+    "  InternalNode1: {} {} {}\n"
+    "  InternalNode2: {} {}\n"
+    "  InternalNode3: {}\n"
+    "  LeafNode0: {} {} {}\n"
+    "  LeafNode1: {} {} {}\n"
+    "  LeafNode2: {} {}\n"
+    "  LeafNode3: {}",
+    _STAGE_T(InternalNode0)::template insert_size<KeyT::VIEW>(key_view, i_value),
+    NXT_T(_STAGE_T(InternalNode0))::template insert_size<KeyT::VIEW>(key_view, i_value),
+    NXT_T(NXT_T(_STAGE_T(InternalNode0)))::template insert_size<KeyT::VIEW>(key_view, i_value),
+    _STAGE_T(InternalNode1)::template insert_size<KeyT::VIEW>(key_view, i_value),
+    NXT_T(_STAGE_T(InternalNode1))::template insert_size<KeyT::VIEW>(key_view, i_value),
+    NXT_T(NXT_T(_STAGE_T(InternalNode1)))::template insert_size<KeyT::VIEW>(key_view, i_value),
+    _STAGE_T(InternalNode2)::template insert_size<KeyT::VIEW>(key_view, i_value),
+    NXT_T(_STAGE_T(InternalNode2))::template insert_size<KeyT::VIEW>(key_view, i_value),
+    _STAGE_T(InternalNode3)::template insert_size<KeyT::VIEW>(key_view, i_value),
+    _STAGE_T(LeafNode0)::template insert_size<KeyT::HOBJ>(key, value),
+    NXT_T(_STAGE_T(LeafNode0))::template insert_size<KeyT::HOBJ>(key, value),
+    NXT_T(NXT_T(_STAGE_T(LeafNode0)))::template insert_size<KeyT::HOBJ>(key, value),
+    _STAGE_T(LeafNode1)::template insert_size<KeyT::HOBJ>(key, value),
+    NXT_T(_STAGE_T(LeafNode1))::template insert_size<KeyT::HOBJ>(key, value),
+    NXT_T(NXT_T(_STAGE_T(LeafNode1)))::template insert_size<KeyT::HOBJ>(key, value),
+    _STAGE_T(LeafNode2)::template insert_size<KeyT::HOBJ>(key, value),
+    NXT_T(_STAGE_T(LeafNode2))::template insert_size<KeyT::HOBJ>(key, value),
+    _STAGE_T(LeafNode3)::template insert_size<KeyT::HOBJ>(key, value)
+  );
+  std::free(p_mem);
+}
+
+TEST_F(a_basic_test_t, 2_node_sizes)
+{
+  run_async([this] {
+    auto nm = NodeExtentManager::create_dummy(IS_DUMMY_SYNC);
+    auto t = make_transaction();
+    context_t c{*nm, *t};
+    std::array<std::pair<NodeImplURef, NodeExtentMutable>, 16> nodes = {
+      InternalNode0::allocate(c, false, 1u).unsafe_get0().make_pair(),
+      InternalNode1::allocate(c, false, 1u).unsafe_get0().make_pair(),
+      InternalNode2::allocate(c, false, 1u).unsafe_get0().make_pair(),
+      InternalNode3::allocate(c, false, 1u).unsafe_get0().make_pair(),
+      InternalNode0::allocate(c, true, 1u).unsafe_get0().make_pair(),
+      InternalNode1::allocate(c, true, 1u).unsafe_get0().make_pair(),
+      InternalNode2::allocate(c, true, 1u).unsafe_get0().make_pair(),
+      InternalNode3::allocate(c, true, 1u).unsafe_get0().make_pair(),
+      LeafNode0::allocate(c, false, 0u).unsafe_get0().make_pair(),
+      LeafNode1::allocate(c, false, 0u).unsafe_get0().make_pair(),
+      LeafNode2::allocate(c, false, 0u).unsafe_get0().make_pair(),
+      LeafNode3::allocate(c, false, 0u).unsafe_get0().make_pair(),
+      LeafNode0::allocate(c, true, 0u).unsafe_get0().make_pair(),
+      LeafNode1::allocate(c, true, 0u).unsafe_get0().make_pair(),
+      LeafNode2::allocate(c, true, 0u).unsafe_get0().make_pair(),
+      LeafNode3::allocate(c, true, 0u).unsafe_get0().make_pair()
+    };
+    std::ostringstream oss;
+    oss << "\nallocated nodes:";
+    for (auto iter = nodes.begin(); iter != nodes.end(); ++iter) {
+      oss << "\n  ";
+      auto& ref_node = iter->first;
+      ref_node->dump_brief(oss);
+    }
+    logger().info("{}", oss.str());
+  });
+}
+
+struct b_dummy_tree_test_t : public seastar_test_suite_t {
+  NodeExtentManagerURef moved_nm;
+  TransactionRef ref_t;
+  Transaction& t;
+  context_t c;
+  Btree tree;
+
+  b_dummy_tree_test_t()
+    : moved_nm{NodeExtentManager::create_dummy(IS_DUMMY_SYNC)},
+      ref_t{make_transaction()},
+      t{*ref_t},
+      c{*moved_nm, t},
+      tree{std::move(moved_nm)} {}
+
+  seastar::future<> set_up_fut() override final {
+    return tree.mkfs(t).handle_error(
+      crimson::ct_error::all_same_way([] {
+        ASSERT_FALSE("Unable to mkfs");
+      })
+    );
+  }
+};
+
+TEST_F(b_dummy_tree_test_t, 3_random_insert_leaf_node)
+{
+  run_async([this] {
+    logger().info("\n---------------------------------------------"
+                  "\nrandomized leaf node insert:\n");
+    auto key_s = make_ghobj(0, 0, 0, "ns", "oid", 0, 0);
+    auto key_e = make_ghobj(
+        std::numeric_limits<shard_t>::max(), 0, 0, "ns", "oid", 0, 0);
+    ASSERT_TRUE(tree.find(t, key_s).unsafe_get0().is_end());
+    ASSERT_TRUE(tree.begin(t).unsafe_get0().is_end());
+    ASSERT_TRUE(tree.last(t).unsafe_get0().is_end());
+
+    std::vector<std::tuple<ghobject_t,
+                           const onode_t*,
+                           Btree::Cursor>> insert_history;
+    auto f_validate_insert_new = [this, &insert_history] (
+        const ghobject_t& key, const onode_t& value) {
+      auto [cursor, success] = tree.insert(t, key, value).unsafe_get0();
+      ceph_assert(success);
+      insert_history.emplace_back(key, &value, cursor);
+      Onodes::validate_cursor(cursor, key, value);
+      auto cursor_ = tree.lower_bound(t, key).unsafe_get0();
+      ceph_assert(cursor_.get_ghobj() == key);
+      ceph_assert(cursor_.value() == cursor.value());
+      return cursor.value();
+    };
+    auto onodes = Onodes(15);
+
+    // insert key1, onode1 at STAGE_LEFT
+    auto key1 = make_ghobj(3, 3, 3, "ns3", "oid3", 3, 3);
+    auto& onode1 = onodes.pick();
+    auto p_value1 = f_validate_insert_new(key1, onode1);
+
+    // validate lookup
+    {
+      auto cursor1_s = tree.lower_bound(t, key_s).unsafe_get0();
+      ASSERT_EQ(cursor1_s.get_ghobj(), key1);
+      ASSERT_EQ(cursor1_s.value(), p_value1);
+      auto cursor1_e = tree.lower_bound(t, key_e).unsafe_get0();
+      ASSERT_TRUE(cursor1_e.is_end());
+    }
+
+    // insert the same key1 with a different onode
+    {
+      auto& onode1_dup = onodes.pick();
+      auto [cursor1_dup, ret1_dup] = tree.insert(
+          t, key1, onode1_dup).unsafe_get0();
+      ASSERT_FALSE(ret1_dup);
+      Onodes::validate_cursor(cursor1_dup, key1, onode1);
+    }
+
+    // insert key2, onode2 to key1's left at STAGE_LEFT
+    // insert node front at STAGE_LEFT
+    auto key2 = make_ghobj(2, 2, 2, "ns3", "oid3", 3, 3);
+    auto& onode2 = onodes.pick();
+    f_validate_insert_new(key2, onode2);
+
+    // insert key3, onode3 to key1's right at STAGE_LEFT
+    // insert node last at STAGE_LEFT
+    auto key3 = make_ghobj(4, 4, 4, "ns3", "oid3", 3, 3);
+    auto& onode3 = onodes.pick();
+    f_validate_insert_new(key3, onode3);
+
+    // insert key4, onode4 to key1's left at STAGE_STRING (collision)
+    auto key4 = make_ghobj(3, 3, 3, "ns2", "oid2", 3, 3);
+    auto& onode4 = onodes.pick();
+    f_validate_insert_new(key4, onode4);
+
+    // insert key5, onode5 to key1's right at STAGE_STRING (collision)
+    auto key5 = make_ghobj(3, 3, 3, "ns4", "oid4", 3, 3);
+    auto& onode5 = onodes.pick();
+    f_validate_insert_new(key5, onode5);
+
+    // insert key6, onode6 to key1's left at STAGE_RIGHT
+    auto key6 = make_ghobj(3, 3, 3, "ns3", "oid3", 2, 2);
+    auto& onode6 = onodes.pick();
+    f_validate_insert_new(key6, onode6);
+
+    // insert key7, onode7 to key1's right at STAGE_RIGHT
+    auto key7 = make_ghobj(3, 3, 3, "ns3", "oid3", 4, 4);
+    auto& onode7 = onodes.pick();
+    f_validate_insert_new(key7, onode7);
+
+    // insert node front at STAGE_RIGHT
+    auto key8 = make_ghobj(2, 2, 2, "ns3", "oid3", 2, 2);
+    auto& onode8 = onodes.pick();
+    f_validate_insert_new(key8, onode8);
+
+    // insert node front at STAGE_STRING (collision)
+    auto key9 = make_ghobj(2, 2, 2, "ns2", "oid2", 3, 3);
+    auto& onode9 = onodes.pick();
+    f_validate_insert_new(key9, onode9);
+
+    // insert node last at STAGE_RIGHT
+    auto key10 = make_ghobj(4, 4, 4, "ns3", "oid3", 4, 4);
+    auto& onode10 = onodes.pick();
+    f_validate_insert_new(key10, onode10);
+
+    // insert node last at STAGE_STRING (collision)
+    auto key11 = make_ghobj(4, 4, 4, "ns4", "oid4", 3, 3);
+    auto& onode11 = onodes.pick();
+    f_validate_insert_new(key11, onode11);
+
+    // insert key, value randomly until a perfect 3-ary tree is formed
+    std::vector<std::pair<ghobject_t, const onode_t*>> kvs{
+      {make_ghobj(2, 2, 2, "ns2", "oid2", 2, 2), &onodes.pick()},
+      {make_ghobj(2, 2, 2, "ns2", "oid2", 4, 4), &onodes.pick()},
+      {make_ghobj(2, 2, 2, "ns3", "oid3", 4, 4), &onodes.pick()},
+      {make_ghobj(2, 2, 2, "ns4", "oid4", 2, 2), &onodes.pick()},
+      {make_ghobj(2, 2, 2, "ns4", "oid4", 3, 3), &onodes.pick()},
+      {make_ghobj(2, 2, 2, "ns4", "oid4", 4, 4), &onodes.pick()},
+      {make_ghobj(3, 3, 3, "ns2", "oid2", 2, 2), &onodes.pick()},
+      {make_ghobj(3, 3, 3, "ns2", "oid2", 4, 4), &onodes.pick()},
+      {make_ghobj(3, 3, 3, "ns4", "oid4", 2, 2), &onodes.pick()},
+      {make_ghobj(3, 3, 3, "ns4", "oid4", 4, 4), &onodes.pick()},
+      {make_ghobj(4, 4, 4, "ns2", "oid2", 2, 2), &onodes.pick()},
+      {make_ghobj(4, 4, 4, "ns2", "oid2", 3, 3), &onodes.pick()},
+      {make_ghobj(4, 4, 4, "ns2", "oid2", 4, 4), &onodes.pick()},
+      {make_ghobj(4, 4, 4, "ns3", "oid3", 2, 2), &onodes.pick()},
+      {make_ghobj(4, 4, 4, "ns4", "oid4", 2, 2), &onodes.pick()},
+      {make_ghobj(4, 4, 4, "ns4", "oid4", 4, 4), &onodes.pick()}};
+    auto [smallest_key, smallest_value] = kvs[0];
+    auto [largest_key, largest_value] = kvs[kvs.size() - 1];
+    std::random_shuffle(kvs.begin(), kvs.end());
+    std::for_each(kvs.begin(), kvs.end(), [&f_validate_insert_new] (auto& kv) {
+      f_validate_insert_new(kv.first, *kv.second);
+    });
+    ASSERT_EQ(tree.height(t).unsafe_get0(), 1);
+    ASSERT_FALSE(tree.test_is_clean());
+
+    for (auto& [k, v, c] : insert_history) {
+      // validate values in tree keep intact
+      auto cursor = tree.lower_bound(t, k).unsafe_get0();
+      Onodes::validate_cursor(cursor, k, *v);
+      // validate values in cursors keep intact
+      Onodes::validate_cursor(c, k, *v);
+    }
+    Onodes::validate_cursor(
+        tree.lower_bound(t, key_s).unsafe_get0(), smallest_key, *smallest_value);
+    Onodes::validate_cursor(
+        tree.begin(t).unsafe_get0(), smallest_key, *smallest_value);
+    Onodes::validate_cursor(
+        tree.last(t).unsafe_get0(), largest_key, *largest_value);
+
+    std::ostringstream oss;
+    tree.dump(t, oss);
+    logger().info("\n{}\n", oss.str());
+
+    insert_history.clear();
+  });
+}
+
+static std::set<ghobject_t> build_key_set(
+    std::pair<unsigned, unsigned> range_2,
+    std::pair<unsigned, unsigned> range_1,
+    std::pair<unsigned, unsigned> range_0,
+    std::string padding = "",
+    bool is_internal = false) {
+  ceph_assert(range_1.second <= 10);
+  std::set<ghobject_t> ret;
+  ghobject_t key;
+  for (unsigned i = range_2.first; i < range_2.second; ++i) {
+    for (unsigned j = range_1.first; j < range_1.second; ++j) {
+      for (unsigned k = range_0.first; k < range_0.second; ++k) {
+        std::ostringstream os_ns;
+        os_ns << "ns" << j;
+        std::ostringstream os_oid;
+        os_oid << "oid" << j << padding;
+        key = make_ghobj(i, i, i, os_ns.str(), os_oid.str(), k, k);
+        ret.insert(key);
+      }
+    }
+  }
+  if (is_internal) {
+    ret.insert(make_ghobj(9, 9, 9, "ns~last", "oid~last", 9, 9));
+  }
+  return ret;
+}
+
+class TestTree {
+ public:
+  TestTree()
+    : moved_nm{NodeExtentManager::create_dummy(IS_DUMMY_SYNC)},
+      ref_t{make_transaction()},
+      t{*ref_t},
+      c{*moved_nm, t},
+      tree{std::move(moved_nm)},
+      onodes{0} {}
+
+  seastar::future<> build_tree(
+      std::pair<unsigned, unsigned> range_2,
+      std::pair<unsigned, unsigned> range_1,
+      std::pair<unsigned, unsigned> range_0,
+      size_t onode_size) {
+    return seastar::async([this, range_2, range_1, range_0, onode_size] {
+      tree.mkfs(t).unsafe_get0();
+      //logger().info("\n---------------------------------------------"
+      //              "\nbefore leaf node split:\n");
+      auto keys = build_key_set(range_2, range_1, range_0);
+      for (auto& key : keys) {
+        auto& value = onodes.create(onode_size);
+        insert_tree(key, value).get0();
+      }
+      ASSERT_EQ(tree.height(t).unsafe_get0(), 1);
+      ASSERT_FALSE(tree.test_is_clean());
+      //std::ostringstream oss;
+      //tree.dump(t, oss);
+      //logger().info("\n{}\n", oss.str());
+    });
+  }
+
+  seastar::future<> build_tree(
+      const std::vector<ghobject_t>& keys, const std::vector<const onode_t*>& values) {
+    return seastar::async([this, keys, values] {
+      tree.mkfs(t).unsafe_get0();
+      //logger().info("\n---------------------------------------------"
+      //              "\nbefore leaf node split:\n");
+      ASSERT_EQ(keys.size(), values.size());
+      auto key_iter = keys.begin();
+      auto value_iter = values.begin();
+      while (key_iter != keys.end()) {
+        insert_tree(*key_iter, **value_iter).get0();
+        ++key_iter;
+        ++value_iter;
+      }
+      ASSERT_EQ(tree.height(t).unsafe_get0(), 1);
+      ASSERT_FALSE(tree.test_is_clean());
+      //std::ostringstream oss;
+      //tree.dump(t, oss);
+      //logger().info("\n{}\n", oss.str());
+    });
+  }
+
+  seastar::future<> split(const ghobject_t& key, const onode_t& value,
+                          const split_expectation_t& expected) {
+    return seastar::async([this, key, &value, expected] {
+      Btree tree_clone(NodeExtentManager::create_dummy(IS_DUMMY_SYNC));
+      auto ref_t_clone = make_transaction();
+      Transaction& t_clone = *ref_t_clone;
+      tree_clone.test_clone_from(t_clone, t, tree).unsafe_get0();
+
+      logger().info("insert {}:", key_hobj_t(key));
+      auto [cursor, success] = tree_clone.insert(t_clone, key, value).unsafe_get0();
+      ASSERT_TRUE(success);
+      Onodes::validate_cursor(cursor, key, value);
+
+      std::ostringstream oss;
+      tree_clone.dump(t_clone, oss);
+      logger().info("dump new root:\n{}", oss.str());
+      EXPECT_EQ(tree_clone.height(t_clone).unsafe_get0(), 2);
+
+      for (auto& [k, v, c] : insert_history) {
+        auto result = tree_clone.lower_bound(t_clone, k).unsafe_get0();
+        Onodes::validate_cursor(result, k, *v);
+      }
+      auto result = tree_clone.lower_bound(t_clone, key).unsafe_get0();
+      Onodes::validate_cursor(result, key, value);
+      EXPECT_TRUE(last_split.match(expected));
+    });
+  }
+
+  const onode_t& create_onode(size_t size) {
+    return onodes.create(size);
+  }
+
+ private:
+  seastar::future<> insert_tree(const ghobject_t& key, const onode_t& value) {
+    return seastar::async([this, &key, &value] {
+      auto [cursor, success] = tree.insert(t, key, value).unsafe_get0();
+      ASSERT_TRUE(success);
+      Onodes::validate_cursor(cursor, key, value);
+      insert_history.emplace_back(key, &value, cursor);
+    });
+  }
+
+  NodeExtentManagerURef moved_nm;
+  TransactionRef ref_t;
+  Transaction& t;
+  context_t c;
+  Btree tree;
+  Onodes onodes;
+  std::vector<std::tuple<
+    ghobject_t, const onode_t*, Btree::Cursor>> insert_history;
+};
+
+struct c_dummy_test_t : public seastar_test_suite_t {};
+
+TEST_F(c_dummy_test_t, 4_split_leaf_node)
+{
+  run_async([this] {
+    {
+      TestTree test;
+      test.build_tree({2, 5}, {2, 5}, {2, 5}, 120).get0();
+
+      auto& onode = test.create_onode(1144);
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to left front at stage 2, 1, 0\n");
+      test.split(make_ghobj(1, 1, 1, "ns3", "oid3", 3, 3), onode,
+                 {2u, 2u, true, InsertType::BEGIN}).get0();
+      test.split(make_ghobj(2, 2, 2, "ns1", "oid1", 3, 3), onode,
+                 {2u, 1u, true, InsertType::BEGIN}).get0();
+      test.split(make_ghobj(2, 2, 2, "ns2", "oid2", 1, 1), onode,
+                 {2u, 0u, true, InsertType::BEGIN}).get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to left back at stage 0, 1, 2, 1, 0\n");
+      test.split(make_ghobj(2, 2, 2, "ns4", "oid4", 5, 5), onode,
+                 {2u, 0u, true, InsertType::LAST}).get0();
+      test.split(make_ghobj(2, 2, 2, "ns5", "oid5", 3, 3), onode,
+                 {2u, 1u, true, InsertType::LAST}).get0();
+      test.split(make_ghobj(2, 3, 3, "ns3", "oid3", 3, 3), onode,
+                 {2u, 2u, true, InsertType::LAST}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns1", "oid1", 3, 3), onode,
+                 {2u, 1u, true, InsertType::LAST}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns2", "oid2", 1, 1), onode,
+                 {2u, 0u, true, InsertType::LAST}).get0();
+
+      auto& onode0 = test.create_onode(1416);
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to right front at stage 0, 1, 2, 1, 0\n");
+      test.split(make_ghobj(3, 3, 3, "ns4", "oid4", 5, 5), onode0,
+                 {2u, 0u, false, InsertType::BEGIN}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns5", "oid5", 3, 3), onode0,
+                 {2u, 1u, false, InsertType::BEGIN}).get0();
+      test.split(make_ghobj(3, 4, 4, "ns3", "oid3", 3, 3), onode0,
+                 {2u, 2u, false, InsertType::BEGIN}).get0();
+      test.split(make_ghobj(4, 4, 4, "ns1", "oid1", 3, 3), onode0,
+                 {2u, 1u, false, InsertType::BEGIN}).get0();
+      test.split(make_ghobj(4, 4, 4, "ns2", "oid2", 1, 1), onode0,
+                 {2u, 0u, false, InsertType::BEGIN}).get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to right back at stage 0, 1, 2\n");
+      test.split(make_ghobj(4, 4, 4, "ns4", "oid4", 5, 5), onode0,
+                 {2u, 0u, false, InsertType::LAST}).get0();
+      test.split(make_ghobj(4, 4, 4, "ns5", "oid5", 3, 3), onode0,
+                 {2u, 1u, false, InsertType::LAST}).get0();
+      test.split(make_ghobj(5, 5, 5, "ns3", "oid3", 3, 3), onode0,
+                 {2u, 2u, false, InsertType::LAST}).get0();
+
+      auto& onode1 = test.create_onode(316);
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 1; insert to left middle at stage 0, 1, 2, 1, 0\n");
+      test.split(make_ghobj(2, 2, 2, "ns4", "oid4", 5, 5), onode1,
+                 {1u, 0u, true, InsertType::MID}).get0();
+      test.split(make_ghobj(2, 2, 2, "ns5", "oid5", 3, 3), onode1,
+                 {1u, 1u, true, InsertType::MID}).get0();
+      test.split(make_ghobj(2, 2, 3, "ns3", "oid3", 3, 3), onode1,
+                 {1u, 2u, true, InsertType::MID}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns1", "oid1", 3, 3), onode1,
+                 {1u, 1u, true, InsertType::MID}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns2", "oid2", 1, 1), onode1,
+                 {1u, 0u, true, InsertType::MID}).get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 1; insert to left back at stage 0, 1, 0\n");
+      test.split(make_ghobj(3, 3, 3, "ns2", "oid2", 5, 5), onode1,
+                 {1u, 0u, true, InsertType::LAST}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns2", "oid3", 3, 3), onode1,
+                 {1u, 1u, true, InsertType::LAST}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns3", "oid3", 1, 1), onode1,
+                 {1u, 0u, true, InsertType::LAST}).get0();
+
+      auto& onode2 = test.create_onode(452);
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 1; insert to right front at stage 0, 1, 0\n");
+      test.split(make_ghobj(3, 3, 3, "ns3", "oid3", 5, 5), onode2,
+                 {1u, 0u, false, InsertType::BEGIN}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns3", "oid4", 3, 3), onode2,
+                 {1u, 1u, false, InsertType::BEGIN}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns4", "oid4", 1, 1), onode2,
+                 {1u, 0u, false, InsertType::BEGIN}).get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 1; insert to right middle at stage 0, 1, 2, 1, 0\n");
+      test.split(make_ghobj(3, 3, 3, "ns4", "oid4", 5, 5), onode2,
+                 {1u, 0u, false, InsertType::MID}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns5", "oid5", 3, 3), onode2,
+                 {1u, 1u, false, InsertType::MID}).get0();
+      test.split(make_ghobj(3, 3, 4, "ns3", "oid3", 3, 3), onode2,
+                 {1u, 2u, false, InsertType::MID}).get0();
+      test.split(make_ghobj(4, 4, 4, "ns1", "oid1", 3, 3), onode2,
+                 {1u, 1u, false, InsertType::MID}).get0();
+      test.split(make_ghobj(4, 4, 4, "ns2", "oid2", 1, 1), onode2,
+                 {1u, 0u, false, InsertType::MID}).get0();
+
+      auto& onode3 = test.create_onode(834);
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 0; insert to right middle at stage 0, 1, 2, 1, 0\n");
+      test.split(make_ghobj(3, 3, 3, "ns4", "oid4", 5, 5), onode3,
+                 {0u, 0u, false, InsertType::MID}).get0();
+      test.split(make_ghobj(3, 3, 3, "ns5", "oid5", 3, 3), onode3,
+                 {0u, 1u, false, InsertType::MID}).get0();
+      test.split(make_ghobj(3, 3, 4, "ns3", "oid3", 3, 3), onode3,
+                 {0u, 2u, false, InsertType::MID}).get0();
+      test.split(make_ghobj(4, 4, 4, "ns1", "oid1", 3, 3), onode3,
+                 {0u, 1u, false, InsertType::MID}).get0();
+      test.split(make_ghobj(4, 4, 4, "ns2", "oid2", 1, 1), onode3,
+                 {0u, 0u, false, InsertType::MID}).get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 0; insert to right front at stage 0\n");
+      test.split(make_ghobj(3, 3, 3, "ns4", "oid4", 2, 3), onode3,
+                 {0u, 0u, false, InsertType::BEGIN}).get0();
+
+      auto& onode4 = test.create_onode(572);
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 0; insert to left back at stage 0\n");
+      test.split(make_ghobj(3, 3, 3, "ns2", "oid2", 3, 4), onode4,
+                 {0u, 0u, true, InsertType::LAST}).get0();
+    }
+
+    {
+      TestTree test;
+      test.build_tree({2, 4}, {2, 4}, {2, 4}, 232).get0();
+      auto& onode = test.create_onode(1996);
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at [0, 0, 0]; insert to left front at stage 2, 1, 0\n");
+      test.split(make_ghobj(1, 1, 1, "ns3", "oid3", 3, 3), onode,
+                 {2u, 2u, true, InsertType::BEGIN}).get0();
+      EXPECT_TRUE(last_split.match_split_pos({0, {0, {0}}}));
+      test.split(make_ghobj(2, 2, 2, "ns1", "oid1", 3, 3), onode,
+                 {2u, 1u, true, InsertType::BEGIN}).get0();
+      EXPECT_TRUE(last_split.match_split_pos({0, {0, {0}}}));
+      test.split(make_ghobj(2, 2, 2, "ns2", "oid2", 1, 1), onode,
+                 {2u, 0u, true, InsertType::BEGIN}).get0();
+      EXPECT_TRUE(last_split.match_split_pos({0, {0, {0}}}));
+    }
+
+    {
+      TestTree test;
+      std::vector<ghobject_t> keys = {
+        make_ghobj(2, 2, 2, "ns3", "oid3", 3, 3),
+        make_ghobj(3, 3, 3, "ns3", "oid3", 3, 3)};
+      std::vector<const onode_t*> values = {
+        &test.create_onode(1360),
+        &test.create_onode(1632)};
+      test.build_tree(keys, values).get0();
+      auto& onode = test.create_onode(1640);
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at [END, END, END]; insert to right at stage 0, 1, 2\n");
+      test.split(make_ghobj(3, 3, 3, "ns3", "oid3", 4, 4), onode,
+                 {0u, 0u, false, InsertType::BEGIN}).get0();
+      EXPECT_TRUE(last_split.match_split_pos({1, {0, {1}}}));
+      test.split(make_ghobj(3, 3, 3, "ns4", "oid4", 3, 3), onode,
+                 {1u, 1u, false, InsertType::BEGIN}).get0();
+      EXPECT_TRUE(last_split.match_split_pos({1, {1, {0}}}));
+      test.split(make_ghobj(4, 4, 4, "ns3", "oid3", 3, 3), onode,
+                 {2u, 2u, false, InsertType::BEGIN}).get0();
+      EXPECT_TRUE(last_split.match_split_pos({2, {0, {0}}}));
+    }
+  });
+}
+
+namespace crimson::os::seastore::onode {
+
+class DummyChildPool {
+  class DummyChildImpl final : public NodeImpl {
+   public:
+    using URef = std::unique_ptr<DummyChildImpl>;
+    DummyChildImpl(const std::set<ghobject_t>& keys, bool is_level_tail, laddr_t laddr)
+        : keys{keys}, _is_level_tail{is_level_tail}, _laddr{laddr} {
+      std::tie(key_view, p_mem_key_view) = build_key_view(*keys.crbegin());
+    }
+    ~DummyChildImpl() override {
+      std::free(p_mem_key_view);
+    }
+
+    const std::set<ghobject_t>& get_keys() const { return keys; }
+
+    void reset(const std::set<ghobject_t>& _keys, bool level_tail) {
+      keys = _keys;
+      _is_level_tail = level_tail;
+      std::free(p_mem_key_view);
+      std::tie(key_view, p_mem_key_view) = build_key_view(*keys.crbegin());
+    }
+
+   public:
+    laddr_t laddr() const override { return _laddr; }
+    bool is_level_tail() const override { return _is_level_tail; }
+
+   protected:
+    field_type_t field_type() const override { return field_type_t::N0; }
+    level_t level() const override { return 0u; }
+    key_view_t get_largest_key_view() const override { return key_view; }
+    void prepare_mutate(context_t) override {
+      ceph_abort("impossible path"); }
+    bool is_empty() const override {
+      ceph_abort("impossible path"); }
+    node_offset_t free_size() const override {
+      ceph_abort("impossible path"); }
+    key_view_t get_key_view(const search_position_t&) const override {
+      ceph_abort("impossible path"); }
+    void next_position(search_position_t&) const override {
+      ceph_abort("impossible path"); }
+    node_stats_t get_stats() const override {
+      ceph_abort("impossible path"); }
+    std::ostream& dump(std::ostream&) const override {
+      ceph_abort("impossible path"); }
+    std::ostream& dump_brief(std::ostream&) const override {
+      ceph_abort("impossible path"); }
+    void validate_layout() const override {
+      ceph_abort("impossible path"); }
+    void test_copy_to(NodeExtentMutable&) const override {
+      ceph_abort("impossible path"); }
+    void test_set_tail(NodeExtentMutable&) override {
+      ceph_abort("impossible path"); }
+
+   private:
+    std::set<ghobject_t> keys;
+    bool _is_level_tail;
+    laddr_t _laddr;
+
+    key_view_t key_view;
+    void* p_mem_key_view;
+  };
+
+  class DummyChild final : public Node {
+   public:
+    ~DummyChild() override = default;
+
+    node_future<> populate_split(
+        context_t c, std::set<Ref<DummyChild>>& splitable_nodes) {
+      ceph_assert(can_split());
+      ceph_assert(splitable_nodes.find(this) != splitable_nodes.end());
+
+      size_t index;
+      const auto& keys = impl->get_keys();
+      if (keys.size() == 2) {
+        index = 1;
+      } else {
+        index = rd() % (keys.size() - 2) + 1;
+      }
+      auto iter = keys.begin();
+      std::advance(iter, index);
+
+      std::set<ghobject_t> left_keys(keys.begin(), iter);
+      std::set<ghobject_t> right_keys(iter, keys.end());
+      bool right_is_tail = impl->is_level_tail();
+      impl->reset(left_keys, false);
+      auto right_child = DummyChild::create_new(right_keys, right_is_tail, pool);
+      if (!can_split()) {
+        splitable_nodes.erase(this);
+      }
+      if (right_child->can_split()) {
+        splitable_nodes.insert(right_child);
+      }
+      return insert_parent(c, right_child);
+    }
+
+    node_future<> insert_and_split(
+        context_t c, const ghobject_t& insert_key,
+        std::set<Ref<DummyChild>>& splitable_nodes) {
+      const auto& keys = impl->get_keys();
+      ceph_assert(keys.size() == 1);
+      auto& key = *keys.begin();
+      ceph_assert(insert_key < key);
+
+      std::set<ghobject_t> new_keys;
+      new_keys.insert(insert_key);
+      new_keys.insert(key);
+      impl->reset(new_keys, impl->is_level_tail());
+
+      splitable_nodes.clear();
+      splitable_nodes.insert(this);
+      auto fut = populate_split(c, splitable_nodes);
+      ceph_assert(splitable_nodes.size() == 0);
+      return fut;
+    }
+
+    bool match_pos(const search_position_t& pos) const {
+      ceph_assert(!is_root());
+      return pos == parent_info().position;
+    }
+
+    static Ref<DummyChild> create(
+        const std::set<ghobject_t>& keys, bool is_level_tail,
+        laddr_t addr, DummyChildPool& pool) {
+      auto ref_impl = std::make_unique<DummyChildImpl>(keys, is_level_tail, addr);
+      return new DummyChild(ref_impl.get(), std::move(ref_impl), pool);
+    }
+
+    static Ref<DummyChild> create_new(
+        const std::set<ghobject_t>& keys, bool is_level_tail, DummyChildPool& pool) {
+      static laddr_t seed = 0;
+      return create(keys, is_level_tail, seed++, pool);
+    }
+
+    static node_future<Ref<DummyChild>> create_initial(
+        context_t c, const std::set<ghobject_t>& keys,
+        DummyChildPool& pool, RootNodeTracker& root_tracker) {
+      auto initial = create_new(keys, true, pool);
+      return c.nm.get_super(c.t, root_tracker
+      ).safe_then([c, &pool, initial](auto super) {
+        initial->make_root_new(c, std::move(super));
+        return initial->upgrade_root(c).safe_then([initial] {
+          return initial;
+        });
+      });
+    }
+
+   protected:
+    node_future<> test_clone_non_root(
+        context_t, Ref<InternalNode> new_parent) const override {
+      ceph_assert(!is_root());
+      auto p_pool_clone = pool.pool_clone_in_progress;
+      ceph_assert(p_pool_clone != nullptr);
+      auto clone = create(
+          impl->get_keys(), impl->is_level_tail(), impl->laddr(), *p_pool_clone);
+      clone->as_child(parent_info().position, new_parent);
+      return node_ertr::now();
+    }
+    node_future<Ref<tree_cursor_t>> lookup_smallest(context_t) override {
+      ceph_abort("impossible path"); }
+    node_future<Ref<tree_cursor_t>> lookup_largest(context_t) override {
+      ceph_abort("impossible path"); }
+    node_future<> test_clone_root(context_t, RootNodeTracker&) const override {
+      ceph_abort("impossible path"); }
+    node_future<search_result_t> lower_bound_tracked(
+        context_t, const key_hobj_t&, MatchHistory&) override {
+      ceph_abort("impossible path"); }
+    node_future<> do_get_tree_stats(context_t, tree_stats_t&) override {
+      ceph_abort("impossible path"); }
+
+   private:
+    DummyChild(DummyChildImpl* impl, DummyChildImpl::URef&& ref, DummyChildPool& pool)
+      : Node(std::move(ref)), impl{impl}, pool{pool} {
+      pool.track_node(this);
+    }
+
+    bool can_split() const { return impl->get_keys().size() > 1; }
+
+    DummyChildImpl* impl;
+    DummyChildPool& pool;
+    mutable std::random_device rd;
+  };
+
+ public:
+  using node_ertr = Node::node_ertr;
+  template <class ValueT=void>
+  using node_future = Node::node_future<ValueT>;
+
+  DummyChildPool() = default;
+  ~DummyChildPool() { reset(); }
+
+  node_future<> build_tree(const std::set<ghobject_t>& keys) {
+    reset();
+
+    // create tree
+    auto ref_nm = NodeExtentManager::create_dummy(IS_DUMMY_SYNC);
+    p_nm = ref_nm.get();
+    p_btree.emplace(std::move(ref_nm));
+    return DummyChild::create_initial(get_context(), keys, *this, *p_btree->root_tracker
+    ).safe_then([this](auto initial_child) {
+      // split
+      splitable_nodes.insert(initial_child);
+      return crimson::do_until([this] {
+        if (splitable_nodes.empty()) {
+          return node_ertr::make_ready_future<bool>(true);
+        }
+        auto index = rd() % splitable_nodes.size();
+        auto iter = splitable_nodes.begin();
+        std::advance(iter, index);
+        Ref<DummyChild> child = *iter;
+        return child->populate_split(get_context(), splitable_nodes
+        ).safe_then([] {
+          return node_ertr::make_ready_future<bool>(false);
+        });
+      });
+    }).safe_then([this] {
+      //std::ostringstream oss;
+      //p_btree->dump(t(), oss);
+      //logger().info("\n{}\n", oss.str());
+      return p_btree->height(t());
+    }).safe_then([](auto height) {
+      ceph_assert(height == 2);
+    });
+  }
+
+  seastar::future<> test_split(ghobject_t key, search_position_t pos,
+                               const split_expectation_t& expected) {
+    return seastar::async([this, key, pos, expected] {
+      logger().info("insert {} at {}:", key_hobj_t(key), pos);
+      DummyChildPool pool_clone;
+      pool_clone_in_progress = &pool_clone;
+      auto ref_nm = NodeExtentManager::create_dummy(IS_DUMMY_SYNC);
+      pool_clone.p_nm = ref_nm.get();
+      pool_clone.p_btree.emplace(std::move(ref_nm));
+      pool_clone.p_btree->test_clone_from(
+        pool_clone.t(), t(), *p_btree).unsafe_get0();
+      pool_clone_in_progress = nullptr;
+      auto node_to_split = pool_clone.get_node_by_pos(pos);
+      node_to_split->insert_and_split(
+        pool_clone.get_context(), key, pool_clone.splitable_nodes).unsafe_get0();
+      std::ostringstream oss;
+      pool_clone.p_btree->dump(pool_clone.t(), oss);
+      logger().info("dump new root:\n{}", oss.str());
+      EXPECT_EQ(pool_clone.p_btree->height(pool_clone.t()).unsafe_get0(), 3);
+      EXPECT_TRUE(last_split.match(expected));
+    });
+  }
+
+ private:
+  void reset() {
+    ceph_assert(pool_clone_in_progress == nullptr);
+    if (tracked_children.size()) {
+      ceph_assert(!p_btree->test_is_clean());
+      tracked_children.clear();
+      ceph_assert(p_btree->test_is_clean());
+      p_nm = nullptr;
+      p_btree.reset();
+    } else {
+      ceph_assert(!p_btree.has_value());
+    }
+    splitable_nodes.clear();
+  }
+
+  void track_node(Ref<DummyChild> node) {
+    ceph_assert(tracked_children.find(node) == tracked_children.end());
+    tracked_children.insert(node);
+  }
+
+  Ref<DummyChild> get_node_by_pos(const search_position_t& pos) const {
+    auto iter = std::find_if(
+        tracked_children.begin(), tracked_children.end(), [&pos](auto& child) {
+      return child->match_pos(pos);
+    });
+    ceph_assert(iter != tracked_children.end());
+    return *iter;
+  }
+
+  context_t get_context() {
+    ceph_assert(p_nm != nullptr);
+    return {*p_nm, t()};
+  }
+
+  Transaction& t() const { return *ref_t; }
+
+  std::set<Ref<DummyChild>> tracked_children;
+  std::optional<Btree> p_btree;
+  NodeExtentManager* p_nm = nullptr;
+  TransactionRef ref_t = make_transaction();
+
+  std::random_device rd;
+  std::set<Ref<DummyChild>> splitable_nodes;
+
+  DummyChildPool* pool_clone_in_progress = nullptr;
+};
+
+}
+
+TEST_F(c_dummy_test_t, 5_split_internal_node)
+{
+  run_async([this] {
+    DummyChildPool pool;
+    {
+      logger().info("\n---------------------------------------------"
+                    "\nbefore internal node insert:\n");
+      auto padding = std::string(250, '_');
+      auto keys = build_key_set({2, 6}, {2, 5}, {2, 5}, padding, true);
+      keys.erase(make_ghobj(2, 2, 2, "ns2", "oid2" + padding, 2, 2));
+      keys.erase(make_ghobj(2, 2, 2, "ns2", "oid2" + padding, 3, 3));
+      keys.erase(make_ghobj(2, 2, 2, "ns2", "oid2" + padding, 4, 4));
+      keys.erase(make_ghobj(5, 5, 5, "ns4", "oid4" + padding, 2, 2));
+      keys.erase(make_ghobj(5, 5, 5, "ns4", "oid4" + padding, 3, 3));
+      keys.erase(make_ghobj(5, 5, 5, "ns4", "oid4" + padding, 4, 4));
+      auto padding_s = std::string(257, '_');
+      keys.insert(make_ghobj(2, 2, 2, "ns2", "oid2" + padding_s, 2, 2));
+      keys.insert(make_ghobj(2, 2, 2, "ns2", "oid2" + padding_s, 3, 3));
+      keys.insert(make_ghobj(2, 2, 2, "ns2", "oid2" + padding_s, 4, 4));
+      auto padding_e = std::string(248, '_');
+      keys.insert(make_ghobj(5, 5, 5, "ns4", "oid4" + padding_e, 2, 2));
+      keys.insert(make_ghobj(5, 5, 5, "ns4", "oid4" + padding_e, 3, 3));
+      keys.insert(make_ghobj(5, 5, 5, "ns4", "oid4" + padding_e, 4, 4));
+      pool.build_tree(keys).unsafe_get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to right front at stage 0, 1, 2, 1, 0\n");
+      pool.test_split(make_ghobj(3, 3, 3, "ns4", "oid4" + padding, 5, 5), {2, {0, {0}}},
+                      {2u, 0u, false, InsertType::BEGIN}).get();
+      pool.test_split(make_ghobj(3, 3, 3, "ns5", "oid5", 3, 3), {2, {0, {0}}},
+                      {2u, 1u, false, InsertType::BEGIN}).get();
+      pool.test_split(make_ghobj(3, 4, 4, "ns3", "oid3", 3, 3), {2, {0, {0}}},
+                      {2u, 2u, false, InsertType::BEGIN}).get();
+      pool.test_split(make_ghobj(4, 4, 4, "ns1", "oid1", 3, 3), {2, {0, {0}}},
+                      {2u, 1u, false, InsertType::BEGIN}).get();
+      pool.test_split(make_ghobj(4, 4, 4, "ns2", "oid2" + padding, 1, 1), {2, {0, {0}}},
+                      {2u, 0u, false, InsertType::BEGIN}).get();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to right middle at stage 0, 1, 2, 1, 0\n");
+      pool.test_split(make_ghobj(4, 4, 4, "ns4", "oid4" + padding, 5, 5), {3, {0, {0}}},
+                      {2u, 0u, false, InsertType::MID}).get();
+      pool.test_split(make_ghobj(4, 4, 4, "ns5", "oid5", 3, 3), {3, {0, {0}}},
+                      {2u, 1u, false, InsertType::MID}).get();
+      pool.test_split(make_ghobj(4, 4, 5, "ns3", "oid3", 3, 3), {3, {0, {0}}},
+                      {2u, 2u, false, InsertType::MID}).get();
+      pool.test_split(make_ghobj(5, 5, 5, "ns1", "oid1", 3, 3), {3, {0, {0}}},
+                      {2u, 1u, false, InsertType::MID}).get();
+      pool.test_split(make_ghobj(5, 5, 5, "ns2", "oid2" + padding, 1, 1), {3, {0, {0}}},
+                      {2u, 0u, false, InsertType::MID}).get();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to right back at stage 0, 1, 2\n");
+      pool.test_split(make_ghobj(5, 5, 5, "ns4", "oid4" + padding_e, 5, 5), search_position_t::end(),
+                      {2u, 0u, false, InsertType::LAST}).get();
+      pool.test_split(make_ghobj(5, 5, 5, "ns5", "oid5", 3, 3), search_position_t::end(),
+                      {2u, 1u, false, InsertType::LAST}).get();
+      pool.test_split(make_ghobj(6, 6, 6, "ns3", "oid3", 3, 3), search_position_t::end(),
+                      {2u, 2u, false, InsertType::LAST}).get();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 0; insert to left front at stage 2, 1, 0\n");
+      pool.test_split(make_ghobj(1, 1, 1, "ns3", "oid3", 3, 3), {0, {0, {0}}},
+                      {0u, 2u, true, InsertType::BEGIN}).get();
+      pool.test_split(make_ghobj(2, 2, 2, "ns1", "oid1", 3, 3), {0, {0, {0}}},
+                      {0u, 1u, true, InsertType::BEGIN}).get();
+      pool.test_split(make_ghobj(2, 2, 2, "ns2", "oid2" + padding_s, 1, 1), {0, {0, {0}}},
+                      {0u, 0u, true, InsertType::BEGIN}).get();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 0; insert to left middle at stage 0, 1, 2, 1, 0\n");
+      pool.test_split(make_ghobj(2, 2, 2, "ns4", "oid4" + padding, 5, 5), {1, {0, {0}}},
+                      {0u, 0u, true, InsertType::MID}).get();
+      pool.test_split(make_ghobj(2, 2, 2, "ns5", "oid5", 3, 3), {1, {0, {0}}},
+                      {0u, 1u, true, InsertType::MID}).get();
+      pool.test_split(make_ghobj(2, 2, 3, "ns3", "oid3" + std::string(80, '_'), 3, 3), {1, {0, {0}}},
+                      {0u, 2u, true, InsertType::MID}).get();
+      pool.test_split(make_ghobj(3, 3, 3, "ns1", "oid1", 3, 3), {1, {0, {0}}},
+                      {0u, 1u, true, InsertType::MID}).get();
+      pool.test_split(make_ghobj(3, 3, 3, "ns2", "oid2" + padding, 1, 1), {1, {0, {0}}},
+                      {0u, 0u, true, InsertType::MID}).get();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 0; insert to left back at stage 0\n");
+      pool.test_split(make_ghobj(3, 3, 3, "ns4", "oid4" + padding, 3, 4), {1, {2, {2}}},
+                      {0u, 0u, true, InsertType::LAST}).get();
+    }
+
+    {
+      logger().info("\n---------------------------------------------"
+                    "\nbefore internal node insert (1):\n");
+      auto padding = std::string(244, '_');
+      auto keys = build_key_set({2, 6}, {2, 5}, {2, 5}, padding, true);
+      keys.insert(make_ghobj(5, 5, 5, "ns4", "oid4" + padding, 5, 5));
+      keys.insert(make_ghobj(5, 5, 5, "ns4", "oid4" + padding, 6, 6));
+      keys.insert(make_ghobj(5, 5, 5, "ns4", "oid4" + padding, 7, 7));
+      pool.build_tree(keys).unsafe_get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to left back at stage 0, 1, 2, 1\n");
+      pool.test_split(make_ghobj(3, 3, 3, "ns4", "oid4" + padding, 5, 5), {2, {0, {0}}},
+                      {2u, 0u, true, InsertType::LAST}).get();
+      pool.test_split(make_ghobj(3, 3, 3, "ns5", "oid5", 3, 3), {2, {0, {0}}},
+                      {2u, 1u, true, InsertType::LAST}).get();
+      pool.test_split(make_ghobj(3, 4, 4, "n", "o", 3, 3), {2, {0, {0}}},
+                      {2u, 2u, true, InsertType::LAST}).get();
+      pool.test_split(make_ghobj(4, 4, 4, "n", "o", 3, 3), {2, {0, {0}}},
+                      {2u, 1u, true, InsertType::LAST}).get();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to left middle at stage 2\n");
+      pool.test_split(make_ghobj(2, 3, 3, "n", "o", 3, 3), {1, {0, {0}}},
+                      {2u, 2u, true, InsertType::MID}).get();
+    }
+
+    {
+      logger().info("\n---------------------------------------------"
+                    "\nbefore internal node insert (2):\n");
+      auto padding = std::string(243, '_');
+      auto keys = build_key_set({2, 6}, {2, 5}, {2, 5}, padding, true);
+      keys.insert(make_ghobj(4, 4, 4, "n", "o", 3, 3));
+      keys.insert(make_ghobj(5, 5, 5, "ns4", "oid4" + padding, 5, 5));
+      keys.insert(make_ghobj(5, 5, 5, "ns4", "oid4" + padding, 6, 6));
+      pool.build_tree(keys).unsafe_get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 2; insert to left back at stage (0, 1, 2, 1,) 0\n");
+      pool.test_split(make_ghobj(4, 4, 4, "n", "o", 2, 2), {2, {0, {0}}},
+                      {2u, 0u, true, InsertType::LAST}).get();
+    }
+
+    {
+      logger().info("\n---------------------------------------------"
+                    "\nbefore internal node insert (3):\n");
+      auto padding = std::string(420, '_');
+      auto keys = build_key_set({2, 5}, {2, 5}, {2, 5}, padding, true);
+      keys.erase(make_ghobj(4, 4, 4, "ns4", "oid4" + padding, 2, 2));
+      keys.erase(make_ghobj(4, 4, 4, "ns4", "oid4" + padding, 3, 3));
+      keys.erase(make_ghobj(4, 4, 4, "ns4", "oid4" + padding, 4, 4));
+      pool.build_tree(keys).unsafe_get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 1; insert to right front at stage 0, 1, 0\n");
+      pool.test_split(make_ghobj(3, 3, 3, "ns2", "oid2" + padding, 5, 5), {1, {1, {0}}},
+                      {1u, 0u, false, InsertType::BEGIN}).get();
+      pool.test_split(make_ghobj(3, 3, 3, "ns2", "oid3", 3, 3), {1, {1, {0}}},
+                      {1u, 1u, false, InsertType::BEGIN}).get();
+      pool.test_split(make_ghobj(3, 3, 3, "ns3", "oid3" + padding, 1, 1), {1, {1, {0}}},
+                      {1u, 0u, false, InsertType::BEGIN}).get();
+    }
+
+    {
+      logger().info("\n---------------------------------------------"
+                    "\nbefore internal node insert (4):\n");
+      auto padding = std::string(361, '_');
+      auto keys = build_key_set({2, 5}, {2, 5}, {2, 5}, padding, true);
+      keys.erase(make_ghobj(2, 2, 2, "ns2", "oid2" + padding, 2, 2));
+      keys.erase(make_ghobj(2, 2, 2, "ns2", "oid2" + padding, 3, 3));
+      keys.erase(make_ghobj(2, 2, 2, "ns2", "oid2" + padding, 4, 4));
+      auto padding_s = std::string(387, '_');
+      keys.insert(make_ghobj(2, 2, 2, "ns2", "oid2" + padding_s, 2, 2));
+      keys.insert(make_ghobj(2, 2, 2, "ns2", "oid2" + padding_s, 3, 3));
+      keys.insert(make_ghobj(2, 2, 2, "ns2", "oid2" + padding_s, 4, 4));
+      pool.build_tree(keys).unsafe_get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 1; insert to left back at stage 0, 1\n");
+      pool.test_split(make_ghobj(3, 3, 3, "ns2", "oid2" + padding, 5, 5), {1, {1, {0}}},
+                      {1u, 0u, true, InsertType::LAST}).get();
+      pool.test_split(make_ghobj(3, 3, 3, "ns2", "oid3", 3, 3), {1, {1, {0}}},
+                      {1u, 1u, true, InsertType::LAST}).get();
+    }
+
+    {
+      logger().info("\n---------------------------------------------"
+                    "\nbefore internal node insert (5):\n");
+      auto padding = std::string(412, '_');
+      auto keys = build_key_set({2, 5}, {2, 5}, {2, 5}, padding);
+      keys.insert(make_ghobj(3, 3, 3, "ns2", "oid3", 3, 3));
+      keys.insert(make_ghobj(4, 4, 4, "ns3", "oid3" + padding, 5, 5));
+      keys.insert(make_ghobj(9, 9, 9, "ns~last", "oid~last", 9, 9));
+      keys.erase(make_ghobj(4, 4, 4, "ns4", "oid4" + padding, 2, 2));
+      keys.erase(make_ghobj(4, 4, 4, "ns4", "oid4" + padding, 3, 3));
+      keys.erase(make_ghobj(4, 4, 4, "ns4", "oid4" + padding, 4, 4));
+      pool.build_tree(keys).unsafe_get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 1; insert to left back at stage (0, 1,) 0\n");
+      pool.test_split(make_ghobj(3, 3, 3, "ns2", "oid3", 2, 2), {1, {1, {0}}},
+                      {1u, 0u, true, InsertType::LAST}).get();
+    }
+
+    {
+      logger().info("\n---------------------------------------------"
+                    "\nbefore internal node insert (6):\n");
+      auto padding = std::string(328, '_');
+      auto keys = build_key_set({2, 5}, {2, 5}, {2, 5}, padding);
+      keys.insert(make_ghobj(5, 5, 5, "ns3", "oid3" + std::string(271, '_'), 3, 3));
+      keys.insert(make_ghobj(9, 9, 9, "ns~last", "oid~last", 9, 9));
+      pool.build_tree(keys).unsafe_get0();
+
+      logger().info("\n---------------------------------------------"
+                    "\nsplit at stage 0; insert to right front at stage 0\n");
+      pool.test_split(make_ghobj(3, 3, 3, "ns3", "oid3" + padding, 2, 3), {1, {1, {1}}},
+                      {0u, 0u, false, InsertType::BEGIN}).get();
+    }
+
+    // Impossible to split at {0, 0, 0}
+    // Impossible to split at [END, END, END]
+  });
+}
+
+struct d_seastore_tm_test_t :
+    public seastar_test_suite_t, TMTestState {
+  seastar::future<> set_up_fut() override final {
+    return tm_setup();
+  }
+  seastar::future<> tear_down_fut() override final {
+    return tm_teardown();
+  }
+};
+
+TEST_F(d_seastore_tm_test_t, 6_random_insert_leaf_node)
+{
+  run_async([this] {
+    constexpr bool TEST_SEASTORE = true;
+    constexpr bool TRACK_CURSORS = true;
+    KVPool kvs{{8, 11, 64, 256, 301, 320},
+               {8, 16, 128, 512, 576, 640},
+               {0, 32}, {0, 10}, {0, 4}};
+    auto tree = std::make_unique<TreeBuilder<TRACK_CURSORS>>(kvs,
+        (TEST_SEASTORE ? NodeExtentManager::create_seastore(*tm)
+                       : NodeExtentManager::create_dummy(IS_DUMMY_SYNC)));
+    {
+      auto t = tm->create_transaction();
+      tree->bootstrap(*t).unsafe_get();
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+    {
+      auto t = tm->create_transaction();
+      tree->insert(*t).unsafe_get();
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+    {
+      auto t = tm->create_transaction();
+      tree->get_stats(*t).unsafe_get();
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+    if constexpr (TEST_SEASTORE) {
+      logger().info("seastore replay begin");
+      restart();
+      tree->reload(NodeExtentManager::create_seastore(*tm));
+      logger().info("seastore replay end");
+    }
+    {
+      // Note: tm->create_weak_transaction() can also work, but too slow.
+      auto t = tm->create_transaction();
+      tree->validate(*t).unsafe_get();
+    }
+    tree.reset();
+  });
+}
diff --git a/src/test/crimson/seastore/test_block.cc b/src/test/crimson/seastore/test_block.cc
new file mode 100644
index 000000000..f3d6531bd
--- /dev/null
+++ b/src/test/crimson/seastore/test_block.cc
@@ -0,0 +1,25 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/crimson/seastore/test_block.h"
+
+namespace crimson::os::seastore {
+
+
+ceph::bufferlist TestBlock::get_delta() {
+  ceph::bufferlist bl;
+  encode(delta, bl);
+  return bl;
+}
+
+
+void TestBlock::apply_delta(const ceph::bufferlist &bl) {
+  auto biter = bl.begin();
+  decltype(delta) deltas;
+  decode(deltas, biter);
+  for (auto &&d : deltas) {
+    set_contents(d.val, d.offset, d.len);
+  }
+}
+
+}
diff --git a/src/test/crimson/seastore/test_block.h b/src/test/crimson/seastore/test_block.h
new file mode 100644
index 000000000..44ec65a23
--- /dev/null
+++ b/src/test/crimson/seastore/test_block.h
@@ -0,0 +1,147 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include <random>
+
+#include "crimson/os/seastore/transaction_manager.h"
+
+namespace crimson::os::seastore {
+
+struct test_extent_desc_t {
+  size_t len = 0;
+  unsigned checksum = 0;
+
+  bool operator==(const test_extent_desc_t &rhs) const {
+    return (len == rhs.len &&
+	    checksum == rhs.checksum);
+  }
+  bool operator!=(const test_extent_desc_t &rhs) const {
+    return !(*this == rhs);
+  }
+};
+
+struct test_block_delta_t {
+  int8_t val = 0;
+  uint16_t offset = 0;
+  uint16_t len = 0;
+
+
+  DENC(test_block_delta_t, v, p) {
+    DENC_START(1, 1, p);
+    denc(v.val, p);
+    denc(v.offset, p);
+    denc(v.len, p);
+    DENC_FINISH(p);
+  }
+};
+
+inline std::ostream &operator<<(
+  std::ostream &lhs, const test_extent_desc_t &rhs) {
+  return lhs << "test_extent_desc_t(len=" << rhs.len
+	     << ", checksum=" << rhs.checksum << ")";
+}
+
+struct TestBlock : crimson::os::seastore::LogicalCachedExtent {
+  constexpr static segment_off_t SIZE = 4<<10;
+  using Ref = TCachedExtentRef<TestBlock>;
+
+  std::vector<test_block_delta_t> delta = {};
+
+  TestBlock(ceph::bufferptr &&ptr)
+    : LogicalCachedExtent(std::move(ptr)) {}
+  TestBlock(const TestBlock &other)
+    : LogicalCachedExtent(other) {}
+
+  CachedExtentRef duplicate_for_write() final {
+    return CachedExtentRef(new TestBlock(*this));
+  };
+
+  static constexpr extent_types_t TYPE = extent_types_t::TEST_BLOCK;
+  extent_types_t get_type() const final {
+    return TYPE;
+  }
+
+  ceph::bufferlist get_delta() final;
+
+  void set_contents(char c, uint16_t offset, uint16_t len) {
+    ::memset(get_bptr().c_str() + offset, c, len);
+    delta.push_back({c, offset, len});
+  }
+
+  void set_contents(char c) {
+    set_contents(c, 0, get_length());
+  }
+
+  test_extent_desc_t get_desc() {
+    return { get_length(), get_crc32c() };
+  }
+
+  void apply_delta(const ceph::bufferlist &bl) final;
+};
+using TestBlockRef = TCachedExtentRef<TestBlock>;
+
+struct TestBlockPhysical : crimson::os::seastore::CachedExtent{
+  constexpr static segment_off_t SIZE = 4<<10;
+  using Ref = TCachedExtentRef<TestBlockPhysical>;
+
+  std::vector<test_block_delta_t> delta = {};
+
+  TestBlockPhysical(ceph::bufferptr &&ptr)
+    : CachedExtent(std::move(ptr)) {}
+  TestBlockPhysical(const TestBlock &other)
+    : CachedExtent(other) {}
+
+  CachedExtentRef duplicate_for_write() final {
+    return CachedExtentRef(new TestBlockPhysical(*this));
+  };
+
+  static constexpr extent_types_t TYPE = extent_types_t::TEST_BLOCK_PHYSICAL;
+  extent_types_t get_type() const final {
+    return TYPE;
+  }
+
+  void set_contents(char c, uint16_t offset, uint16_t len) {
+    ::memset(get_bptr().c_str() + offset, c, len);
+  }
+
+  void set_contents(char c) {
+    set_contents(c, 0, get_length());
+  }
+
+  ceph::bufferlist get_delta() final { return ceph::bufferlist(); }
+
+  void apply_delta_and_adjust_crc(paddr_t, const ceph::bufferlist &bl) final {}
+};
+using TestBlockPhysicalRef = TCachedExtentRef<TestBlockPhysical>;
+
+struct test_block_mutator_t {
+  std::uniform_int_distribution<int8_t>
+  contents_distribution = std::uniform_int_distribution<int8_t>(
+    std::numeric_limits<int8_t>::min(),
+    std::numeric_limits<int8_t>::max());
+
+  std::uniform_int_distribution<uint16_t>
+  offset_distribution = std::uniform_int_distribution<uint16_t>(
+    0, TestBlock::SIZE - 1);
+
+  std::uniform_int_distribution<uint16_t> length_distribution(uint16_t offset) {
+    return std::uniform_int_distribution<uint16_t>(
+      0, TestBlock::SIZE - offset - 1);
+  }
+
+
+  template <typename generator_t>
+  void mutate(TestBlock &block, generator_t &gen) {
+    auto offset = offset_distribution(gen);
+    block.set_contents(
+      contents_distribution(gen),
+      offset,
+      length_distribution(offset)(gen));
+  }
+};
+
+}
+
+WRITE_CLASS_DENC_BOUNDED(crimson::os::seastore::test_block_delta_t)
diff --git a/src/test/crimson/seastore/test_btree_lba_manager.cc b/src/test/crimson/seastore/test_btree_lba_manager.cc
new file mode 100644
index 000000000..60d5c3497
--- /dev/null
+++ b/src/test/crimson/seastore/test_btree_lba_manager.cc
@@ -0,0 +1,429 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/crimson/gtest_seastar.h"
+
+#include "crimson/common/log.h"
+
+#include "crimson/os/seastore/journal.h"
+#include "crimson/os/seastore/cache.h"
+#include "crimson/os/seastore/segment_manager/ephemeral.h"
+#include "crimson/os/seastore/lba_manager/btree/btree_lba_manager.h"
+
+#include "test/crimson/seastore/test_block.h"
+
+namespace {
+  [[maybe_unused]] seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_test);
+  }
+}
+
+using namespace crimson;
+using namespace crimson::os;
+using namespace crimson::os::seastore;
+using namespace crimson::os::seastore::lba_manager;
+using namespace crimson::os::seastore::lba_manager::btree;
+
+struct btree_lba_manager_test :
+  public seastar_test_suite_t, JournalSegmentProvider {
+  segment_manager::EphemeralSegmentManagerRef segment_manager;
+  Journal journal;
+  Cache cache;
+  BtreeLBAManagerRef lba_manager;
+
+  const size_t block_size;
+
+  btree_lba_manager_test()
+    : segment_manager(segment_manager::create_test_ephemeral()),
+      journal(*segment_manager),
+      cache(*segment_manager),
+      lba_manager(new BtreeLBAManager(*segment_manager, cache)),
+      block_size(segment_manager->get_block_size())
+  {
+    journal.set_segment_provider(this);
+  }
+
+  segment_id_t next = 0;
+  get_segment_ret get_segment() final {
+    return get_segment_ret(
+      get_segment_ertr::ready_future_marker{},
+      next++);
+  }
+
+  journal_seq_t get_journal_tail_target() const final { return journal_seq_t{}; }
+  void update_journal_tail_committed(journal_seq_t committed) final {}
+
+  auto submit_transaction(TransactionRef t)
+  {
+    auto record = cache.try_construct_record(*t);
+    if (!record) {
+      ceph_assert(0 == "cannot fail");
+    }
+
+    return journal.submit_record(std::move(*record)).safe_then(
+      [this, t=std::move(t)](auto p) mutable {
+	auto [addr, seq] = p;
+	cache.complete_commit(*t, addr, seq);
+	lba_manager->complete_transaction(*t);
+      },
+      crimson::ct_error::assert_all{});
+  }
+
+  seastar::future<> set_up_fut() final {
+    return segment_manager->init(
+    ).safe_then([this] {
+      return journal.open_for_write();
+    }).safe_then([this](auto addr) {
+      return seastar::do_with(
+	make_transaction(),
+	[this](auto &transaction) {
+	  cache.init();
+	  return cache.mkfs(*transaction
+	  ).safe_then([this, &transaction] {
+	    return lba_manager->mkfs(*transaction);
+	  }).safe_then([this, &transaction] {
+	    return submit_transaction(std::move(transaction));
+	  });
+	});
+    }).handle_error(
+      crimson::ct_error::all_same_way([] {
+	ceph_assert(0 == "error");
+      })
+    );
+  }
+
+  seastar::future<> tear_down_fut() final {
+    return cache.close(
+    ).safe_then([this] {
+      return journal.close();
+    }).handle_error(
+      crimson::ct_error::all_same_way([] {
+	ASSERT_FALSE("Unable to close");
+      })
+    );
+  }
+
+
+  struct test_extent_t {
+    paddr_t addr;
+    size_t len = 0;
+    unsigned refcount = 0;
+  };
+  using test_lba_mapping_t = std::map<laddr_t, test_extent_t>;
+  test_lba_mapping_t test_lba_mappings;
+  struct test_transaction_t {
+    TransactionRef t;
+    test_lba_mapping_t mappings;
+  };
+
+  auto create_transaction() {
+    auto t = test_transaction_t{
+      make_transaction(),
+      test_lba_mappings
+    };
+    cache.alloc_new_extent<TestBlockPhysical>(*t.t, TestBlockPhysical::SIZE);
+    return t;
+  }
+
+  auto create_weak_transaction() {
+    auto t = test_transaction_t{
+      make_weak_transaction(),
+      test_lba_mappings
+    };
+    return t;
+  }
+
+  void submit_test_transaction(test_transaction_t t) {
+    submit_transaction(std::move(t.t)).get0();
+    test_lba_mappings.swap(t.mappings);
+  }
+
+  auto get_overlap(test_transaction_t &t, laddr_t addr, size_t len) {
+    auto bottom = t.mappings.upper_bound(addr);
+    if (bottom != t.mappings.begin())
+      --bottom;
+    if (bottom != t.mappings.end() &&
+	bottom->first + bottom->second.len <= addr)
+      ++bottom;
+
+    auto top = t.mappings.lower_bound(addr + len);
+    return std::make_pair(
+      bottom,
+      top
+    );
+  }
+
+  segment_off_t next_off = 0;
+  paddr_t get_paddr() {
+    next_off += block_size;
+    return make_fake_paddr(next_off);
+  }
+
+  auto alloc_mapping(
+    test_transaction_t &t,
+    laddr_t hint,
+    size_t len,
+    paddr_t paddr) {
+    auto ret = lba_manager->alloc_extent(*t.t, hint, len, paddr).unsafe_get0();
+    logger().debug("alloc'd: {}", *ret);
+    EXPECT_EQ(len, ret->get_length());
+    auto [b, e] = get_overlap(t, ret->get_laddr(), len);
+    EXPECT_EQ(b, e);
+    t.mappings.emplace(
+      std::make_pair(
+	ret->get_laddr(),
+	test_extent_t{
+	  ret->get_paddr(),
+	  ret->get_length(),
+	  1
+        }
+      ));
+    return ret;
+  }
+
+  auto set_mapping(
+    test_transaction_t &t,
+    laddr_t addr,
+    size_t len,
+    paddr_t paddr) {
+    auto [b, e] = get_overlap(t, addr, len);
+    EXPECT_EQ(b, e);
+
+    auto ret = lba_manager->set_extent(*t.t, addr, len, paddr).unsafe_get0();
+    EXPECT_EQ(addr, ret->get_laddr());
+    EXPECT_EQ(len, ret->get_length());
+    EXPECT_EQ(paddr, ret->get_paddr());
+    t.mappings.emplace(
+      std::make_pair(
+	ret->get_laddr(),
+	test_extent_t{
+	  ret->get_paddr(),
+	  ret->get_length(),
+	  1
+        }
+      ));
+    return ret;
+  }
+
+  auto decref_mapping(
+    test_transaction_t &t,
+    laddr_t addr) {
+    return decref_mapping(t, t.mappings.find(addr));
+  }
+
+  void decref_mapping(
+    test_transaction_t &t,
+    test_lba_mapping_t::iterator target) {
+    ceph_assert(target != t.mappings.end());
+    ceph_assert(target->second.refcount > 0);
+    target->second.refcount--;
+
+    auto refcnt = lba_manager->decref_extent(
+      *t.t,
+      target->first).unsafe_get0().refcount;
+    EXPECT_EQ(refcnt, target->second.refcount);
+    if (target->second.refcount == 0) {
+      t.mappings.erase(target);
+    }
+  }
+
+  auto incref_mapping(
+    test_transaction_t &t,
+    laddr_t addr) {
+    return incref_mapping(t, t.mappings.find(addr));
+  }
+
+  void incref_mapping(
+    test_transaction_t &t,
+    test_lba_mapping_t::iterator target) {
+    ceph_assert(target->second.refcount > 0);
+    target->second.refcount++;
+    auto refcnt = lba_manager->incref_extent(
+      *t.t,
+      target->first).unsafe_get0().refcount;
+    EXPECT_EQ(refcnt, target->second.refcount);
+  }
+
+  std::vector<laddr_t> get_mapped_addresses() {
+    std::vector<laddr_t> addresses;
+    addresses.reserve(test_lba_mappings.size());
+    for (auto &i: test_lba_mappings) {
+      addresses.push_back(i.first);
+    }
+    return addresses;
+  }
+
+  std::vector<laddr_t> get_mapped_addresses(test_transaction_t &t) {
+    std::vector<laddr_t> addresses;
+    addresses.reserve(t.mappings.size());
+    for (auto &i: t.mappings) {
+      addresses.push_back(i.first);
+    }
+    return addresses;
+  }
+
+  void check_mappings() {
+    auto t = create_transaction();
+    check_mappings(t);
+  }
+
+  void check_mappings(test_transaction_t &t) {
+    for (auto &&i: t.mappings) {
+      auto ret_list = lba_manager->get_mapping(
+	*t.t, i.first, i.second.len
+      ).unsafe_get0();
+      EXPECT_EQ(ret_list.size(), 1);
+      auto &ret = *ret_list.begin();
+      EXPECT_EQ(i.second.addr, ret->get_paddr());
+      EXPECT_EQ(i.first, ret->get_laddr());
+      EXPECT_EQ(i.second.len, ret->get_length());
+    }
+    lba_manager->scan_mappings(
+      *t.t,
+      0,
+      L_ADDR_MAX,
+      [iter=t.mappings.begin(), &t](auto l, auto p, auto len) mutable {
+	EXPECT_NE(iter, t.mappings.end());
+	EXPECT_EQ(l, iter->first);
+	EXPECT_EQ(p, iter->second.addr);
+	EXPECT_EQ(len, iter->second.len);
+	++iter;
+      }).unsafe_get();
+  }
+};
+
+TEST_F(btree_lba_manager_test, basic)
+{
+  run_async([this] {
+    laddr_t laddr = 0x12345678 * block_size;
+    {
+      // write initial mapping
+      auto t = create_transaction();
+      check_mappings(t);  // check in progress transaction sees mapping
+      check_mappings();   // check concurrent does not
+      auto ret = alloc_mapping(t, laddr, block_size, get_paddr());
+      submit_test_transaction(std::move(t));
+    }
+    check_mappings();     // check new transaction post commit sees it
+  });
+}
+
+TEST_F(btree_lba_manager_test, force_split)
+{
+  run_async([this] {
+    for (unsigned i = 0; i < 40; ++i) {
+      auto t = create_transaction();
+      logger().debug("opened transaction");
+      for (unsigned j = 0; j < 5; ++j) {
+	auto ret = alloc_mapping(t, 0, block_size, get_paddr());
+	if ((i % 10 == 0) && (j == 3)) {
+	  check_mappings(t);
+	  check_mappings();
+	}
+      }
+      logger().debug("submitting transaction");
+      submit_test_transaction(std::move(t));
+      check_mappings();
+    }
+  });
+}
+
+TEST_F(btree_lba_manager_test, force_split_merge)
+{
+  run_async([this] {
+    for (unsigned i = 0; i < 80; ++i) {
+      auto t = create_transaction();
+      logger().debug("opened transaction");
+      for (unsigned j = 0; j < 5; ++j) {
+	auto ret = alloc_mapping(t, 0, block_size, get_paddr());
+	// just to speed things up a bit
+	if ((i % 100 == 0) && (j == 3)) {
+	  check_mappings(t);
+	  check_mappings();
+	}
+	incref_mapping(t, ret->get_laddr());
+	decref_mapping(t, ret->get_laddr());
+      }
+      logger().debug("submitting transaction");
+      submit_test_transaction(std::move(t));
+      if (i % 50 == 0) {
+	check_mappings();
+      }
+    }
+    {
+      auto addresses = get_mapped_addresses();
+      auto t = create_transaction();
+      for (unsigned i = 0; i != addresses.size(); ++i) {
+	if (i % 2 == 0) {
+	  incref_mapping(t, addresses[i]);
+	  decref_mapping(t, addresses[i]);
+	  decref_mapping(t, addresses[i]);
+	}
+	logger().debug("submitting transaction");
+	if (i % 7 == 0) {
+	  submit_test_transaction(std::move(t));
+	  t = create_transaction();
+	}
+	if (i % 13 == 0) {
+	  check_mappings();
+	  check_mappings(t);
+	}
+      }
+      submit_test_transaction(std::move(t));
+    }
+    {
+      auto addresses = get_mapped_addresses();
+      auto t = create_transaction();
+      for (unsigned i = 0; i != addresses.size(); ++i) {
+	incref_mapping(t, addresses[i]);
+	decref_mapping(t, addresses[i]);
+	decref_mapping(t, addresses[i]);
+      }
+      check_mappings(t);
+      submit_test_transaction(std::move(t));
+      check_mappings();
+    }
+  });
+}
+
+TEST_F(btree_lba_manager_test, single_transaction_split_merge)
+{
+  run_async([this] {
+    {
+      auto t = create_transaction();
+      for (unsigned i = 0; i < 600; ++i) {
+	alloc_mapping(t, 0, block_size, get_paddr());
+      }
+      check_mappings(t);
+      submit_test_transaction(std::move(t));
+    }
+    check_mappings();
+
+    {
+      auto addresses = get_mapped_addresses();
+      auto t = create_transaction();
+      for (unsigned i = 0; i != addresses.size(); ++i) {
+	if (i % 4 != 0) {
+	  decref_mapping(t, addresses[i]);
+	}
+      }
+      check_mappings(t);
+      submit_test_transaction(std::move(t));
+    }
+    check_mappings();
+
+    {
+      auto t = create_transaction();
+      for (unsigned i = 0; i < 600; ++i) {
+	alloc_mapping(t, 0, block_size, get_paddr());
+      }
+      auto addresses = get_mapped_addresses(t);
+      for (unsigned i = 0; i != addresses.size(); ++i) {
+	decref_mapping(t, addresses[i]);
+      }
+      check_mappings(t);
+      submit_test_transaction(std::move(t));
+    }
+    check_mappings();
+  });
+}
diff --git a/src/test/crimson/seastore/test_extmap_manager.cc b/src/test/crimson/seastore/test_extmap_manager.cc
new file mode 100644
index 000000000..8b2588011
--- /dev/null
+++ b/src/test/crimson/seastore/test_extmap_manager.cc
@@ -0,0 +1,283 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/crimson/gtest_seastar.h"
+#include "test/crimson/seastore/transaction_manager_test_state.h"
+
+#include "crimson/os/seastore/cache.h"
+#include "crimson/os/seastore/transaction_manager.h"
+#include "crimson/os/seastore/segment_manager.h"
+#include "crimson/os/seastore/extentmap_manager.h"
+
+#include "test/crimson/seastore/test_block.h"
+
+using namespace crimson;
+using namespace crimson::os;
+using namespace crimson::os::seastore;
+
+namespace {
+  [[maybe_unused]] seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_test);
+  }
+}
+
+
+struct extentmap_manager_test_t :
+  public seastar_test_suite_t,
+  TMTestState {
+
+  ExtentMapManagerRef extmap_manager;
+
+  extentmap_manager_test_t() {}
+
+  seastar::future<> set_up_fut() final {
+    return tm_setup().then([this] {
+      extmap_manager = extentmap_manager::create_extentmap_manager(*tm);
+      return seastar::now();
+    });
+  }
+
+  seastar::future<> tear_down_fut() final {
+    return tm_teardown().then([this] {
+      extmap_manager.reset();
+      return seastar::now();
+    });
+  }
+
+  using test_extmap_t = std::map<uint32_t, lext_map_val_t>;
+  test_extmap_t test_ext_mappings;
+
+  extent_mapping_t insert_extent(
+    extmap_root_t &extmap_root,
+    Transaction &t,
+    uint32_t lo,
+    lext_map_val_t val) {
+    auto extent = extmap_manager->add_lextent(extmap_root, t, lo, val).unsafe_get0();
+    EXPECT_EQ(lo, extent.logical_offset);
+    EXPECT_EQ(val.laddr, extent.laddr);
+    EXPECT_EQ(val.length, extent.length);
+    test_ext_mappings.emplace(extent.logical_offset,
+			      lext_map_val_t{extent.laddr, extent.length});
+    return extent;
+  }
+
+  extent_map_list_t find_extent(
+    extmap_root_t &extmap_root,
+    Transaction &t,
+    uint32_t lo,
+    uint32_t len) {
+    auto extent = extmap_manager->find_lextent(extmap_root, t, lo, len).unsafe_get0();
+    EXPECT_EQ(lo, extent.front().logical_offset);
+    EXPECT_EQ(len, extent.front().length);
+    return extent;
+  }
+
+  extent_map_list_t findno_extent(
+    extmap_root_t &extmap_root,
+    Transaction &t,
+    uint32_t lo,
+    uint32_t len) {
+    auto extent = extmap_manager->find_lextent(extmap_root, t, lo, len).unsafe_get0();
+    EXPECT_EQ(extent.empty(), true);
+    return extent;
+  }
+
+  void rm_extent(
+    extmap_root_t &extmap_root,
+    Transaction &t,
+    uint32_t lo,
+    lext_map_val_t val ) {
+    auto ret = extmap_manager->rm_lextent(extmap_root, t, lo, val).unsafe_get0();
+    EXPECT_TRUE(ret);
+    test_ext_mappings.erase(lo);
+  }
+
+  void check_mappings(extmap_root_t &extmap_root, Transaction &t) {
+    for (const auto& [lo, ext]: test_ext_mappings){
+      const auto ext_list = find_extent(extmap_root, t, lo, ext.length);
+      ASSERT_EQ(ext_list.size(), 1);
+      const auto& ext_map = ext_list.front();
+      EXPECT_EQ(ext.laddr, ext_map.laddr);
+      EXPECT_EQ(ext.length, ext_map.length);
+    }
+  }
+
+  void check_mappings(extmap_root_t &extmap_root) {
+    auto t = tm->create_transaction();
+    check_mappings(extmap_root, *t);
+  }
+
+  void replay() {
+    logger().debug("{}: begin", __func__);
+    restart();
+    extmap_manager = extentmap_manager::create_extentmap_manager(*tm);
+    logger().debug("{}: end", __func__);
+  }
+
+
+};
+
+TEST_F(extentmap_manager_test_t, basic)
+{
+  run_async([this] {
+    extmap_root_t extmap_root(0, L_ADDR_NULL);
+    {
+      auto t = tm->create_transaction();
+      extmap_root = extmap_manager->initialize_extmap(*t).unsafe_get0();
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+
+    uint32_t len = 4096;
+    uint32_t lo = 0x1 * len;
+    {
+      auto t = tm->create_transaction();
+      logger().debug("first transaction");
+      [[maybe_unused]] auto addref = insert_extent(extmap_root, *t, lo, {lo, len});
+      [[maybe_unused]] auto seekref = find_extent(extmap_root, *t, lo, len);
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+    {
+      auto t = tm->create_transaction();
+      logger().debug("second transaction");
+      auto seekref = find_extent(extmap_root, *t, lo, len);
+      rm_extent(extmap_root, *t, lo, {seekref.front().laddr, len});
+      [[maybe_unused]] auto seekref2 = findno_extent(extmap_root, *t, lo, len);
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+    {
+      auto t = tm->create_transaction();
+      logger().debug("third transaction");
+      [[maybe_unused]] auto seekref = findno_extent(extmap_root, *t, lo, len);
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+  });
+}
+
+TEST_F(extentmap_manager_test_t, force_leafnode_split)
+{
+  run_async([this] {
+    extmap_root_t extmap_root(0, L_ADDR_NULL);
+    {
+      auto t = tm->create_transaction();
+      extmap_root = extmap_manager->initialize_extmap(*t).unsafe_get0();
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+    uint32_t len = 4096;
+    uint32_t lo = 0;
+    for (unsigned i = 0; i < 40; i++) {
+      auto t = tm->create_transaction();
+      logger().debug("opened transaction");
+      for (unsigned j = 0; j < 10; ++j) {
+        [[maybe_unused]] auto addref = insert_extent(extmap_root, *t, lo, {lo, len});
+        lo += len;
+        if ((i % 20 == 0) && (j == 5)) {
+          check_mappings(extmap_root, *t);
+        }
+      }
+      logger().debug("force split submit transaction i = {}", i);
+      tm->submit_transaction(std::move(t)).unsafe_get();
+      check_mappings(extmap_root);
+    }
+  });
+
+}
+
+TEST_F(extentmap_manager_test_t, force_leafnode_split_merge)
+{
+  run_async([this] {
+    extmap_root_t extmap_root(0, L_ADDR_NULL);
+    {
+      auto t = tm->create_transaction();
+      extmap_root = extmap_manager->initialize_extmap(*t).unsafe_get0();
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+    uint32_t len = 4096;
+    uint32_t lo = 0;
+    for (unsigned i = 0; i < 80; i++) {
+      auto t = tm->create_transaction();
+      logger().debug("opened split_merge transaction");
+      for (unsigned j = 0; j < 5; ++j) {
+        [[maybe_unused]] auto addref = insert_extent(extmap_root, *t, lo, {lo, len});
+        lo += len;
+        if ((i % 10 == 0) && (j == 3)) {
+          check_mappings(extmap_root, *t);
+        }
+      }
+      logger().debug("submitting transaction");
+      tm->submit_transaction(std::move(t)).unsafe_get();
+      if (i % 50 == 0) {
+        check_mappings(extmap_root);
+      }
+    }
+    auto t = tm->create_transaction();
+    int i = 0;
+    for (auto iter = test_ext_mappings.begin(); iter != test_ext_mappings.end();) {
+      auto [lo, ext] = *iter;
+      ++iter;
+      if (i % 3 != 0) {
+        rm_extent(extmap_root, *t, lo, ext);
+      }
+      i++;
+
+      if (i % 10 == 0) {
+        logger().debug("submitting transaction i= {}", i);
+        tm->submit_transaction(std::move(t)).unsafe_get();
+        t = tm->create_transaction();
+      }
+      if (i % 100 == 0) {
+        logger().debug("check_mappings  i= {}", i);
+        check_mappings(extmap_root, *t);
+        check_mappings(extmap_root);
+      }
+    }
+    logger().debug("finally submitting transaction ");
+    tm->submit_transaction(std::move(t)).unsafe_get();
+  });
+}
+
+TEST_F(extentmap_manager_test_t, force_leafnode_split_merge_replay)
+{
+  run_async([this] {
+    extmap_root_t extmap_root(0, L_ADDR_NULL);
+    {
+      auto t = tm->create_transaction();
+      extmap_root = extmap_manager->initialize_extmap(*t).unsafe_get0();
+      tm->submit_transaction(std::move(t)).unsafe_get();
+      replay();
+    }
+    uint32_t len = 4096;
+    uint32_t lo = 0;
+    for (unsigned i = 0; i < 50; i++) {
+      auto t = tm->create_transaction();
+      logger().debug("opened split_merge transaction");
+      for (unsigned j = 0; j < 5; ++j) {
+        [[maybe_unused]] auto addref = insert_extent(extmap_root, *t, lo, {lo, len});
+        lo += len;
+      }
+      logger().debug("submitting transaction");
+      tm->submit_transaction(std::move(t)).unsafe_get();
+    }
+    replay();
+    auto t = tm->create_transaction();
+    int i = 0;
+    for (auto iter = test_ext_mappings.begin(); iter != test_ext_mappings.end();) {
+      auto [lo, ext] = *iter;
+      ++iter;
+      rm_extent(extmap_root, *t, lo, ext);
+      i++;
+
+      if (i % 10 == 0) {
+        logger().debug("submitting transaction i= {}", i);
+        tm->submit_transaction(std::move(t)).unsafe_get();
+        t = tm->create_transaction();
+      }
+      if (i% 100 == 0){
+        check_mappings(extmap_root);
+      }
+    }
+    logger().debug("finally submitting transaction ");
+    tm->submit_transaction(std::move(t)).unsafe_get();
+    replay();
+    check_mappings(extmap_root);
+  });
+}
diff --git a/src/test/crimson/seastore/test_seastore_cache.cc b/src/test/crimson/seastore/test_seastore_cache.cc
new file mode 100644
index 000000000..913668b08
--- /dev/null
+++ b/src/test/crimson/seastore/test_seastore_cache.cc
@@ -0,0 +1,235 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/crimson/gtest_seastar.h"
+
+#include "crimson/common/log.h"
+#include "crimson/os/seastore/cache.h"
+#include "crimson/os/seastore/segment_manager/ephemeral.h"
+
+#include "test/crimson/seastore/test_block.h"
+
+using namespace crimson;
+using namespace crimson::os;
+using namespace crimson::os::seastore;
+
+namespace {
+  [[maybe_unused]] seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_test);
+  }
+}
+
+struct cache_test_t : public seastar_test_suite_t {
+  segment_manager::EphemeralSegmentManagerRef segment_manager;
+  Cache cache;
+  paddr_t current{0, 0};
+  journal_seq_t seq;
+
+  cache_test_t()
+    : segment_manager(segment_manager::create_test_ephemeral()),
+      cache(*segment_manager) {}
+
+  seastar::future<std::optional<paddr_t>> submit_transaction(
+    TransactionRef t) {
+    auto record = cache.try_construct_record(*t);
+    if (!record) {
+      return seastar::make_ready_future<std::optional<paddr_t>>(
+	std::nullopt);
+    }
+
+    bufferlist bl;
+    for (auto &&block : record->extents) {
+      bl.append(block.bl);
+    }
+
+    ceph_assert((segment_off_t)bl.length() <
+		segment_manager->get_segment_size());
+    if (current.offset + (segment_off_t)bl.length() >
+	segment_manager->get_segment_size())
+      current = paddr_t{current.segment + 1, 0};
+
+    auto prev = current;
+    current.offset += bl.length();
+    return segment_manager->segment_write(
+      prev,
+      std::move(bl),
+      true
+    ).safe_then(
+      [this, prev, t=std::move(t)]() mutable {
+	cache.complete_commit(*t, prev, seq /* TODO */);
+	return seastar::make_ready_future<std::optional<paddr_t>>(prev);
+      },
+      crimson::ct_error::all_same_way([](auto e) {
+	ASSERT_FALSE("failed to submit");
+      })
+     );
+  }
+
+  auto get_transaction() {
+    return make_transaction();
+  }
+
+  seastar::future<> set_up_fut() final {
+    return segment_manager->init(
+    ).safe_then(
+      [this] {
+	return seastar::do_with(
+	  make_transaction(),
+	  [this](auto &transaction) {
+	    cache.init();
+	    return cache.mkfs(*transaction).safe_then(
+	      [this, &transaction] {
+		return submit_transaction(std::move(transaction)).then(
+		  [](auto p) {
+		    ASSERT_TRUE(p);
+		  });
+	      });
+	  });
+      }).handle_error(
+	crimson::ct_error::all_same_way([](auto e) {
+	  ASSERT_FALSE("failed to submit");
+	})
+      );
+  }
+
+  seastar::future<> tear_down_fut() final {
+    return cache.close().handle_error(
+      Cache::close_ertr::assert_all{});
+  }
+};
+
+TEST_F(cache_test_t, test_addr_fixup)
+{
+  run_async([this] {
+    paddr_t addr;
+    int csum = 0;
+    {
+      auto t = get_transaction();
+      auto extent = cache.alloc_new_extent<TestBlockPhysical>(
+	*t,
+	TestBlockPhysical::SIZE);
+      extent->set_contents('c');
+      csum = extent->get_crc32c();
+      auto ret = submit_transaction(std::move(t)).get0();
+      ASSERT_TRUE(ret);
+      addr = extent->get_paddr();
+    }
+    {
+      auto t = get_transaction();
+      auto extent = cache.get_extent<TestBlockPhysical>(
+	*t,
+	addr,
+	TestBlockPhysical::SIZE).unsafe_get0();
+      ASSERT_EQ(extent->get_paddr(), addr);
+      ASSERT_EQ(extent->get_crc32c(), csum);
+    }
+  });
+}
+
+TEST_F(cache_test_t, test_dirty_extent)
+{
+  run_async([this] {
+    paddr_t addr;
+    int csum = 0;
+    int csum2 = 0;
+    {
+      // write out initial test block
+      auto t = get_transaction();
+      auto extent = cache.alloc_new_extent<TestBlockPhysical>(
+	*t,
+	TestBlockPhysical::SIZE);
+      extent->set_contents('c');
+      csum = extent->get_crc32c();
+      auto reladdr = extent->get_paddr();
+      ASSERT_TRUE(reladdr.is_relative());
+      {
+	// test that read with same transaction sees new block though
+	// uncommitted
+	auto extent = cache.get_extent<TestBlockPhysical>(
+	  *t,
+	  reladdr,
+	  TestBlockPhysical::SIZE).unsafe_get0();
+	ASSERT_TRUE(extent->is_clean());
+	ASSERT_TRUE(extent->is_pending());
+	ASSERT_TRUE(extent->get_paddr().is_relative());
+	ASSERT_EQ(extent->get_version(), 0);
+	ASSERT_EQ(csum, extent->get_crc32c());
+      }
+      auto ret = submit_transaction(std::move(t)).get0();
+      ASSERT_TRUE(ret);
+      addr = extent->get_paddr();
+    }
+    {
+      // test that consecutive reads on the same extent get the same ref
+      auto t = get_transaction();
+      auto extent = cache.get_extent<TestBlockPhysical>(
+	*t,
+	addr,
+	TestBlockPhysical::SIZE).unsafe_get0();
+      auto t2 = get_transaction();
+      auto extent2 = cache.get_extent<TestBlockPhysical>(
+	*t2,
+	addr,
+	TestBlockPhysical::SIZE).unsafe_get0();
+      ASSERT_EQ(&*extent, &*extent2);
+    }
+    {
+      // read back test block
+      auto t = get_transaction();
+      auto extent = cache.get_extent<TestBlockPhysical>(
+	*t,
+	addr,
+	TestBlockPhysical::SIZE).unsafe_get0();
+      // duplicate and reset contents
+      extent = cache.duplicate_for_write(*t, extent)->cast<TestBlockPhysical>();
+      extent->set_contents('c');
+      csum2 = extent->get_crc32c();
+      ASSERT_EQ(extent->get_paddr(), addr);
+      {
+	// test that concurrent read with fresh transaction sees old
+        // block
+	auto t2 = get_transaction();
+	auto extent = cache.get_extent<TestBlockPhysical>(
+	  *t2,
+	  addr,
+	  TestBlockPhysical::SIZE).unsafe_get0();
+	ASSERT_TRUE(extent->is_clean());
+	ASSERT_FALSE(extent->is_pending());
+	ASSERT_EQ(addr, extent->get_paddr());
+	ASSERT_EQ(extent->get_version(), 0);
+	ASSERT_EQ(csum, extent->get_crc32c());
+      }
+      {
+	// test that read with same transaction sees new block
+	auto extent = cache.get_extent<TestBlockPhysical>(
+	  *t,
+	  addr,
+	  TestBlockPhysical::SIZE).unsafe_get0();
+	ASSERT_TRUE(extent->is_dirty());
+	ASSERT_TRUE(extent->is_pending());
+	ASSERT_EQ(addr, extent->get_paddr());
+	ASSERT_EQ(extent->get_version(), 1);
+	ASSERT_EQ(csum2, extent->get_crc32c());
+      }
+      // submit transaction
+      auto ret = submit_transaction(std::move(t)).get0();
+      ASSERT_TRUE(ret);
+      ASSERT_TRUE(extent->is_dirty());
+      ASSERT_EQ(addr, extent->get_paddr());
+      ASSERT_EQ(extent->get_version(), 1);
+      ASSERT_EQ(extent->get_crc32c(), csum2);
+    }
+    {
+      // test that fresh transaction now sees newly dirty block
+      auto t = get_transaction();
+      auto extent = cache.get_extent<TestBlockPhysical>(
+	*t,
+	addr,
+	TestBlockPhysical::SIZE).unsafe_get0();
+      ASSERT_TRUE(extent->is_dirty());
+      ASSERT_EQ(addr, extent->get_paddr());
+      ASSERT_EQ(extent->get_version(), 1);
+      ASSERT_EQ(csum2, extent->get_crc32c());
+    }
+  });
+}
diff --git a/src/test/crimson/seastore/test_seastore_journal.cc b/src/test/crimson/seastore/test_seastore_journal.cc
new file mode 100644
index 000000000..0bed505ff
--- /dev/null
+++ b/src/test/crimson/seastore/test_seastore_journal.cc
@@ -0,0 +1,260 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/crimson/gtest_seastar.h"
+
+#include <random>
+
+#include "crimson/common/log.h"
+#include "crimson/os/seastore/journal.h"
+#include "crimson/os/seastore/segment_manager/ephemeral.h"
+
+using namespace crimson;
+using namespace crimson::os;
+using namespace crimson::os::seastore;
+
+namespace {
+  [[maybe_unused]] seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_test);
+  }
+}
+
+struct record_validator_t {
+  record_t record;
+  paddr_t record_final_offset;
+
+  template <typename... T>
+  record_validator_t(T&&... record) : record(std::forward<T>(record)...) {}
+
+  void validate(SegmentManager &manager) {
+    paddr_t addr = make_record_relative_paddr(0);
+    for (auto &&block : record.extents) {
+      auto test = manager.read(
+	record_final_offset.add_relative(addr),
+	block.bl.length()).unsafe_get0();
+      addr.offset += block.bl.length();
+      bufferlist bl;
+      bl.push_back(test);
+      ASSERT_EQ(
+	bl.length(),
+	block.bl.length());
+      ASSERT_EQ(
+	bl.begin().crc32c(bl.length(), 1),
+	block.bl.begin().crc32c(block.bl.length(), 1));
+    }
+  }
+
+  auto get_replay_handler() {
+    auto checker = [this, iter=record.deltas.begin()] (
+      paddr_t base,
+      const delta_info_t &di) mutable {
+      EXPECT_EQ(base, record_final_offset);
+      ceph_assert(iter != record.deltas.end());
+      EXPECT_EQ(di, *iter++);
+      EXPECT_EQ(base, record_final_offset);
+      return iter != record.deltas.end();
+    };
+    if (record.deltas.size()) {
+      return std::make_optional(std::move(checker));
+    } else {
+      return std::optional<decltype(checker)>();
+    }
+  }
+};
+
+struct journal_test_t : seastar_test_suite_t, JournalSegmentProvider {
+  segment_manager::EphemeralSegmentManagerRef segment_manager;
+  std::unique_ptr<Journal> journal;
+
+  std::vector<record_validator_t> records;
+
+  std::default_random_engine generator;
+
+  const segment_off_t block_size;
+
+  journal_test_t()
+    : segment_manager(segment_manager::create_test_ephemeral()),
+      block_size(segment_manager->get_block_size())
+  {
+  }
+
+  segment_id_t next = 0;
+  get_segment_ret get_segment() final {
+    return get_segment_ret(
+      get_segment_ertr::ready_future_marker{},
+      next++);
+  }
+
+  journal_seq_t get_journal_tail_target() const final { return journal_seq_t{}; }
+  void update_journal_tail_committed(journal_seq_t paddr) final {}
+
+  seastar::future<> set_up_fut() final {
+    journal.reset(new Journal(*segment_manager));
+    journal->set_segment_provider(this);
+    return segment_manager->init(
+    ).safe_then([this] {
+      return journal->open_for_write();
+    }).safe_then(
+      [](auto){},
+      crimson::ct_error::all_same_way([] {
+	ASSERT_FALSE("Unable to mount");
+      }));
+  }
+
+  template <typename T>
+  auto replay(T &&f) {
+    return journal->close(
+    ).safe_then([this, f=std::move(f)]() mutable {
+      journal.reset(new Journal(*segment_manager));
+      journal->set_segment_provider(this);
+      return journal->replay(std::forward<T>(std::move(f)));
+    }).safe_then([this] {
+      return journal->open_for_write();
+    });
+  }
+
+  auto replay_and_check() {
+    auto record_iter = records.begin();
+    decltype(record_iter->get_replay_handler()) delta_checker = std::nullopt;
+    auto advance = [this, &record_iter, &delta_checker] {
+      ceph_assert(!delta_checker);
+      while (record_iter != records.end()) {
+	auto checker = record_iter->get_replay_handler();
+	record_iter++;
+	if (checker) {
+	  delta_checker.emplace(std::move(*checker));
+	  break;
+	}
+      }
+    };
+    advance();
+    replay(
+      [&advance,
+       &delta_checker]
+      (auto seq, auto base, const auto &di) mutable {
+	if (!delta_checker) {
+	  EXPECT_FALSE("No Deltas Left");
+	}
+	if (!(*delta_checker)(base, di)) {
+	  delta_checker = std::nullopt;
+	  advance();
+	}
+	return Journal::replay_ertr::now();
+      }).unsafe_get0();
+    ASSERT_EQ(record_iter, records.end());
+    for (auto &i : records) {
+      i.validate(*segment_manager);
+    }
+  }
+
+  template <typename... T>
+  auto submit_record(T&&... _record) {
+    auto record{std::forward<T>(_record)...};
+    records.push_back(record);
+    auto [addr, _] = journal->submit_record(std::move(record)).unsafe_get0();
+    records.back().record_final_offset = addr;
+    return addr;
+  }
+
+  seastar::future<> tear_down_fut() final {
+    return seastar::now();
+  }
+
+  extent_t generate_extent(size_t blocks) {
+    std::uniform_int_distribution<char> distribution(
+      std::numeric_limits<char>::min(),
+      std::numeric_limits<char>::max()
+    );
+    char contents = distribution(generator);
+    bufferlist bl;
+    bl.append(buffer::ptr(buffer::create(blocks * block_size, contents)));
+    return extent_t{extent_types_t::TEST_BLOCK, L_ADDR_NULL, bl};
+  }
+
+  delta_info_t generate_delta(size_t bytes) {
+    std::uniform_int_distribution<char> distribution(
+      std::numeric_limits<char>::min(),
+      std::numeric_limits<char>::max()
+    );
+    char contents = distribution(generator);
+    bufferlist bl;
+    bl.append(buffer::ptr(buffer::create(bytes, contents)));
+    return delta_info_t{
+      extent_types_t::TEST_BLOCK,
+      paddr_t{},
+      L_ADDR_NULL,
+      0, 0,
+      block_size,
+      1,
+      bl
+    };
+  }
+};
+
+TEST_F(journal_test_t, replay_one_journal_segment)
+{
+ run_async([this] {
+   submit_record(record_t{
+     { generate_extent(1), generate_extent(2) },
+     { generate_delta(23), generate_delta(30) }
+     });
+   replay_and_check();
+ });
+}
+
+TEST_F(journal_test_t, replay_two_records)
+{
+ run_async([this] {
+   submit_record(record_t{
+     { generate_extent(1), generate_extent(2) },
+     { generate_delta(23), generate_delta(30) }
+     });
+   submit_record(record_t{
+     { generate_extent(4), generate_extent(1) },
+     { generate_delta(23), generate_delta(400) }
+     });
+   replay_and_check();
+ });
+}
+
+TEST_F(journal_test_t, replay_twice)
+{
+ run_async([this] {
+   submit_record(record_t{
+     { generate_extent(1), generate_extent(2) },
+     { generate_delta(23), generate_delta(30) }
+     });
+   submit_record(record_t{
+     { generate_extent(4), generate_extent(1) },
+     { generate_delta(23), generate_delta(400) }
+     });
+   replay_and_check();
+   submit_record(record_t{
+     { generate_extent(2), generate_extent(5) },
+     { generate_delta(230), generate_delta(40) }
+     });
+   replay_and_check();
+ });
+}
+
+TEST_F(journal_test_t, roll_journal_and_replay)
+{
+ run_async([this] {
+   paddr_t current = submit_record(
+     record_t{
+       { generate_extent(1), generate_extent(2) },
+       { generate_delta(23), generate_delta(30) }
+     });
+   auto starting_segment = current.segment;
+   unsigned so_far = 0;
+   while (current.segment == starting_segment) {
+     current = submit_record(record_t{
+	 { generate_extent(512), generate_extent(512) },
+	 { generate_delta(23), generate_delta(400) }
+       });
+     ++so_far;
+     ASSERT_FALSE(so_far > 10);
+   }
+   replay_and_check();
+ });
+}
diff --git a/src/test/crimson/seastore/test_transaction_manager.cc b/src/test/crimson/seastore/test_transaction_manager.cc
new file mode 100644
index 000000000..9906f938a
--- /dev/null
+++ b/src/test/crimson/seastore/test_transaction_manager.cc
@@ -0,0 +1,495 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <random>
+
+#include "test/crimson/gtest_seastar.h"
+#include "test/crimson/seastore/transaction_manager_test_state.h"
+
+#include "crimson/os/seastore/segment_cleaner.h"
+#include "crimson/os/seastore/cache.h"
+#include "crimson/os/seastore/transaction_manager.h"
+#include "crimson/os/seastore/segment_manager/ephemeral.h"
+#include "crimson/os/seastore/segment_manager.h"
+
+#include "test/crimson/seastore/test_block.h"
+
+using namespace crimson;
+using namespace crimson::os;
+using namespace crimson::os::seastore;
+
+namespace {
+  [[maybe_unused]] seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_test);
+  }
+}
+
+struct test_extent_record_t {
+  test_extent_desc_t desc;
+  unsigned refcount = 0;
+  test_extent_record_t() = default;
+  test_extent_record_t(
+    const test_extent_desc_t &desc,
+    unsigned refcount) : desc(desc), refcount(refcount) {}
+
+  void update(const test_extent_desc_t &to) {
+    desc = to;
+  }
+
+  bool operator==(const test_extent_desc_t &rhs) const {
+    return desc == rhs;
+  }
+  bool operator!=(const test_extent_desc_t &rhs) const {
+    return desc != rhs;
+  }
+};
+
+std::ostream &operator<<(std::ostream &lhs, const test_extent_record_t &rhs) {
+  return lhs << "test_extent_record_t(" << rhs.desc
+	     << ", refcount=" << rhs.refcount << ")";
+}
+
+struct transaction_manager_test_t :
+  public seastar_test_suite_t,
+  TMTestState {
+
+  std::random_device rd;
+  std::mt19937 gen;
+
+  transaction_manager_test_t()
+    : gen(rd()) {
+    init();
+  }
+
+  laddr_t get_random_laddr(size_t block_size, laddr_t limit) {
+    return block_size *
+      std::uniform_int_distribution<>(0, (limit / block_size) - 1)(gen);
+  }
+
+  char get_random_contents() {
+    return static_cast<char>(std::uniform_int_distribution<>(0, 255)(gen));
+  }
+
+  seastar::future<> set_up_fut() final {
+    return tm_setup();
+  }
+
+  seastar::future<> tear_down_fut() final {
+    return tm_teardown();
+  }
+
+  struct test_extents_t : std::map<laddr_t, test_extent_record_t> {
+  private:
+    void check_available(laddr_t addr, extent_len_t len) {
+      auto iter = upper_bound(addr);
+      if (iter != begin()) {
+	auto liter = iter;
+	liter--;
+	EXPECT_FALSE(liter->first + liter->second.desc.len > addr);
+      }
+      if (iter != end()) {
+	EXPECT_FALSE(iter->first < addr + len);
+      }
+    }
+    void check_hint(laddr_t hint, laddr_t addr, extent_len_t len) {
+      auto iter = lower_bound(hint);
+      laddr_t last = hint;
+      while (true) {
+	if (iter == end() || iter->first > addr) {
+	  EXPECT_EQ(addr, last);
+	  break;
+	}
+	EXPECT_FALSE(iter->first - last > len);
+	last = iter->first + iter->second.desc.len;
+	++iter;
+      }
+    }
+  public:
+    void insert(TestBlock &extent) {
+      check_available(extent.get_laddr(), extent.get_length());
+      emplace(
+	std::make_pair(
+	  extent.get_laddr(),
+	  test_extent_record_t{extent.get_desc(), 1}
+	));
+    }
+    void alloced(laddr_t hint, TestBlock &extent) {
+      check_hint(hint, extent.get_laddr(), extent.get_length());
+      insert(extent);
+    }
+  } test_mappings;
+
+  struct test_transaction_t {
+    TransactionRef t;
+    test_extents_t mappings;
+  };
+
+  test_transaction_t create_transaction() {
+    return { tm->create_transaction(), test_mappings };
+  }
+
+  test_transaction_t create_weak_transaction() {
+    return { tm->create_weak_transaction(), test_mappings };
+  }
+
+  TestBlockRef alloc_extent(
+    test_transaction_t &t,
+    laddr_t hint,
+    extent_len_t len,
+    char contents) {
+    auto extent = tm->alloc_extent<TestBlock>(
+      *(t.t),
+      hint,
+      len).unsafe_get0();
+    extent->set_contents(contents);
+    EXPECT_FALSE(t.mappings.count(extent->get_laddr()));
+    EXPECT_EQ(len, extent->get_length());
+    t.mappings.alloced(hint, *extent);
+    return extent;
+  }
+
+  TestBlockRef alloc_extent(
+    test_transaction_t &t,
+    laddr_t hint,
+    extent_len_t len) {
+    return alloc_extent(
+      t,
+      hint,
+      len,
+      get_random_contents());
+  }
+
+  bool check_usage() {
+    auto t = create_weak_transaction();
+    SpaceTrackerIRef tracker(segment_cleaner->get_empty_space_tracker());
+    lba_manager->scan_mapped_space(
+      *t.t,
+      [&tracker](auto offset, auto len) {
+	tracker->allocate(
+	  offset.segment,
+	  offset.offset,
+	  len);
+      }).unsafe_get0();
+    return segment_cleaner->debug_check_space(*tracker);
+  }
+
+  void replay() {
+    logger().debug("{}: begin", __func__);
+    EXPECT_TRUE(check_usage());
+    restart();
+    logger().debug("{}: end", __func__);
+  }
+
+  void check() {
+    check_mappings();
+    check_usage();
+  }
+
+  void check_mappings() {
+    auto t = create_weak_transaction();
+    check_mappings(t);
+  }
+
+  TestBlockRef get_extent(
+    test_transaction_t &t,
+    laddr_t addr,
+    extent_len_t len) {
+    ceph_assert(t.mappings.count(addr));
+    ceph_assert(t.mappings[addr].desc.len == len);
+
+    auto ret_list = tm->read_extents<TestBlock>(
+      *t.t, addr, len
+    ).unsafe_get0();
+    EXPECT_EQ(ret_list.size(), 1);
+    auto &ext = ret_list.begin()->second;
+    auto &laddr = ret_list.begin()->first;
+    EXPECT_EQ(addr, laddr);
+    EXPECT_EQ(addr, ext->get_laddr());
+    return ext;
+  }
+
+  test_block_mutator_t mutator;
+  TestBlockRef mutate_extent(
+    test_transaction_t &t,
+    TestBlockRef ref) {
+    ceph_assert(t.mappings.count(ref->get_laddr()));
+    ceph_assert(t.mappings[ref->get_laddr()].desc.len == ref->get_length());
+    auto ext = tm->get_mutable_extent(*t.t, ref)->cast<TestBlock>();
+    EXPECT_EQ(ext->get_laddr(), ref->get_laddr());
+    EXPECT_EQ(ext->get_desc(), ref->get_desc());
+    mutator.mutate(*ext, gen);
+    t.mappings[ext->get_laddr()].update(ext->get_desc());
+    return ext;
+  }
+
+  void inc_ref(test_transaction_t &t, laddr_t offset) {
+    ceph_assert(t.mappings.count(offset));
+    ceph_assert(t.mappings[offset].refcount > 0);
+    auto refcnt = tm->inc_ref(*t.t, offset).unsafe_get0();
+    t.mappings[offset].refcount++;
+    EXPECT_EQ(refcnt, t.mappings[offset].refcount);
+  }
+
+  void dec_ref(test_transaction_t &t, laddr_t offset) {
+    ceph_assert(t.mappings.count(offset));
+    ceph_assert(t.mappings[offset].refcount > 0);
+    auto refcnt = tm->dec_ref(*t.t, offset).unsafe_get0();
+    t.mappings[offset].refcount--;
+    EXPECT_EQ(refcnt, t.mappings[offset].refcount);
+    if (t.mappings[offset].refcount == 0) {
+      t.mappings.erase(offset);
+    }
+  }
+
+  void check_mappings(test_transaction_t &t) {
+    for (auto &i: t.mappings) {
+      logger().debug("check_mappings: {}->{}", i.first, i.second);
+      auto ext = get_extent(t, i.first, i.second.desc.len);
+      EXPECT_EQ(i.second, ext->get_desc());
+    }
+    auto lt = create_weak_transaction();
+    lba_manager->scan_mappings(
+      *lt.t,
+      0,
+      L_ADDR_MAX,
+      [iter=lt.mappings.begin(), &lt](auto l, auto p, auto len) mutable {
+	EXPECT_NE(iter, lt.mappings.end());
+	EXPECT_EQ(l, iter->first);
+	++iter;
+      }).unsafe_get0();
+  }
+
+  void submit_transaction(test_transaction_t t) {
+    tm->submit_transaction(std::move(t.t)).unsafe_get();
+    test_mappings = t.mappings;
+  }
+};
+
+TEST_F(transaction_manager_test_t, basic)
+{
+  constexpr laddr_t SIZE = 4096;
+  run_async([this] {
+    constexpr laddr_t ADDR = 0xFF * SIZE;
+    {
+      auto t = create_transaction();
+      auto extent = alloc_extent(
+	t,
+	ADDR,
+	SIZE,
+	'a');
+      ASSERT_EQ(ADDR, extent->get_laddr());
+      check_mappings(t);
+      check();
+      submit_transaction(std::move(t));
+      check();
+    }
+  });
+}
+
+TEST_F(transaction_manager_test_t, mutate)
+{
+  constexpr laddr_t SIZE = 4096;
+  run_async([this] {
+    constexpr laddr_t ADDR = 0xFF * SIZE;
+    {
+      auto t = create_transaction();
+      auto extent = alloc_extent(
+	t,
+	ADDR,
+	SIZE,
+	'a');
+      ASSERT_EQ(ADDR, extent->get_laddr());
+      check_mappings(t);
+      check();
+      submit_transaction(std::move(t));
+      check();
+    }
+    ASSERT_TRUE(check_usage());
+    replay();
+    {
+      auto t = create_transaction();
+      auto ext = get_extent(
+	t,
+	ADDR,
+	SIZE);
+      auto mut = mutate_extent(t, ext);
+      check_mappings(t);
+      check();
+      submit_transaction(std::move(t));
+      check();
+    }
+    ASSERT_TRUE(check_usage());
+    replay();
+    check();
+  });
+}
+
+TEST_F(transaction_manager_test_t, create_remove_same_transaction)
+{
+  constexpr laddr_t SIZE = 4096;
+  run_async([this] {
+    constexpr laddr_t ADDR = 0xFF * SIZE;
+    {
+      auto t = create_transaction();
+      auto extent = alloc_extent(
+	t,
+	ADDR,
+	SIZE,
+	'a');
+      ASSERT_EQ(ADDR, extent->get_laddr());
+      check_mappings(t);
+      dec_ref(t, ADDR);
+      check_mappings(t);
+
+      extent = alloc_extent(
+	t,
+	ADDR,
+	SIZE,
+	'a');
+
+      submit_transaction(std::move(t));
+      check();
+    }
+    replay();
+    check();
+  });
+}
+
+TEST_F(transaction_manager_test_t, split_merge_read_same_transaction)
+{
+  constexpr laddr_t SIZE = 4096;
+  run_async([this] {
+    {
+      auto t = create_transaction();
+      for (unsigned i = 0; i < 300; ++i) {
+	auto extent = alloc_extent(
+	  t,
+	  laddr_t(i * SIZE),
+	  SIZE);
+      }
+      check_mappings(t);
+      submit_transaction(std::move(t));
+      check();
+    }
+    {
+      auto t = create_transaction();
+      for (unsigned i = 0; i < 240; ++i) {
+	dec_ref(
+	  t,
+	  laddr_t(i * SIZE));
+      }
+      check_mappings(t);
+      submit_transaction(std::move(t));
+      check();
+    }
+  });
+}
+
+
+TEST_F(transaction_manager_test_t, inc_dec_ref)
+{
+  constexpr laddr_t SIZE = 4096;
+  run_async([this] {
+    constexpr laddr_t ADDR = 0xFF * SIZE;
+    {
+      auto t = create_transaction();
+      auto extent = alloc_extent(
+	t,
+	ADDR,
+	SIZE,
+	'a');
+      ASSERT_EQ(ADDR, extent->get_laddr());
+      check_mappings(t);
+      check();
+      submit_transaction(std::move(t));
+      check();
+    }
+    replay();
+    {
+      auto t = create_transaction();
+      inc_ref(t, ADDR);
+      check_mappings(t);
+      check();
+      submit_transaction(std::move(t));
+      check();
+    }
+    {
+      auto t = create_transaction();
+      dec_ref(t, ADDR);
+      check_mappings(t);
+      check();
+      submit_transaction(std::move(t));
+      check();
+    }
+    replay();
+    {
+      auto t = create_transaction();
+      dec_ref(t, ADDR);
+      check_mappings(t);
+      check();
+      submit_transaction(std::move(t));
+      check();
+    }
+  });
+}
+
+TEST_F(transaction_manager_test_t, cause_lba_split)
+{
+  constexpr laddr_t SIZE = 4096;
+  run_async([this] {
+    for (unsigned i = 0; i < 200; ++i) {
+      auto t = create_transaction();
+      auto extent = alloc_extent(
+	t,
+	i * SIZE,
+	SIZE,
+	(char)(i & 0xFF));
+      ASSERT_EQ(i * SIZE, extent->get_laddr());
+      submit_transaction(std::move(t));
+    }
+    check();
+  });
+}
+
+TEST_F(transaction_manager_test_t, random_writes)
+{
+  constexpr size_t TOTAL = 4<<20;
+  constexpr size_t BSIZE = 4<<10;
+  constexpr size_t PADDING_SIZE = 256<<10;
+  constexpr size_t BLOCKS = TOTAL / BSIZE;
+  run_async([this] {
+    for (unsigned i = 0; i < BLOCKS; ++i) {
+      auto t = create_transaction();
+      auto extent = alloc_extent(
+	t,
+	i * BSIZE,
+	BSIZE);
+      ASSERT_EQ(i * BSIZE, extent->get_laddr());
+      submit_transaction(std::move(t));
+    }
+
+    for (unsigned i = 0; i < 4; ++i) {
+      for (unsigned j = 0; j < 65; ++j) {
+	auto t = create_transaction();
+	for (unsigned k = 0; k < 2; ++k) {
+	  auto ext = get_extent(
+	    t,
+	    get_random_laddr(BSIZE, TOTAL),
+	    BSIZE);
+	  auto mut = mutate_extent(t, ext);
+	  // pad out transaction
+	  auto padding = alloc_extent(
+	    t,
+	    TOTAL + (k * PADDING_SIZE),
+	    PADDING_SIZE);
+	  dec_ref(t, padding->get_laddr());
+	}
+	submit_transaction(std::move(t));
+      }
+      replay();
+      logger().debug("random_writes: checking");
+      check();
+      logger().debug("random_writes: done replaying/checking");
+    }
+  });
+}
diff --git a/src/test/crimson/seastore/transaction_manager_test_state.h b/src/test/crimson/seastore/transaction_manager_test_state.h
new file mode 100644
index 000000000..bf2440923
--- /dev/null
+++ b/src/test/crimson/seastore/transaction_manager_test_state.h
@@ -0,0 +1,82 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include <random>
+
+#include "crimson/os/seastore/segment_cleaner.h"
+#include "crimson/os/seastore/cache.h"
+#include "crimson/os/seastore/transaction_manager.h"
+#include "crimson/os/seastore/segment_manager/ephemeral.h"
+#include "crimson/os/seastore/segment_manager.h"
+
+using namespace crimson;
+using namespace crimson::os;
+using namespace crimson::os::seastore;
+
+class TMTestState {
+protected:
+  std::unique_ptr<segment_manager::EphemeralSegmentManager> segment_manager;
+  std::unique_ptr<SegmentCleaner> segment_cleaner;
+  std::unique_ptr<Journal> journal;
+  std::unique_ptr<Cache> cache;
+  LBAManagerRef lba_manager;
+  std::unique_ptr<TransactionManager> tm;
+
+  TMTestState()
+    : segment_manager(segment_manager::create_test_ephemeral()) {
+    init();
+  }
+
+  void init() {
+    segment_cleaner = std::make_unique<SegmentCleaner>(
+      SegmentCleaner::config_t::default_from_segment_manager(
+	*segment_manager),
+      true);
+    journal = std::make_unique<Journal>(*segment_manager);
+    cache = std::make_unique<Cache>(*segment_manager);
+    lba_manager = lba_manager::create_lba_manager(*segment_manager, *cache);
+    tm = std::make_unique<TransactionManager>(
+      *segment_manager, *segment_cleaner, *journal, *cache, *lba_manager);
+
+    journal->set_segment_provider(&*segment_cleaner);
+    segment_cleaner->set_extent_callback(&*tm);
+  }
+
+  void destroy() {
+    tm.reset();
+    lba_manager.reset();
+    cache.reset();
+    journal.reset();
+    segment_cleaner.reset();
+  }
+
+  void restart() {
+    tm->close().unsafe_get();
+    destroy();
+    static_cast<segment_manager::EphemeralSegmentManager*>(&*segment_manager)->remount();
+    init();
+    tm->mount().unsafe_get();
+  }
+
+  seastar::future<> tm_setup() {
+    return segment_manager->init(
+    ).safe_then([this] {
+      return tm->mkfs();
+    }).safe_then([this] {
+      return tm->close();
+    }).safe_then([this] {
+      destroy();
+      static_cast<segment_manager::EphemeralSegmentManager*>(
+	&*segment_manager)->remount();
+      init();
+      return tm->mount();
+    }).handle_error(crimson::ct_error::assert_all{});
+  }
+
+  seastar::future<> tm_teardown() {
+    return tm->close(
+    ).handle_error(crimson::ct_error::assert_all{});
+  }
+};