craft: S4 truncate path — CraftReplDev::truncate

conanfile.py

@@ -10,7 +10,7 @@
 
 class HomeBlocksConan(ConanFile):
     name = "homeblocks"
-    version = "6.0.2"
+    version = "6.0.3"
 
     homepage = "https://github.com/eBay/HomeBlocks"
     description = "Block Store built on HomeStore"
@@ -48,7 +48,7 @@ def configure(self):
 
     def build_requirements(self):
         self.test_requires("gtest/[^1.17]")
-        self.test_requires("ublkpp/[^0.34]@oss/main")
+        self.test_requires("ublkpp/[^0.35]@oss/main")
 
     def requirements(self):
         self.requires("homestore/[^8.0]@oss/dev", transitive_headers=True)

src/lib/craft/CMakeLists.txt

@@ -13,3 +13,5 @@ target_link_libraries(${PROJECT_NAME}_craft
 # (the volume_handle free functions) and craft_repl_dev.cpp is the engine behind it. The CRAFT client, the wire
 # codec, the in-memory reference model and the reference server all live in the standalone craft_client package;
 # the ublk driver lives in ublkpp. None of them are built or linked here.
+
+add_subdirectory(tests)

src/lib/craft/craft_repl_dev.cpp

@@ -15,25 +15,127 @@
 
 #include "craft_repl_dev.hpp"
 
+#include <homestore/logstore/log_store.hpp> // home_log_store, logstore_seq_num_t
+
 namespace homeblocks {
 
+// ─── HomeStore journal backend ────────────────────────────────────────────────
+//
+// Production backend: wraps a HomeStore home_log_store. write_slot / read_slot
+// are stubs implemented in S2 (PR #165). truncate_to() is implemented here (S4).
+
+class HomeStoreCraftJournalBackend : public CraftJournalBackend {
+public:
+    explicit HomeStoreCraftJournalBackend(shared< homestore::home_log_store > logstore) :
+            logstore_{std::move(logstore)} {}
+
+    async_status write_slot(int64_t lsn, lba_t /* lba */, lba_count_t /* len */, sisl::sg_list /* data */) override {
+        LOGW("HomeStoreCraftJournalBackend::write_slot lsn={} not yet implemented", lsn);
+        co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
+    }
+
+    async_result< JournalSlot > read_slot(int64_t lsn) override {
+        LOGW("HomeStoreCraftJournalBackend::read_slot lsn={} not yet implemented", lsn);
+        co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
+    }
+
+    // Drop all journal entries with seq_num > lsn; lsn becomes the new tail.
+    // home_log_store::rollback(to_lsn) removes everything ABOVE to_lsn, which is exactly what we need.
+    async_status truncate_to(int64_t lsn) override {
+        if (!logstore_->rollback(static_cast< homestore::logstore_seq_num_t >(lsn)))
+            co_return std::unexpected(std::make_error_condition(std::errc::io_error));
+        co_return ok();
+    }
+
+private:
+    shared< homestore::home_log_store > logstore_;
+};
+
+unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore::home_log_store > logstore) {
+    return std::make_unique< HomeStoreCraftJournalBackend >(std::move(logstore));
+}
+
 // ─── constructor ──────────────────────────────────────────────────────────────
 
 CraftReplDev::CraftReplDev(volume_id_t vol_id, unique< CraftJournalBackend > journal) :
         vol_id_{vol_id}, journal_{std::move(journal)}, raft_listener_{this} {}
 
 // ─── get_lsns / get_rs_commit_lsn ────────────────────────────────────────────
-// These are real: they just snapshot the in-memory partition state.
+// Snapshot the in-memory partition state under missing_mu_ for consistency with
+// write() which updates state_ under the same lock.
 
 async_result< craft::lsn_pair > CraftReplDev::get_lsns(volume_id_t /* vol_id */) {
-    co_return craft::lsn_pair{state_.commit_lsn, state_.last_append_lsn};
+    craft::lsn_pair pair{};
+    {
+        std::lock_guard lk{missing_mu_};
+        pair = {state_.commit_lsn, state_.last_append_lsn};
+    }
+    co_return pair;
 }
 
 async_result< craft::lsn_pair > CraftReplDev::get_rs_commit_lsn() {
-    co_return craft::lsn_pair{state_.commit_lsn, state_.last_append_lsn};
+    craft::lsn_pair pair{};
+    {
+        std::lock_guard lk{missing_mu_};
+        pair = {state_.commit_lsn, state_.last_append_lsn};
+    }
+    co_return pair;
+}
+
+// ─── truncate (S4) ────────────────────────────────────────────────────────────
+//
+// Called only during the login sequence, while no writes are in-flight (the
+// CRAFT write path is quiesced by login serialisation). Three atomic steps:
+//   1. Journal rollback: drop all entries with dLSN > lsn (tail truncation).
+//   2. Clamp last_append_lsn to lsn if it is higher.
+//   3. Erase all missing-set entries above lsn.
+// commit_lsn is not touched: the new rs_commit_lsn passed by the caller is the
+// dLSN up to which RAFT consensus has RESOLVED entries. Entries above that are
+// the ones being dropped. The missing set tracks gaps in [commit_lsn+1,
+// last_append_lsn]; after truncation every entry > lsn is gone from both the
+// journal and the missing set.
+
+async_status CraftReplDev::truncate(int64_t lsn) {
+    // Guard before touching the journal: truncating below commit_lsn would drop
+    // committed entries and break the commit_lsn <= last_append_lsn invariant.
+    {
+        std::lock_guard lk{missing_mu_};
+        DEBUG_ASSERT_GE(lsn, state_.commit_lsn, "truncate below committed prefix");
+    }
+
+    // Step 1: journal rollback — synchronous; fails fast on I/O error.
+    if (auto r = co_await journal_->truncate_to(lsn); !r) co_return r;
+
+    // Steps 2 + 3 under the same mutex write() uses.
+    {
+        std::lock_guard lk{missing_mu_};
+        if (state_.last_append_lsn > lsn) state_.last_append_lsn = lsn;
+        missing_lsns_.erase(missing_lsns_.upper_bound(lsn), missing_lsns_.end());
+    }
+    co_return ok();
+}
+
+#ifdef _PRERELEASE
+void CraftReplDev::seed_lsns(int64_t last_append, std::initializer_list< int64_t > missing) {
+    std::lock_guard lk{missing_mu_};
+    state_.last_append_lsn = last_append;
+    missing_lsns_.clear();
+    missing_lsns_.insert(missing);
 }
 
-// ─── stubs (S2–S7 will implement these) ──────────────────────────────────────
+void CraftReplDev::seed_commit_lsn(int64_t commit) {
+    std::lock_guard lk{missing_mu_};
+    state_.commit_lsn = commit;
+}
+
+void CraftReplDev::seed_empty(std::initializer_list< int64_t > empty) {
+    std::lock_guard lk{missing_mu_};
+    empty_lsns_.clear();
+    empty_lsns_.insert(empty);
+}
+#endif
+
+// ─── stubs (S1/S2/S3/S5/S6/S7 implement these) ───────────────────────────────
 
 async_result< craft::LoginResult > CraftReplDev::login(uint64_t /* client_token */) {
     LOGW("CraftReplDev::login not yet implemented");
@@ -69,19 +171,59 @@ async_result< craft::resolution_result > CraftReplDev::request_resolution(craft:
     co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
 }
 
-async_status CraftReplDev::truncate(int64_t /* lsn */) {
-    LOGW("CraftReplDev::truncate not yet implemented");
-    co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
-}
-
 async_status CraftReplDev::append(int64_t /* sync_to */, uint64_t /* client_token */) {
     LOGW("CraftReplDev::append not yet implemented");
     co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
 }
 
-async_result< std::vector< JournalSlot > > CraftReplDev::fetch_data(std::vector< int64_t > /* lsns */) {
-    LOGW("CraftReplDev::fetch_data not yet implemented");
-    co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
+// ─── fetch_data (S6) ──────────────────────────────────────────────────────────
+//
+// Four-way response per requested LSN (in request order; omitted slots absent):
+//   is_empty=true  : slot positively verdicted Empty by a prior SyncRSCommitLSN (S5); Empty beats data
+//   present+data   : slot in journal, all_zeros=false; data payload present
+//   present+zero   : slot in journal, all_zeros=true; no data payload (WRITE_ZEROES / range-unmap)
+//   omitted        : slot not locally held — in missing_lsns_ or above last_append_lsn
+//
+// The Empty verdict is leader-only (S5 pre-resolution); this responder only reports state.
+// empty_lsns_ is checked first: a slot in both empty_lsns_ and the journal returns is_empty=true
+// (Empty beats data, the reconciliation invariant from S5).
+//
+// The missing_mu_ lock is dropped before each co_await read_slot() call to avoid holding a mutex
+// across a suspension point. Callers are serialised by the login sequence (no concurrent writes
+// while fetch_data runs), so the snapshot taken under the lock is stable.
+//
+// A read_slot() I/O error aborts the batch immediately (fail-fast); the partial result is discarded.
+
+async_result< std::vector< JournalSlot > > CraftReplDev::fetch_data(std::vector< int64_t > lsns) {
+    std::vector< JournalSlot > result;
+    result.reserve(lsns.size());
+
+    for (int64_t lsn : lsns) {
+        enum class SlotKind { Empty, Present, Absent };
+        SlotKind kind;
+        {
+            std::lock_guard lk{missing_mu_};
+            if (empty_lsns_.contains(lsn)) {
+                kind = SlotKind::Empty;
+            } else if (lsn >= 0 && lsn <= state_.last_append_lsn && !missing_lsns_.contains(lsn)) {
+                kind = SlotKind::Present;
+            } else {
+                kind = SlotKind::Absent;
+            }
+        }
+
+        if (kind == SlotKind::Empty) {
+            result.push_back(JournalSlot{.lsn = lsn, .is_empty = true});
+        } else if (kind == SlotKind::Present) {
+            auto slot_r = co_await journal_->read_slot(lsn);
+            if (!slot_r) co_return std::unexpected(slot_r.error());
+            slot_r->lsn = lsn;
+            result.push_back(std::move(*slot_r));
+        }
+        // Absent: omit from result (not-present-here)
+    }
+
+    co_return result;
 }
 
 // ─── RAFT listener ────────────────────────────────────────────────────────────
@@ -95,17 +237,14 @@ void CraftReplDev::CraftRaftListener::on_commit(int64_t lsn, sisl::blob const& /
     LOGD("CraftRaftListener::on_commit lsn={} (entry dispatch not yet implemented)", lsn);
 }
 
-// ─── RAFT apply helpers (S5 will implement) ───────────────────────────────────
+// ─── RAFT apply helpers (S5 implements) ──────────────────────────────────────
 
 void CraftReplDev::apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint64_t /* client_token */) {
-    // Advance commit_lsn to rs_commit_lsn, calling fetch_data() for any missing slots.
     LOGD("apply_sync_rs_commit_lsn rs_commit_lsn={} (not yet implemented)", rs_commit_lsn);
 }
 
 void CraftReplDev::apply_internal_login(uint64_t client_token, uint64_t term) {
-    // Update state_.client_token and state_.term; subsequent IOs with a different
-    // term will be rejected with ETERM.
     LOGD("apply_internal_login client_token={} term={} (not yet implemented)", client_token, term);
 }
 
-} // namespace homeblocks
+} // namespace homeblocks

src/lib/craft/craft_repl_dev.hpp

@@ -18,10 +18,15 @@
 #include <homestore/replication/repl_dev.hpp>
 
 #include <atomic>
+#include <initializer_list>
 #include <mutex>
 #include <set>
 #include <vector>
 
+namespace homestore {
+class home_log_store;
+}
+
 namespace homeblocks {
 
 // ─── CRAFT vocabulary vs. this backend's own state ───────────────────────────
@@ -67,14 +72,20 @@ class CraftJournalBackend {
 public:
     virtual async_status write_slot(int64_t lsn, lba_t lba, lba_count_t len, sisl::sg_list data) = 0;
     virtual async_result< JournalSlot > read_slot(int64_t lsn) = 0;
+    // Drop all entries with seq_num > lsn; lsn becomes the new journal tail.
     virtual async_status truncate_to(int64_t lsn) = 0;
     virtual ~CraftJournalBackend() = default;
 };
 
+// Factory that wraps a HomeStore log store. Used by volume.cpp when creating a CRAFT-mode volume.
+// Tests inject MockCraftJournalBackend directly.
+unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore::home_log_store > logstore);
+
 // ─── CraftReplDev ─────────────────────────────────────────────────────────────
 //
-// Parallel to HomeStore's ReplDisk. Each CRAFT-mode volume owns one instance
-// instead of the solo repl_dev. Non-CRAFT volumes are unaffected.
+// One instance per CRAFT-mode volume. Implements the full CRAFT data plane
+// (write, read, login, truncate, ...) on top of a HomeStore log store and
+// index. Non-CRAFT volumes are unaffected.
 
 class CraftReplDev {
 public:
@@ -140,7 +151,8 @@ class CraftReplDev {
     // Alias of get_lsns exposed to peer servers during GetRSCommitLSN broadcast.
     async_result< craft::lsn_pair > get_rs_commit_lsn();
 
-    // Drop all journal entries with dLSN > lsn; clear missing-set entries above lsn.
+    // Drop all journal entries with dLSN > lsn; clear missing-set entries above lsn; clamp last_append_lsn.
+    // Called only during login (quiesced -- no concurrent writes). commit_lsn is NOT changed.
     async_status truncate(int64_t lsn);
 
     // Propose a SyncRSCommitLSN RAFT entry (called by watchdog or leader during login).
@@ -150,6 +162,42 @@ class CraftReplDev {
     // JournalSlot{.is_empty=true} rather than an error.
     async_result< std::vector< JournalSlot > > fetch_data(std::vector< int64_t > lsns);
 
+    // ── observability ─────────────────────────────────────────────────────
+
+    size_t missing_count() const {
+        std::lock_guard lk{missing_mu_};
+        return missing_lsns_.size();
+    }
+
+    bool is_missing(int64_t lsn) const {
+        std::lock_guard lk{missing_mu_};
+        return missing_lsns_.contains(lsn);
+    }
+
+    bool is_empty_slot(int64_t lsn) const {
+        std::lock_guard lk{missing_mu_};
+        return empty_lsns_.contains(lsn);
+    }
+
+    int64_t last_append_lsn() const {
+        std::lock_guard lk{missing_mu_};
+        return state_.last_append_lsn;
+    }
+    int64_t commit_lsn() const {
+        std::lock_guard lk{missing_mu_};
+        return state_.commit_lsn;
+    }
+
+#ifdef _PRERELEASE
+    // Seeds partition watermarks and the missing set directly, bypassing write().
+    // Only compiled when _PRERELEASE is defined; never present in production binaries.
+    void seed_lsns(int64_t last_append, std::initializer_list< int64_t > missing = {});
+    // Seeds commit_lsn independently of seed_lsns (which only touches last_append + missing).
+    void seed_commit_lsn(int64_t commit);
+    // Seeds the Empty-verdict set; does not affect missing_lsns_. Clears any prior seeded empties.
+    void seed_empty(std::initializer_list< int64_t > empty);
+#endif
+
 private:
     // ── RAFT listener ──────────────────────────────────────────────────────
     //
@@ -215,10 +263,11 @@ class CraftReplDev {
     unique< CraftJournalBackend > journal_;
     CraftPartitionState state_;
     std::set< int64_t > missing_lsns_; // gaps between commit_lsn and last_append_lsn
-    std::mutex missing_mu_;
+    std::set< int64_t > empty_lsns_;   // slots positively verdicted Empty by a prior SyncRSCommitLSN (S5)
+    mutable std::mutex missing_mu_;    // guards state_, missing_lsns_, and empty_lsns_
     bool login_in_progress_{false};
     std::mutex login_mu_;
     CraftRaftListener raft_listener_;
 };
 
-} // namespace homeblocks
+} // namespace homeblocks

src/lib/craft/tests/CMakeLists.txt

@@ -0,0 +1,31 @@
+cmake_minimum_required(VERSION 3.11)
+
+# Unit test for CraftReplDev::truncate (S4). Compiles craft_repl_dev.cpp directly to avoid
+# dragging in craft_api.cpp → volume.hpp (heavy HomeStore volume plumbing not needed here).
+add_executable(test_craft_truncate)
+target_sources(test_craft_truncate PRIVATE
+    test_craft_truncate.cpp
+    ../craft_repl_dev.cpp
+)
+target_compile_definitions(test_craft_truncate PRIVATE _PRERELEASE)
+target_link_libraries(test_craft_truncate
+    ${COMMON_TEST_DEPS}
+    -rdynamic
+)
+
+add_test(NAME CraftTruncateTest COMMAND test_craft_truncate)
+
+# Unit tests for CraftReplDev::get_lsns(), get_rs_commit_lsn(), and fetch_data() (S6).
+# Same pattern as test_craft_truncate: compile craft_repl_dev.cpp directly.
+add_executable(test_craft_peer_exchange)
+target_sources(test_craft_peer_exchange PRIVATE
+    test_craft_peer_exchange.cpp
+    ../craft_repl_dev.cpp
+)
+target_compile_definitions(test_craft_peer_exchange PRIVATE _PRERELEASE)
+target_link_libraries(test_craft_peer_exchange
+    ${COMMON_TEST_DEPS}
+    -rdynamic
+)
+
+add_test(NAME CraftPeerExchangeTest COMMAND test_craft_peer_exchange)