CraftReplDev write path

conanfile.py

@@ -10,7 +10,7 @@
 
 class HomeBlocksConan(ConanFile):
     name = "homeblocks"
-    version = "6.0.1"
+    version = "6.0.2"
 
     homepage = "https://github.com/eBay/HomeBlocks"
     description = "Block Store built on HomeStore"

src/include/homeblks/home_blocks.hpp

@@ -66,7 +66,8 @@ using volume_handle = std::shared_ptr< volume >;
 // ride result<T> while staying branchable: if (r.error() == volume_error::CRC_MISMATCH) { ... }. Anything with a
 // standard equivalent (invalid arg, no space, io error, unsupported op, ...) is returned as
 // std::make_error_condition(std::errc::*) directly rather than duplicated here.
-ENUM(volume_error, uint16_t, UNKNOWN_VOLUME = 1, CRC_MISMATCH, INDEX_ERROR, INTERNAL_ERROR, OFFLINE);
+ENUM(volume_error, uint16_t, UNKNOWN_VOLUME = 1, CRC_MISMATCH, INDEX_ERROR, INTERNAL_ERROR, OFFLINE,
+     STALE_TERM);
 
 ENUM(volume_state, uint32_t,
      INIT,       // created, not yet online

src/lib/craft/CMakeLists.txt

@@ -6,4 +6,6 @@ target_sources(${PROJECT_NAME}_craft PRIVATE
 )
 target_link_libraries(${PROJECT_NAME}_craft
     ${COMMON_DEPS}
-)
+)
+
+add_subdirectory(tests)

src/lib/craft/craft_repl_dev.cpp

@@ -17,6 +17,39 @@
 
 namespace homeblocks {
 
+// ─── HomeStoreCraftJournalBackend ─────────────────────────────────────────────
+//
+// Production journal backend. Wraps the HomeStore data-journal at client-assigned
+// LSN slots. These calls require HomeStore-side CRAFT journal APIs that are not
+// yet implemented; each method is a stub until those APIs land.
+
+class HomeStoreCraftJournalBackend : public CraftJournalBackend {
+public:
+    async_status write_slot(int64_t lsn, lba_t lba, lba_count_t len,
+                            sisl::sg_list /* data */) override {
+        // TODO: call homestore journal write-at-lsn API (zero-copy, out-of-order safe).
+        LOGW("HomeStoreCraftJournalBackend::write_slot lsn={} lba={} len={} not yet implemented",
+             lsn, lba, len);
+        co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
+    }
+
+    async_result< JournalSlot > read_slot(int64_t lsn) override {
+        // TODO: call homestore journal read-at-lsn API.
+        LOGW("HomeStoreCraftJournalBackend::read_slot lsn={} not yet implemented", lsn);
+        co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
+    }
+
+    async_status truncate_to(int64_t lsn) override {
+        // TODO: call homestore journal truncate API to drop entries above lsn.
+        LOGW("HomeStoreCraftJournalBackend::truncate_to lsn={} not yet implemented", lsn);
+        co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
+    }
+};
+
+unique< CraftJournalBackend > make_homestore_journal_backend() {
+    return std::make_unique< HomeStoreCraftJournalBackend >();
+}
+
 // ─── constructor ──────────────────────────────────────────────────────────────
 
 CraftReplDev::CraftReplDev(volume_id_t vol_id, unique< CraftJournalBackend > journal)
@@ -41,27 +74,60 @@ async_result< LoginResult > CraftReplDev::login(uint64_t /* client_token */,
     co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
 }
 
-async_status CraftReplDev::write(uint64_t /* term */, int64_t /* lsn */,
-                                  lba_t /* lba */, lba_count_t /* len */,
-                                  sisl::sg_list /* data */) {
-    LOGW("CraftReplDev::write not yet implemented");
-    co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
+async_status CraftReplDev::write(uint64_t term, int64_t lsn, int64_t /* glsn */,
+                                  lba_t lba, lba_count_t len, sisl::sg_list data) {
+    // 1. Term check — old-session client; must re-login.
+    if (term != state_.term) {
+        LOGW("write rejected: stale term want={} got={} lsn={}", state_.term, term, lsn);
+        co_return std::unexpected(make_error_condition(volume_error::STALE_TERM));
+    }
+
+    // 2. Update missing set and last_append_lsn under lock.
+    //    Done before the journal write so the missing set is always conservative:
+    //    a failed write leaves the LSN in the set (correctly marked absent).
+    {
+        std::lock_guard lock{missing_mu_};
+        // Any LSNs between the old last_append and this one are gaps.
+        for (int64_t gap = state_.last_append_lsn + 1; gap < lsn; ++gap) {
+            missing_lsns_.insert(gap);
+        }
+        // If this is a new high-watermark or an already-known gap, keep it marked missing until write_slot succeeds.
+        if ((lsn > state_.last_append_lsn) || missing_lsns_.contains(lsn)) {
+            missing_lsns_.insert(lsn);
+        }
+        state_.last_append_lsn = std::max(state_.last_append_lsn, lsn);
+    }
+
+    // 3. Journal append — zero-copy; data buffer is not copied on the hot path.
+    auto res = co_await journal_->write_slot(lsn, lba, len, std::move(data));
+    if (!res) {
+        LOGE("write_slot failed lsn={} lba={} len={}: {}", lsn, lba, len, res.error().message());
+        co_return std::unexpected(res.error());
+    }
+
+
+    {
+        std::lock_guard lock{missing_mu_};
+        missing_lsns_.erase(lsn);
+    }
+
+    LOGT("write ok lsn={} lba={} len={}", lsn, lba, len);
+    co_return homestore::ok();
 }
 
 async_result< sisl::sg_list > CraftReplDev::read(uint64_t /* term */,
-                                                   int64_t /* read_lsn */,
+                                                   int64_t /* min_commit_lsn */,
                                                    lba_t /* lba */, lba_count_t /* len */) {
     LOGW("CraftReplDev::read not yet implemented");
     co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
 }
 
-async_result< LSNPair > CraftReplDev::commit(uint64_t /* term */, int64_t /* lsn */) {
+async_status CraftReplDev::commit(uint64_t /* term */, int64_t /* lsn */) {
     LOGW("CraftReplDev::commit not yet implemented");
     co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
 }
 
-async_result< LSNPair > CraftReplDev::keep_alive(int64_t /* commit_lsn */,
-                                                  int64_t /* all_committed_lsn */) {
+async_status CraftReplDev::keep_alive(int64_t /* commit_lsn */) {
     LOGW("CraftReplDev::keep_alive not yet implemented");
     co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
 }

src/lib/craft/craft_repl_dev.hpp

@@ -61,7 +61,7 @@ struct JournalSlot {
     bool        is_empty{false};
     lba_t       lba{0};
     lba_count_t len{0};
-    sisl::sg_list data;
+    sisl::sg_list data{};
 };
 
 // ─── journal backend abstraction ─────────────────────────────────────────────
@@ -128,6 +128,16 @@ class CraftReplDev {
     // Drop all journal entries with dLSN > lsn; clear missing-set entries above lsn.
     async_status truncate(int64_t lsn);
 
+    // ── observability ──────────────────────────────────────────────────────
+    size_t missing_count() const {
+        std::lock_guard lock{missing_mu_};
+        return missing_lsns_.size();
+    }
+    bool is_missing(int64_t lsn) const {
+        std::lock_guard lock{missing_mu_};
+        return missing_lsns_.contains(lsn);
+    }
+
     // Propose a SyncRSCommitLSN RAFT entry (called by watchdog or leader during login).
     async_status append(int64_t sync_to, uint64_t client_token);
 
@@ -199,10 +209,14 @@ 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_;
+    mutable std::mutex           missing_mu_;
     bool                         login_in_progress_{false};
     std::mutex                   login_mu_;
     CraftRaftListener            raft_listener_;
 };
 
+// Factory that produces the production HomeStore-backed journal. Used by volume.cpp
+// when creating a CRAFT-mode volume. Tests inject MockCraftJournalBackend directly.
+unique< CraftJournalBackend > make_homestore_journal_backend();
+
 } // namespace homeblocks

src/lib/craft/tests/CMakeLists.txt

@@ -0,0 +1,13 @@
+cmake_minimum_required (VERSION 3.11)
+
+add_executable(test_craft_write)
+target_sources(test_craft_write PRIVATE
+    test_craft_write.cpp
+)
+target_link_libraries(test_craft_write
+    ${PROJECT_NAME}
+    ${COMMON_TEST_DEPS}
+    -rdynamic
+)
+
+add_test(NAME CraftWriteTest COMMAND test_craft_write)

src/lib/craft/tests/test_craft_write.cpp

@@ -0,0 +1,153 @@
+/*********************************************************************************
+ * Modifications Copyright 2026 eBay Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed
+ * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+ * CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ *********************************************************************************/
+
+#include <map>
+#include <gtest/gtest.h>
+#include <boost/uuid/random_generator.hpp>
+#include <sisl/options/options.h>
+#include <sisl/logging/logging.h>
+
+#include "../craft_repl_dev.hpp"
+#include "../../coro_helpers.hpp"
+
+SISL_LOGGING_INIT(HOMEBLOCKS_LOG_MODS)
+SISL_OPTIONS_ENABLE(logging)
+
+using namespace homeblocks;
+using namespace homeblocks::detail;
+
+// ─── MockCraftJournalBackend ──────────────────────────────────────────────────
+//
+// In-memory journal for unit tests. Stores slots in a map keyed by LSN;
+// does not copy data buffers (test writes pass empty sg_lists).
+
+class MockCraftJournalBackend : public CraftJournalBackend {
+public:
+    async_status write_slot(int64_t lsn, lba_t lba, lba_count_t len,
+                            sisl::sg_list /* data */) override {
+        slots_[lsn] = JournalSlot{lsn, false, lba, len, {}};
+        co_return homestore::ok();
+    }
+
+    async_result< JournalSlot > read_slot(int64_t lsn) override {
+        auto it = slots_.find(lsn);
+        if (it == slots_.end()) { co_return JournalSlot{.lsn = lsn, .is_empty = true}; }
+        co_return it->second;
+    }
+
+    async_status truncate_to(int64_t lsn) override {
+        std::erase_if(slots_, [lsn](auto& kv) { return kv.first > lsn; });
+        co_return homestore::ok();
+    }
+
+    bool has_slot(int64_t lsn) const { return slots_.contains(lsn); }
+    size_t slot_count() const { return slots_.size(); }
+
+private:
+    std::map< int64_t, JournalSlot > slots_;
+};
+
+// ─── test fixture ─────────────────────────────────────────────────────────────
+
+class CraftWriteTest : public ::testing::Test {
+protected:
+    void SetUp() override {
+        vol_id_ = boost::uuids::random_generator()();
+        auto mock_owned = std::make_unique< MockCraftJournalBackend >();
+        mock_ = mock_owned.get();
+        dev_ = std::make_unique< CraftReplDev >(vol_id_, std::move(mock_owned));
+    }
+
+    // Convenience: drive a write coroutine synchronously.
+    // term=0 matches initial state_.term; lba and glsn equal lsn for simplicity.
+    status do_write(uint64_t term, int64_t lsn) {
+        return sync_get(dev_->write(term, lsn, lsn, static_cast< lba_t >(lsn),
+                                    1 /* len */, sisl::sg_list{}));
+    }
+
+    volume_id_t                   vol_id_;
+    MockCraftJournalBackend*      mock_{nullptr};
+    std::unique_ptr< CraftReplDev > dev_;
+};
+
+// ─── tests ────────────────────────────────────────────────────────────────────
+
+// In-order writes: lsns 0,1,2 arrive sequentially.
+// Missing set stays empty; last_append_lsn advances to 2; all slots land in journal.
+TEST_F(CraftWriteTest, InOrderWrites) {
+    for (int64_t lsn = 0; lsn < 3; ++lsn) {
+        auto res = do_write(0 /* term */, lsn);
+        ASSERT_TRUE(res.has_value()) << "write lsn=" << lsn << " failed: " << res.error().message();
+    }
+
+    EXPECT_EQ(mock_->slot_count(), 3u);
+    EXPECT_EQ(dev_->missing_count(), 0u);
+
+    auto lsns = sync_get(dev_->get_lsns(vol_id_));
+    ASSERT_TRUE(lsns.has_value());
+    EXPECT_EQ(lsns->last_append_lsn, 2);
+    EXPECT_EQ(lsns->commit_lsn, -1); // commit() is S3; nothing committed yet
+}
+
+// Out-of-order writes: lsn=4 arrives before lsn=1,2,3.
+// Missing set should contain {1,2,3}; filling them in clears the set.
+TEST_F(CraftWriteTest, OutOfOrderWrites) {
+    // First write establishes baseline.
+    ASSERT_TRUE(do_write(0, 0).has_value());
+    EXPECT_EQ(dev_->missing_count(), 0u);
+
+    // Jump ahead — creates gaps {1, 2, 3}.
+    ASSERT_TRUE(do_write(0, 4).has_value());
+    EXPECT_EQ(dev_->missing_count(), 3u);
+    EXPECT_TRUE(dev_->is_missing(1));
+    EXPECT_TRUE(dev_->is_missing(2));
+    EXPECT_TRUE(dev_->is_missing(3));
+    EXPECT_FALSE(dev_->is_missing(0));
+    EXPECT_FALSE(dev_->is_missing(4));
+
+    // Fill in the gaps in reverse order.
+    ASSERT_TRUE(do_write(0, 3).has_value());
+    ASSERT_TRUE(do_write(0, 2).has_value());
+    ASSERT_TRUE(do_write(0, 1).has_value());
+    EXPECT_EQ(dev_->missing_count(), 0u);
+
+    EXPECT_EQ(mock_->slot_count(), 5u);
+
+    auto lsns = sync_get(dev_->get_lsns(vol_id_));
+    ASSERT_TRUE(lsns.has_value());
+    EXPECT_EQ(lsns->last_append_lsn, 4);
+}
+
+// Term rejection: write with term != state_.term returns STALE_TERM.
+// Journal must not be touched.
+TEST_F(CraftWriteTest, TermRejection) {
+    // state_.term starts at 0; term=1 does not match and should be rejected.
+    auto res = do_write(1 /* wrong term */, 0);
+    ASSERT_FALSE(res.has_value());
+    EXPECT_EQ(res.error(), make_error_condition(volume_error::STALE_TERM));
+
+    // Journal untouched; missing set empty.
+    EXPECT_EQ(mock_->slot_count(), 0u);
+    EXPECT_EQ(dev_->missing_count(), 0u);
+}
+
+// ─── main ─────────────────────────────────────────────────────────────────────
+
+int main(int argc, char** argv) {
+    SISL_OPTIONS_LOAD(argc, argv, logging);
+    sisl::logging::SetLogger("test_craft_write");
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}