Sdstor 22888

src/lib/craft/craft_repl_dev.cpp

@@ -20,6 +20,7 @@
 #include <cstring>
 
 #include <homestore/blkdata_service.hpp>    // data_service(), async_alloc_write, blk_alloc_hints
+#include <homestore/checkpoint/cp_mgr.hpp>  // cp_mgr(), CPManager::trigger_cp_flush()
 #include <homestore/logstore/log_store.hpp> // home_log_store, logstore_seq_num_t, log_write_comp_cb_t
 #include <iomgr/iomgr.hpp>                  // iomanager singleton, reactor_regex
 #include <sisl/async/value_awaitable.hpp>   // value_awaitable<T>: lock-free completion-before-suspend-safe bridge
@@ -202,6 +203,15 @@ class HomeStoreCraftJournalBackend : public CraftJournalBackend {
         }
         CraftJournalEntry hdr{};
         std::memcpy(&hdr, buf.bytes(), sizeof(CraftJournalEntry));
+        if (hdr.magic != k_journal_magic || hdr.version != k_journal_version) {
+            LOGE("free_slot: corrupt or foreign entry lsn={} magic={:#x} version={} -- refusing to free", lsn,
+                 hdr.magic, hdr.version);
+            co_return std::unexpected(std::make_error_condition(std::errc::io_error));
+        }
+        if (hdr.lsn != lsn) {
+            LOGE("free_slot: lsn mismatch requested={} stored={} -- refusing to free", lsn, hdr.lsn);
+            co_return std::unexpected(std::make_error_condition(std::errc::io_error));
+        }
         if (hdr.all_zeros) co_return ok();
 
         homestore::multi_blk_id blkid{};
@@ -221,6 +231,23 @@ unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore::
     return std::make_unique< HomeStoreCraftJournalBackend >(std::move(logstore), vol_ordinal);
 }
 
+// ─── HomeStoreCraftCheckpointTrigger (SDSTOR-22888) ──────────────────────────
+//
+// Thin wrapper over homestore::cp_mgr(). One instance is shared by every volume's CraftReplDev.
+
+class HomeStoreCraftCheckpointTrigger : public CraftCheckpointTrigger {
+public:
+    async_status trigger_cp_flush(bool force) override {
+        if (!co_await homestore::cp_mgr().trigger_cp_flush(force))
+            co_return std::unexpected(make_error_condition(volume_error::INTERNAL_ERROR));
+        co_return ok();
+    }
+};
+
+unique< CraftCheckpointTrigger > make_homestore_checkpoint_trigger() {
+    return std::make_unique< HomeStoreCraftCheckpointTrigger >();
+}
+
 // ─── constructor ──────────────────────────────────────────────────────────────
 
 CraftReplDev::CraftReplDev(volume_id_t vol_id, unique< CraftJournalBackend > journal) :
@@ -450,6 +477,9 @@ async_result< craft::read_result > CraftReplDev::read(craft::client_hdr /* hdr *
     co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
 }
 
+// TODO(SDSTOR-22733): once implemented, this is the other commit_lsn-advance path SDSTOR-22888's
+// checkpoint trigger needs to cover (see apply_sync_rs_commit_lsn's own hook) -- same
+// checkpoint_lsn_interval_/last_checkpoint_lsn_ bookkeeping under missing_mu_, same force=false.
 async_result< craft::lsn_pair > CraftReplDev::keep_alive(craft::client_hdr /* hdr */) {
     LOGW("CraftReplDev::keep_alive not yet implemented");
     co_return std::unexpected(std::make_error_condition(std::errc::not_supported));
@@ -624,7 +654,7 @@ async_status CraftReplDev::apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint6
         }
     }
 
-    std::vector< int64_t > to_free;
+    std::unordered_set< int64_t > to_free;
     std::vector< int64_t > to_fetch;
     uint64_t term;
     {
@@ -640,7 +670,8 @@ async_status CraftReplDev::apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint6
         term = state_.term;
 
         for (int64_t lsn : empty_slots) {
-            if (missing_lsns_.erase(lsn)) { to_free.push_back(lsn); }
+            bool const was_missing = missing_lsns_.erase(lsn) > 0;
+            if (!was_missing && lsn <= state_.last_append_lsn && !empty_lsns_.contains(lsn)) { to_free.insert(lsn); }
         }
         empty_lsns_.insert(empty_slots.begin(), empty_slots.end());
 
@@ -703,7 +734,10 @@ async_status CraftReplDev::apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint6
                     blkid_allocated = true;
                 }
 
-                // FIXME: We need to address the case when blkid is not set. How would write_slot handle that?
+                // FIXME: write_slot has no all_zeros branch -- it serializes whatever blkid it's given
+                // relying on multi_blk_id's own serialize()/serialized_size() to degrade safely for a
+                // default instance. That's an implicit, undocumented dependency on HomeStore's current
+                // behavior -- see SDSTOR-25613.
                 auto res = co_await journal_->write_slot(slot.lsn, term, slot.lba_off_bytes, slot.len_bytes, blkid,
                                                          slot.all_zeros);
                 if (!res) {
@@ -733,13 +767,46 @@ async_status CraftReplDev::apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint6
     // KNOWN GAP: this can land late. Because on_commit detaches this coroutine (see the FIXME there),
     // a later-committed entry (InternalLogin, or another SyncRSCommitLSN) may have already applied by
     // the time this advance actually runs, breaking strict RAFT apply ordering.
+    int64_t commit_lsn_snapshot;
+    bool should_checkpoint = false;
     {
         std::lock_guard lk{missing_mu_};
         int64_t next = state_.commit_lsn + 1;
         while (next <= rs_commit_lsn && !missing_lsns_.contains(next)) {
             state_.commit_lsn = next; // resolved (present or Empty) -- Empty is skipped, not gated on
             ++next;
         }
+        commit_lsn_snapshot = state_.commit_lsn;
+        // SDSTOR-22888: nudge HomeStore to checkpoint proactively rather than waiting on its own
+        // timer, so the journal-reclaim / RAFT-log-compaction floor (docs/craft/subtasks.md's S8)
+        // doesn't lag arbitrarily far behind commit_lsn. Interval reuses sync_rs_commit_lsn_interval
+        // (via checkpoint_lsn_interval_) rather than its own knob -- ties checkpoint cadence to the
+        // periodic SyncRSCommitLSN cadence. last_checkpoint_lsn_ is updated right here, before the
+        // lock is released  so that two overlapping apply_sync_rs_commit_lsn calls can't both read
+        // the same stale last_checkpoint_lsn_ and both decide to fire.
+        if (commit_lsn_snapshot - last_checkpoint_lsn_ >= checkpoint_lsn_interval_) {
+            last_checkpoint_lsn_ = commit_lsn_snapshot;
+            should_checkpoint = true;
+        }
+    }
+    if (should_checkpoint) {
+        // force=false: let this coalesce with any checkpoint already in flight rather than forcing
+        // back-to-back flushes under high commit throughput (see CraftCheckpointTrigger's doc
+        // comment). Detached (fire-and-forget) -- same pattern as the free_data cleanup above:
+        // nothing here depends on the flush completing. A failure is logged, not propagated,
+        // same posture as catch-up/fetch failures elsewhere in this function.
+        if (checkpoint_trigger_ == nullptr) {
+            LOGW("apply_sync_rs_commit_lsn: commit_lsn={} crossed checkpoint interval but no "
+                 "checkpoint_trigger_ wired -- skipping",
+                 commit_lsn_snapshot);
+        } else {
+            detail::detach([self, commit_lsn_snapshot]() -> async_status {
+                if (auto cp = co_await self->checkpoint_trigger_->trigger_cp_flush(false); !cp)
+                    LOGE("apply_sync_rs_commit_lsn: checkpoint trigger failed at commit_lsn={}: {}",
+                         commit_lsn_snapshot, cp.error().message());
+                co_return ok();
+            }());
+        }
     }
     LOGT("apply_sync_rs_commit_lsn ok rs_commit_lsn={} client_token={}", rs_commit_lsn, client_token);
     co_return ok();

src/lib/craft/craft_repl_dev.hpp

@@ -123,6 +123,27 @@ class CraftPeerFetcher {
     virtual ~CraftPeerFetcher() = default;
 };
 
+// ─── CraftCheckpointTrigger ───────────────────────────────────────────────────
+//
+// Abstraction over HomeStore's checkpoint manager (homestore::cp_mgr().trigger_cp_flush()).
+// Injected into CraftReplDev so unit tests (which compile craft_repl_dev.cpp directly against a
+// mock journal backend, with no running HomeStore instance -- see test_craft_raft_entries.cpp) can
+// exercise the trigger without touching HomeStore. Production code passes
+// HomeStoreCraftCheckpointTrigger (defined in craft_repl_dev.cpp). Default (null) leaves the
+// trigger stubbed -- same posture as CraftPeerFetcher.
+
+class CraftCheckpointTrigger {
+public:
+    virtual async_status trigger_cp_flush(bool force) = 0;
+    virtual ~CraftCheckpointTrigger() = default;
+};
+
+// Factory that wraps homestore::cp_mgr(). One instance is shared by every volume's CraftReplDev
+// (there is exactly one CPManager per HomeStore instance), unlike make_homestore_journal_backend
+// which is per-volume -- so CraftReplDev takes this via a non-owning pointer (set_checkpoint_trigger),
+// not ownership at construction. Tests inject MockCraftCheckpointTrigger directly.
+unique< CraftCheckpointTrigger > make_homestore_checkpoint_trigger();
+
 // ─── CraftReplDev ─────────────────────────────────────────────────────────────
 //
 // One instance per CRAFT-mode volume. Implements the full CRAFT data plane
@@ -211,6 +232,11 @@ class CraftReplDev : public std::enable_shared_from_this< CraftReplDev > {
 
     // 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.
+    // FIXME(S4/S7): before dropping entries here, force a completed checkpoint --
+    // co_await checkpoint_trigger_->trigger_cp_flush(true). Once entries above/below lsn are gone, the journal is
+    // no longer a durable record of them; if HomeStore's checkpoint has only been requested and not yet
+    // completed, a crash in between loses that data. HomeStore's own IndexTable::destroy() hits the
+    // identical problem and force-flushes before removing its superblock for exactly this reason.
     async_status truncate(int64_t lsn);
 
     // Propose a SyncRSCommitLSN RAFT entry (called by watchdog or leader during login).
@@ -262,6 +288,17 @@ class CraftReplDev : public std::enable_shared_from_this< CraftReplDev > {
     // Production sets this from HB_DYNAMIC_CONFIG(peer_fetch_timeout_ms) after construction (S8/S9).
     void set_peer_fetch_timeout_ms(uint32_t ms) { peer_fetch_timeout_ms_ = ms; }
 
+    // Wires the HomeStore checkpoint trigger used by apply_sync_rs_commit_lsn's periodic checkpoint
+    // (SDSTOR-22888). One CraftCheckpointTrigger instance is shared by every volume's CraftReplDev;
+    // tests inject a mock.
+    void set_checkpoint_trigger(CraftCheckpointTrigger* t) { checkpoint_trigger_ = t; }
+
+    // Overrides the commit_lsn delta between checkpoint triggers (default matches
+    // sync_rs_commit_lsn_interval's own default of 128, tying checkpoint cadence to the periodic
+    // SyncRSCommitLSN cadence). Production sets this from HB_DYNAMIC_CONFIG(sync_rs_commit_lsn_interval)
+    // after construction, same pattern as set_peer_fetch_timeout_ms.
+    void set_checkpoint_lsn_interval(int64_t n) { checkpoint_lsn_interval_ = n; }
+
 #ifdef _PRERELEASE
     // Seeds partition watermarks and the missing set directly, bypassing write().
     // Only compiled when _PRERELEASE is defined; never present in production binaries.
@@ -352,13 +389,23 @@ class CraftReplDev : public std::enable_shared_from_this< CraftReplDev > {
     // insert one LSN at a time under missing_mu_, which is O(gap width) instead of O(log ranges).
     std::set< int64_t > missing_lsns_;         // gaps between commit_lsn and last_append_lsn
     std::unordered_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_
+    mutable std::mutex missing_mu_;            // guards state_, missing_lsns_, empty_lsns_, and last_checkpoint_lsn_
     bool login_in_progress_{false};
     std::mutex login_mu_;
     CraftRaftListener raft_listener_;
     CraftPeerFetcher* peer_fetcher_{nullptr};  // null until S9 wires CraftConnector
     uint32_t peer_fetch_timeout_ms_{5000};     // deadline for fetch_data; overridden via set_peer_fetch_timeout_ms()
     std::atomic< uint64_t > write_counter_{0}; // incremented per write(); triggers periodic SyncRSCommitLSN append
+
+    CraftCheckpointTrigger* checkpoint_trigger_{nullptr}; // null until production wiring; unit tests inject a mock
+    int64_t checkpoint_lsn_interval_{128};                // commit_lsn delta between checkpoint triggers; see
+                                                          // set_checkpoint_lsn_interval()
+    int64_t last_checkpoint_lsn_{-1}; // commit_lsn as of the last triggered checkpoint (guarded by missing_mu_)
+                                      // FIXME(S8/SDSTOR-22745): defaults to -1 in lockstep with state_.commit_lsn
+                                      // When S8 wires recovering commit_lsn from the journal/superblock on restart,
+                                      // seed this to the recovered commit_lsn too (not -1), or the first post-
+                                      // restart apply_sync_rs_commit_lsn will unconditionally fire a checkpoint
+                                      // regardless of how recently one actually happened before the crash.
 };
 
 } // namespace homeblocks

src/lib/craft/tests/test_craft_homestore_backend.cpp

@@ -22,11 +22,19 @@
 // backend directly rather than through CraftReplDev or a volume -- the narrowest test that still
 // runs the real completion path.
 //
+// Also exercises HomeStoreCraftCheckpointTrigger::trigger_cp_flush (SDSTOR-22888) against the REAL
+// homestore::cp_mgr() -- same rationale: MockCraftCheckpointTrigger (test_craft_raft_entries.cpp)
+// covers CraftReplDev's own gating logic, but the wrapper's factory -> cp_mgr().trigger_cp_flush()
+// -> async_status conversion chain had never been compiled and run against a live CPManager.
+//
 // Links the full homeblocks library (unlike the other craft tests, which compile
 // craft_repl_dev.cpp directly to avoid HomeStore bring-up) because a real home_log_store requires
 // a running HomeStore instance.
 
+#include <condition_variable>
 #include <cstring>
+#include <mutex>
+#include <vector>
 
 #include <gtest/gtest.h>
 #include <sisl/options/options.h>
@@ -110,6 +118,69 @@ TEST_F(CraftHomeStoreBackendTest, AllocWriteDataFailsCleanlyForUnregisteredOrdin
     ASSERT_FALSE(alloc_r.has_value());
 }
 
+// free_slot reads the raw entry back off the log store and validates magic/version/lsn before trusting it.
+TEST_F(CraftHomeStoreBackendTest, FreeSlotSucceedsForRealEntry) {
+    auto logstore = make_logstore();
+    ASSERT_TRUE(logstore != nullptr);
+    auto backend = make_homestore_journal_backend(logstore, /* vol_ordinal = */ 0);
+
+    auto w = homeblocks::detail::sync_get(backend->write_slot(/* lsn = */ 0, /* term = */ 1, /* lba = */ 0,
+                                                              /* len = */ 4096, homestore::multi_blk_id{},
+                                                              /* all_zeros = */ true));
+    ASSERT_TRUE(w.has_value());
+
+    auto r = homeblocks::detail::sync_get(backend->free_slot(0));
+    ASSERT_TRUE(r.has_value());
+}
+
+// Writes a raw blob directly to the log store (bypassing write_slot's serialization entirely) that
+// doesn't conform to CraftJournalEntry's magic/version -- simulates a corrupt or foreign record.
+// free_slot must reject it rather than misreading garbage bytes as a valid blkid.
+TEST_F(CraftHomeStoreBackendTest, FreeSlotRejectsCorruptEntry) {
+    auto logstore = make_logstore();
+    ASSERT_TRUE(logstore != nullptr);
+    auto backend = make_homestore_journal_backend(logstore, /* vol_ordinal = */ 0);
+
+    std::vector< uint8_t > garbage(64, 0xEE); // larger than sizeof(CraftJournalEntry); not its magic/version
+    sisl::io_blob raw_blob{garbage.data(), static_cast< uint32_t >(garbage.size()), /* is_aligned = */ false};
+
+    std::mutex mu;
+    std::condition_variable cv;
+    bool done = false;
+    logstore->write_async(/* seq_num = */ 0, raw_blob, nullptr,
+                          [&](homestore::logstore_seq_num_t, sisl::io_blob&, homestore::logdev_key, void*) {
+                              std::lock_guard< std::mutex > lk{mu};
+                              done = true;
+                              cv.notify_one();
+                          });
+    std::unique_lock< std::mutex > lk{mu};
+    cv.wait(lk, [&] { return done; });
+    lk.unlock();
+
+    auto r = homeblocks::detail::sync_get(backend->free_slot(0));
+    ASSERT_FALSE(r.has_value());
+}
+
+// force=false: the value apply_sync_rs_commit_lsn's periodic trigger actually passes today.
+TEST_F(CraftHomeStoreBackendTest, CheckpointTriggerFlushesRealCPManager) {
+    auto trigger = make_homestore_checkpoint_trigger();
+    ASSERT_TRUE(trigger != nullptr);
+
+    auto r = homeblocks::detail::sync_get(trigger->trigger_cp_flush(/* force = */ false));
+    ASSERT_TRUE(r.has_value());
+}
+
+// force=true: untested until now -- this is the value truncate()'s FIXME (craft_repl_dev.hpp) says
+// a future correctness-critical call site will need, but the passthrough itself had never been
+// exercised against the real cp_mgr() for either value.
+TEST_F(CraftHomeStoreBackendTest, CheckpointTriggerHonorsForceFlag) {
+    auto trigger = make_homestore_checkpoint_trigger();
+    ASSERT_TRUE(trigger != nullptr);
+
+    auto r = homeblocks::detail::sync_get(trigger->trigger_cp_flush(/* force = */ true));
+    ASSERT_TRUE(r.has_value());
+}
+
 int main(int argc, char* argv[]) {
     int parsed_argc = argc;
     char** orig_argv = argv;