Load & integration testing for Raft

.gitignore

@@ -30,6 +30,8 @@
 *.exe
 *.out
 *.app
+./tests/go/kvtest/cmd/kvtest/kvtest
+./tests/go/kvtest/kvtest
 
 # Project-specific
 *.bak
@@ -46,3 +48,6 @@ build
 var
 tmp
 current
+
+# Compiled Protobufs
+*.pb.*

Dockerfile

@@ -8,6 +8,7 @@ ARG BUILD_TYPE=release
 RUN apt-get update && \
     apt-get -y install \
     cmake \
+    ninja-build \
     python3 \
     python3-pip \
     python3-virtualenv && \

conan/profiles/debug-clang

@@ -1,6 +1,6 @@
 [buildenv]
-CC=/usr/bin/clang-20
-CXX=/usr/bin/clang++-20
+CC=/usr/bin/clang
+CXX=/usr/bin/clang++
 
 [settings]
 arch=x86_64

infra/assets/tkvpp_config_standalone.json

@@ -3,7 +3,7 @@
         "loggingLevel": "debug"
     },
     "database": {
-        "path": "./var/tkvpp",
+        "path": "/var/tkvpp",
         "walFilename": "wal",
         "manifestFilenamePrefix": "manifest_",
         "mode": "standalone"
@@ -14,24 +14,24 @@
         "storageType": "persistent"
     },
     "lsm": {
-        "flushThreshold": 1024,
+        "flushThreshold": 64000000,
         "maximumLevels": 12,
         "levelZeroCompaction": {
             "compactionStrategy": "levelled",
-            "compactionThreshold": 1024
+            "compactionThreshold": 256000000
         },
         "levelNonZeroCompaction": {
             "compactionStrategy": "tiered",
-            "compactionThreshold": 1024
+            "compactionThreshold": 256000000
         }
     },
     "server": {
         "transport": "grpc",
         "host": "0.0.0.0",
-        "port": 8081,
+        "port": 9891,
         "id": 1,
         "peers": [
-            "0.0.0.0:8081"
+            "0.0.0.0:9891"
         ]
     }
 }

infra/docker-compose.standalone.yaml

@@ -1,7 +1,15 @@
 ---
-version: '3.8'
 services:
-  app:
+  tkvpp-node-standalone:
     image: tinykvpp-clang:latest
-    container_name: tkvpp-node-1
+    security_opt:
+      - seccomp:unconfined
+    container_name: tkvpp-node-standalone
     command: ["/app/tkvpp", "-c", "/app/config.json"]
+    ports:
+      - "9891:9891"
+    volumes:
+      - ./assets/tkvpp_config_standalone.json:/app/config.json
+      - tkvpp-node-standalone:/var/tkvpp
+volumes:
+  tkvpp-node-standalone:

lib/db/db.cpp

@@ -40,7 +40,9 @@ auto serializeOperation(const db::client_request_t &request) -> std::string
     }
     default:
     {
-        spdlog::critical("Unknown operation type: {}", magic_enum::enum_name(request.type));
+        spdlog::critical(
+            "(serializeOperation): Unknown operation type: {}", magic_enum::enum_name(request.type)
+        );
         break;
     }
     }
@@ -398,7 +400,9 @@ auto db_t::onRaftCommit(const raft::v1::LogEntry &entry) -> bool
     }
     default:
     {
-        spdlog::critical("Unknown operation type: {}", magic_enum::enum_name(op.type()));
+        spdlog::critical(
+            "(onRaftCommit): Unknown operation type: {}", magic_enum::enum_name(op.type())
+        );
         return false;
     }
     }

lib/raft/raft.cpp

@@ -27,7 +27,7 @@ const std::string_view gRaftFilename = "RAFT_PERSISTENCE";
 
 auto constructFilename(std::string_view filename, std::uint32_t peerId) -> std::string
 {
-    return fmt::format("./var/tkvpp/{}_NODE_{}", filename, peerId);
+    return fmt::format("/var/tkvpp/{}_NODE_{}", filename, peerId);
 }
 
 auto generate_random_timeout(const int minTimeout, const int maxTimeout) -> int

lib/structures/lsmtree/levels/level.cpp

@@ -119,11 +119,17 @@ auto level_t::compact() const noexcept -> segments::regular_segment::shared_ptr_
     const auto            bytesUsedForLevel{bytes_used()};
 
     // If level size hasn't reached the size limit then skip the compaction
-    if ((index() == 0 &&
-         bytesUsedForLevel < m_pConfig->LSMTreeConfig.LevelZeroCompactionThreshold) ||
-        (index() != 0 &&
-         bytesUsedForLevel <
-             m_pConfig->LSMTreeConfig.LevelNonZeroCompactionThreshold * std::pow(10, index())))
+    const bool isZeroLevel{index() == 0};
+    const auto zeroLevelThreshold{m_pConfig->LSMTreeConfig.LevelZeroCompactionThreshold};
+    if (isZeroLevel && bytesUsedForLevel < zeroLevelThreshold)
+    {
+        return nullptr;
+    }
+
+    const auto nonZeroLevelThreshold{
+        m_pConfig->LSMTreeConfig.LevelNonZeroCompactionThreshold * std::pow(10, index())
+    };
+    if (!isZeroLevel && bytesUsedForLevel < nonZeroLevelThreshold)
     {
         return nullptr;
     }
@@ -135,13 +141,13 @@ auto level_t::compact() const noexcept -> segments::regular_segment::shared_ptr_
         std::vector<std::pair<
             typename memtable::memtable_t::const_iterator,
             typename memtable::memtable_t::const_iterator>>,
-        IteratorCompare<memtable_t, memtable_t>>
+        IteratorCompare<memtable_t>>
         minHeap;
 
     for (const auto &segment : m_storage)
     {
-        // TODO(lnikon): reset memtable inside regular_segment_t at the of the
-        // flush() and recover it here e.g. segment->recover_memtable();
+        // TODO(lnikon): reset memtable inside regular_segment_t at the of the flush() and recover
+        // it here e.g. segment->recover_memtable();
         const auto &currentMemtable = segment->memtable().value();
         minHeap.emplace(currentMemtable.begin(), currentMemtable.end());
     }
@@ -154,8 +160,7 @@ auto level_t::compact() const noexcept -> segments::regular_segment::shared_ptr_
         minHeap.pop();
 
         // Add the smallest element to the merged sequence.
-        // If two elements have the same key, then choose the one with the
-        // greatest timestamp
+        // If two elements have the same key, then choose the one with the greatest timestamp
         if (mergedMemtable.empty() || lastKey != current.first->m_key)
         {
             mergedMemtable.emplace(*current.first);
@@ -171,8 +176,7 @@ auto level_t::compact() const noexcept -> segments::regular_segment::shared_ptr_
     }
 
     // Create a new segment from the compacted segment.
-    // The postfix "_compacted" signals that the segment is an intermediate
-    // result
+    // The postfix "_compacted" signals that the segment is an intermediate result
     auto name{fmt::format("{}_{}_compacted", helpers::segment_name(), index())};
     return segments::factories::lsmtree_segment_factory(
         name, helpers::segment_path(m_pConfig->datadir_path(), name), mergedMemtable
@@ -231,14 +235,13 @@ void level_t::merge(const segments::regular_segment::shared_ptr_t &pSegment) noe
 
     // TODO(lnikon): Make this parameter configurable. Use measurement
     // units(mb).
-    const std::size_t                    segmentSize{1024};
     memtable::memtable_t                 newMemtable;
     segments::storage::segment_storage_t newSegments;
 
     for (const auto &currentRecord : mergedMemtable)
     {
         newMemtable.emplace(currentRecord);
-        if (newMemtable.size() >= segmentSize)
+        if (newMemtable.size() >= m_pConfig->LSMTreeConfig.SegmentSize)
         {
             auto name{fmt::format("{}_{}", helpers::segment_name(), index())};
             newSegments.emplace(

lib/structures/lsmtree/levels/levels.cpp

@@ -98,14 +98,17 @@ auto levels_t::compact() -> segments::regular_segment::shared_ptr_t
             }
         ));
 
-        // Update manifest with new segment
-        ASSERT(m_pManifest->add(
-            db::manifest::manifest_t::segment_record_t{
-                .op = segment_operation_k::add_segment_k,
-                .name = compactedCurrentLevelSegment->get_name(),
-                .level = currentLevel->index()
-            }
-        ));
+        {
+            // Update manifest with new segment
+            auto ok{m_pManifest->add(
+                db::manifest::manifest_t::segment_record_t{
+                    .op = segment_operation_k::add_segment_k,
+                    .name = compactedCurrentLevelSegment->get_name(),
+                    .level = currentLevel->index()
+                }
+            )};
+            ASSERT(ok);
+        }
 
         // If computation succeeded, then flush the compacted segment into disk
         compactedCurrentLevelSegment->flush();
@@ -121,13 +124,16 @@ auto levels_t::compact() -> segments::regular_segment::shared_ptr_t
 
         // Purge the segment representing the compacted level and update the
         // manifest
-        ASSERT(m_pManifest->add(
-            db::manifest::manifest_t::segment_record_t{
-                .op = segment_operation_k::remove_segment_k,
-                .name = compactedCurrentLevelSegment->get_name(),
-                .level = currentLevel->index()
-            }
-        ));
+        {
+            bool ok{m_pManifest->add(
+                db::manifest::manifest_t::segment_record_t{
+                    .op = segment_operation_k::remove_segment_k,
+                    .name = compactedCurrentLevelSegment->get_name(),
+                    .level = currentLevel->index()
+                }
+            )};
+            ASSERT(ok);
+        }
         compactedCurrentLevelSegment->remove_from_disk();
 
         // After merging current level into the next level purge the current
@@ -183,6 +189,7 @@ auto levels_t::size() const noexcept -> levels_t::levels_storage_t::size_type
 [[nodiscard]] auto levels_t::flush_to_level0(memtable::memtable_t memtable) const noexcept
     -> segments::regular_segment::shared_ptr_t
 {
+
     absl::MutexLock lock{&m_mutex};
 
     assert(m_levels[0]);