craft: S2 write path — HS_DATA_LINKED journal append with full correctness hardening

conanfile.py

@@ -10,7 +10,7 @@
 
 class HomeBlocksConan(ConanFile):
     name = "homeblocks"
-    version = "6.0.4"
+    version = "6.0.5"
 
     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.35]@oss/main")
+        self.test_requires("ublkpp/[^0.36]@oss/main")
 
     def requirements(self):
         self.requires("homestore/[^8.0]@oss/dev", transitive_headers=True)

src/include/homeblks/home_blocks.hpp

@@ -68,7 +68,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, STALE_TERM);
+ENUM(volume_error, uint16_t, UNKNOWN_VOLUME = 1, CRC_MISMATCH, INDEX_ERROR, INTERNAL_ERROR, OFFLINE, STALE_TERM,
+     EMPTY_SLOT);
 
 ENUM(volume_state, uint32_t,
      INIT,       // created, not yet online

src/lib/craft/craft_repl_dev.cpp

@@ -15,23 +15,99 @@
 
 #include "craft_repl_dev.hpp"
 
-#include <homestore/logstore/log_store.hpp> // home_log_store, logstore_seq_num_t
+#include <coroutine>
+#include <cstring>
+
+#include <homestore/blkdata_service.hpp>    // data_service(), async_alloc_write, blk_alloc_hints
+#include <homestore/logstore/log_store.hpp> // home_log_store, logstore_seq_num_t, log_write_comp_cb_t
 
 namespace homeblocks {
 
-// ─── HomeStore journal backend ────────────────────────────────────────────────
+// ─── Journal entry on-disk format ─────────────────────────────────────────────
+//
+// Each log slot is: [CraftJournalEntry header][serialized multi_blk_id bytes].
+// The payload (HS_DATA_LINKED) is written directly to the data service; only the
+// block reference is stored here.
+
+#pragma pack(1)
+struct CraftJournalEntry {
+    lba_t lba;
+    lba_count_t len;
+    uint8_t all_zeros;
+};
+#pragma pack()
+
+// ─── Callback-to-coroutine bridge ─────────────────────────────────────────────
+//
+// home_log_store::write_async is callback-based. This C++20 awaitable bridges it
+// to co_await: await_suspend calls write_async and captures the coroutine handle;
+// the callback resumes it when the write completes.
+//
+// blob_ lives in the coroutine frame for the full suspension, so write_async
+// always sees a valid buffer.
+//
+// KNOWN RISK (shutdown): write_async returns 0 without calling `cb` when is_stopping() is true
+// (HomeStore/src/lib/logstore/log_store.cpp:71). The coroutine is permanently suspended until the
+// process exits. Callers must drain all in-flight writes before initiating HomeStore shutdown.
+//
+// KNOWN GAP (I/O errors): write_async fires the same callback for both success and I/O failure
+// with no status argument. await_resume cannot surface the error, so write_slot always returns
+// ok() regardless of the underlying I/O result. A future fix requires a HomeStore API extension.
+
+struct LogstoreWriteAwaitable {
+    homestore::home_log_store* store_;
+    homestore::logstore_seq_num_t seq_;
+    sisl::io_blob_safe blob_;
+
+    bool await_ready() const noexcept { return false; }
+
+    template < typename H >
+    void await_suspend(H h) noexcept {
+        store_->write_async(
+            seq_, blob_, nullptr,
+            [h](homestore::logstore_seq_num_t, sisl::io_blob&, homestore::logdev_key, void*) mutable { h.resume(); });
+    }
+
+    void await_resume() const noexcept {}
+};
+
+// ─── 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).
+// Production backend: wraps a HomeStore home_log_store.
 
 class HomeStoreCraftJournalBackend : public CraftJournalBackend {
 public:
-    explicit HomeStoreCraftJournalBackend(shared< homestore::home_log_store > logstore) :
-            logstore_{std::move(logstore)} {}
+    explicit HomeStoreCraftJournalBackend(shared< homestore::home_log_store > logstore, uint64_t vol_ordinal) :
+            logstore_{std::move(logstore)}, vol_ordinal_{vol_ordinal} {}
+
+    // Allocate blocks via the HomeStore data service and write the payload (zero-copy).
+    // application_hint routes the allocation to this volume's chunk set via VolumeChunkSelector.
+    // Returns the allocated multi_blk_id on success.
+    async_result< homestore::multi_blk_id > alloc_write_data(sisl::sg_list const& data,
+                                                             lba_count_t /* len */) override {
+        homestore::blk_alloc_hints hints;
+        hints.application_hint = vol_ordinal_;
+        homestore::multi_blk_id blkid{};
+        auto res = co_await homestore::data_service().async_alloc_write(data, hints, blkid);
+        if (!res) {
+            LOGE("async_alloc_write failed: {}", res.error().message());
+            co_return std::unexpected(make_error_condition(volume_error::INTERNAL_ERROR));
+        }
+        co_return blkid;
+    }
 
-    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));
+    // Serialize the journal entry (header + blkid) and write it to the log store.
+    async_status write_slot(int64_t lsn, lba_t lba, lba_count_t len, homestore::multi_blk_id blkid,
+                            bool all_zeros) override {
+        CraftJournalEntry hdr{lba, len, static_cast< uint8_t >(all_zeros)};
+        uint32_t blkid_sz = blkid.serialized_size();
+        sisl::io_blob_safe blob{static_cast< uint32_t >(sizeof(CraftJournalEntry)) + blkid_sz};
+        std::memcpy(blob.bytes(), &hdr, sizeof(CraftJournalEntry));
+        sisl::blob blkid_blob = blkid.serialize(); // non-owning view — copy before blkid goes out of scope
+        std::memcpy(blob.bytes() + sizeof(CraftJournalEntry), blkid_blob.cbytes(), blkid_sz);
+        co_await LogstoreWriteAwaitable{logstore_.get(), static_cast< homestore::logstore_seq_num_t >(lsn),
+                                        std::move(blob)};
+        co_return ok();
     }
 
     async_result< JournalSlot > read_slot(int64_t lsn) override {
@@ -47,12 +123,23 @@ class HomeStoreCraftJournalBackend : public CraftJournalBackend {
         co_return ok();
     }
 
+    async_status free_data(homestore::multi_blk_id blkid) override {
+        auto res = co_await homestore::data_service().async_free_blk(blkid);
+        if (!res) {
+            LOGE("async_free_blk failed: {}", res.error().message());
+            co_return std::unexpected(make_error_condition(volume_error::INTERNAL_ERROR));
+        }
+        co_return ok();
+    }
+
 private:
     shared< homestore::home_log_store > logstore_;
+    uint64_t vol_ordinal_;
 };
 
-unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore::home_log_store > logstore) {
-    return std::make_unique< HomeStoreCraftJournalBackend >(std::move(logstore));
+unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore::home_log_store > logstore,
+                                                             uint64_t vol_ordinal) {
+    return std::make_unique< HomeStoreCraftJournalBackend >(std::move(logstore), vol_ordinal);
 }
 
 // ─── constructor ──────────────────────────────────────────────────────────────
@@ -132,6 +219,10 @@ void CraftReplDev::seed_empty(std::initializer_list< int64_t > empty) {
     std::lock_guard lk{missing_mu_};
     empty_lsns_.clear();
     empty_lsns_.insert(empty);
+    // Empty verdict resolves a gap: an LSN that was already in missing_lsns_ must be removed so
+    // it does not permanently stall commit advancement. apply_sync_rs_commit_lsn (S5) must do the same.
+    for (int64_t lsn : empty)
+        missing_lsns_.erase(lsn);
 }
 #endif
 
@@ -148,34 +239,104 @@ async_status CraftReplDev::logout(craft::client_hdr /* hdr */) {
 }
 
 async_result< craft::lsn_pair > CraftReplDev::write(craft::client_hdr hdr, int64_t dlsn, uint64_t addr, uint64_t len,
-                                                    sisl::sg_list data) {
-    if (hdr.term != state_.term) {
-        LOGW("write rejected: stale term want={} got={} dlsn={}", state_.term, hdr.term, dlsn);
-        co_return std::unexpected(make_error_condition(volume_error::STALE_TERM));
+                                                    sisl::sg_list data, bool all_zeros) {
+    if (dlsn < 0) {
+        LOGW("write rejected: invalid dlsn={}", dlsn);
+        co_return std::unexpected(make_error_condition(std::errc::invalid_argument));
     }
+    RELEASE_ASSERT(all_zeros || data.size > 0,
+                   "write: all_zeros=false requires non-empty data; set all_zeros=true for WRITE_ZEROES");
 
     {
         std::lock_guard lock{missing_mu_};
-        for (int64_t gap = state_.last_append_lsn + 1; gap < dlsn; ++gap)
-            missing_lsns_.insert(gap);
+        // Term check is inside the lock: state_.term is mutated by apply_internal_login (S5) under the same mutex.
+        if (hdr.term != state_.term) {
+            LOGW("write rejected: stale term want={} got={} dlsn={}", state_.term, hdr.term, dlsn);
+            co_return std::unexpected(make_error_condition(volume_error::STALE_TERM));
+        }
+        if (empty_lsns_.contains(dlsn)) {
+            LOGW("write rejected: slot is permanently empty dlsn={}", dlsn);
+            co_return std::unexpected(make_error_condition(volume_error::EMPTY_SLOT));
+        }
+        // Idempotent: slot already successfully written — return snapshot without a second write_slot call.
+        if (dlsn <= state_.last_append_lsn && !missing_lsns_.contains(dlsn)) {
+            LOGT("write idempotent: dlsn={} already written", dlsn);
+            co_return craft::lsn_pair{state_.commit_lsn, state_.last_append_lsn};
+        }
+        static constexpr int64_t k_max_ooo_gap = 1'000'000;
+        // Guard 1: prevent signed overflow in the gap subtraction below (dlsn near INT64_MAX).
+        if (dlsn > INT64_MAX - k_max_ooo_gap) {
+            LOGW("write rejected: dlsn={} exceeds safe LSN range", dlsn);
+            co_return std::unexpected(make_error_condition(std::errc::invalid_argument));
+        }
+        // Guard 2: cap the gap to prevent unbounded per-write allocation in missing_lsns_.
+        if (dlsn - state_.last_append_lsn > k_max_ooo_gap) {
+            LOGW("write rejected: dlsn={} too far ahead of last_append_lsn={}", dlsn, state_.last_append_lsn);
+            co_return std::unexpected(make_error_condition(std::errc::value_too_large));
+        }
+        // Empty-verdicted LSNs in the gap are already resolved — skip them to avoid re-stalling commit advancement.
+        for (int64_t gap = state_.last_append_lsn + 1; gap < dlsn; ++gap) {
+            if (!empty_lsns_.contains(gap)) missing_lsns_.insert(gap);
+        }
         if ((dlsn > state_.last_append_lsn) || missing_lsns_.contains(dlsn)) missing_lsns_.insert(dlsn);
         state_.last_append_lsn = std::max(state_.last_append_lsn, dlsn);
     }
 
-    auto res = co_await journal_->write_slot(dlsn, static_cast< lba_t >(addr), static_cast< lba_count_t >(len),
-                                             std::move(data));
+    // HS_DATA_LINKED: allocate blocks and write payload before journalling the block reference.
+    // all_zeros=true (WRITE_ZEROES/unmap) and empty data (test stubs) skip the data-service write.
+    //
+    // LATENT RISK: if the caller sends all_zeros=false with data.size==0 (e.g., a test stub with an
+    // empty sg_list but all_zeros not set), the journal entry records all_zeros=0 with a default-
+    // constructed blkid. On replay this produces a read against a zero blkid. The client wire rejects
+    // this, but internal and test paths must set all_zeros=true whenever no payload is provided.
+    homestore::multi_blk_id blkid{};
+    bool blkid_allocated = false;
+    if (!all_zeros && data.size > 0) {
+        auto alloc_res = co_await journal_->alloc_write_data(data, static_cast< lba_count_t >(len));
+        if (!alloc_res) {
+            LOGE("alloc_write_data failed dlsn={}: {}", dlsn, alloc_res.error().message());
+            co_return std::unexpected(alloc_res.error());
+        }
+        blkid = *alloc_res;
+        blkid_allocated = true;
+    }
+    auto res = co_await journal_->write_slot(dlsn, static_cast< lba_t >(addr), static_cast< lba_count_t >(len), blkid,
+                                             all_zeros);
     if (!res) {
         LOGE("write_slot failed dlsn={} addr={} len={}: {}", dlsn, addr, len, res.error().message());
+        if (blkid_allocated) {
+            if (auto fr = co_await journal_->free_data(blkid); !fr)
+                LOGE("free_data failed after write_slot failure dlsn={}: {}", dlsn, fr.error().message());
+        }
         co_return std::unexpected(res.error());
     }
 
+    // Re-validate term: a new login completing while write_slot was in flight may have run
+    // truncate(). Discarding this result prevents a ghost journal entry surviving past the truncation.
+    // The lock is taken to snapshot the term and update state_, then released BEFORE free_data so we
+    // do not co_await while holding missing_mu_.
+    bool stale_post_flight = false;
     craft::lsn_pair snapshot;
     {
         std::lock_guard lock{missing_mu_};
-        missing_lsns_.erase(dlsn);
-        snapshot = {state_.commit_lsn, state_.last_append_lsn};
+        if (hdr.term != state_.term) {
+            LOGW("write discarded post-flight: term changed dlsn={}", dlsn);
+            stale_post_flight = true;
+        } else {
+            missing_lsns_.erase(dlsn);
+            snapshot = {state_.commit_lsn, state_.last_append_lsn};
+        }
     }
-    LOGT("write ok dlsn={} addr={} len={}", dlsn, addr, len);
+
+    if (stale_post_flight) {
+        if (blkid_allocated) {
+            if (auto fr = co_await journal_->free_data(blkid); !fr)
+                LOGE("free_data failed after post-flight stale-term discard dlsn={}: {}", dlsn, fr.error().message());
+        }
+        co_return std::unexpected(make_error_condition(volume_error::STALE_TERM));
+    }
+
+    LOGT("write ok dlsn={} addr={} len={} all_zeros={}", dlsn, addr, len, all_zeros);
     co_return snapshot;
 }
 

src/lib/craft/craft_repl_dev.hpp

@@ -70,16 +70,25 @@ struct JournalSlot {
 
 class CraftJournalBackend {
 public:
-    virtual async_status write_slot(int64_t lsn, lba_t lba, lba_count_t len, sisl::sg_list data) = 0;
+    // Allocate blocks and write the data payload. Called BEFORE write_slot for non-zero writes.
+    // all_zeros=true and empty data bypass this; write_slot receives an empty multi_blk_id.
+    virtual async_result< homestore::multi_blk_id > alloc_write_data(sisl::sg_list const& data, lba_count_t len) = 0;
+    virtual async_status write_slot(int64_t lsn, lba_t lba, lba_count_t len, homestore::multi_blk_id blkid,
+                                    bool all_zeros) = 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;
+    // Release blocks previously allocated by alloc_write_data. Called when write_slot fails or
+    // when the write is discarded post-flight (stale term). Free errors are logged but non-fatal.
+    virtual async_status free_data(homestore::multi_blk_id blkid) = 0;
     virtual ~CraftJournalBackend() = default;
 };
 
 // Factory that wraps a HomeStore log store. Used by volume.cpp when creating a CRAFT-mode volume.
+// vol_ordinal must match vol_info_->ordinal so async_alloc_write routes to this volume's chunks.
 // Tests inject MockCraftJournalBackend directly.
-unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore::home_log_store > logstore);
+unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore::home_log_store > logstore,
+                                                             uint64_t vol_ordinal);
 
 // ─── CraftReplDev ─────────────────────────────────────────────────────────────
 //
@@ -112,11 +121,12 @@ class CraftReplDev {
     async_status logout(craft::client_hdr hdr);
 
     // Append data at the client-assigned dLSN. Zero-copy; does NOT apply to the LBA index (hdr.commit_lsn drives
-    // that). An EMPTY `data` is a zero write (WRITE_ZEROES/unmap over [addr, addr+len)). The ack returns the
-    // achieved {commit_lsn, last_append_lsn} snapshotted with the append -- every CRAFT IO response piggybacks the
+    // that). Set all_zeros=true for WRITE_ZEROES/unmap over [addr, addr+len); data must be empty in that case.
+    // Precondition: all_zeros=false requires non-empty data (data.size > 0). The ack returns the achieved
+    // {commit_lsn, last_append_lsn} snapshotted with the append -- every CRAFT IO response piggybacks the
     // watermarks (the wire's write_rsp), so any round-trip refreshes the client's model of this member.
     async_result< craft::lsn_pair > write(craft::client_hdr hdr, int64_t dlsn, uint64_t addr, uint64_t len,
-                                          sisl::sg_list data);
+                                          sisl::sg_list data, bool all_zeros = false);
 
     // read_lsn is the horizon H: serve the latest version <= H for [addr, addr+len), from the LBA index if applied
     // or from the journal-tail overlay if only Appended (no index write on the read path). Never fetches from a
@@ -194,7 +204,8 @@ class CraftReplDev {
     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.
+    // Seeds the Empty-verdict set and removes those LSNs from missing_lsns_ (resolving any gap they
+    // represented). Replaces any prior seeded empties. apply_sync_rs_commit_lsn (S5) must do the same.
     void seed_empty(std::initializer_list< int64_t > empty);
 #endif
 

src/lib/craft/tests/test_craft_peer_exchange.cpp

@@ -53,7 +53,12 @@ class MockCraftJournalBackend : public CraftJournalBackend {
     std::map< int64_t, JournalSlot > slots;
     std::optional< int64_t > fail_on_read; // if set, read_slot for that LSN returns io_error
 
-    async_status write_slot(int64_t, lba_t, lba_count_t, sisl::sg_list) override {
+    async_result< homestore::multi_blk_id > alloc_write_data(sisl::sg_list const& /* data */,
+                                                             lba_count_t /* len */) override {
+        co_return homestore::multi_blk_id{};
+    }
+
+    async_status write_slot(int64_t, lba_t, lba_count_t, homestore::multi_blk_id, bool) override {
         co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
     }
 
@@ -67,6 +72,7 @@ class MockCraftJournalBackend : public CraftJournalBackend {
     }
 
     async_status truncate_to(int64_t) override { co_return ok(); }
+    async_status free_data(homestore::multi_blk_id) override { co_return ok(); }
 };
 
 // ── test fixture ─────────────────────────────────────────────────────────────