diff --git a/conanfile.py b/conanfile.py index 5cc084b..52eac79 100644 --- a/conanfile.py +++ b/conanfile.py @@ -10,7 +10,7 @@ class HomeBlocksConan(ConanFile): name = "homeblocks" - version = "6.0.6" + version = "6.0.7" homepage = "https://github.com/eBay/HomeBlocks" description = "Block Store built on HomeStore" @@ -108,7 +108,7 @@ def build(self): cmake.configure() cmake.build() if not self.conf.get("tools.build:skip_test", default=False): - jobs = self.conf.get("tools.build:jobs", default=3) + jobs = self.conf.get("tools.build:jobs", default=4) env = Environment() env.define("CTEST_PARALLEL_LEVEL", str(jobs)) if self.options.get_safe("sanitize") == "thread": diff --git a/docs/craft/rpcs.md b/docs/craft/rpcs.md index 0f1700d..340e5bd 100644 --- a/docs/craft/rpcs.md +++ b/docs/craft/rpcs.md @@ -196,11 +196,15 @@ RAFT entry payload: { rs_commit_lsn: int64, client_token: uint64, empty_slots: [ Proposed by the leader via `CraftReplDev::append()` — triggered by login, the watchdog, the periodic checkpoint, or the client-requested **Resolve** RPC (#5). **Before proposing**, the leader resolves every unresolved slot ≤ `rs_commit_lsn`: fetch from any holder, or record an `Empty` verdict on -quorum-lacks evidence; it never proposes past an unresolved slot. On RAFT commit each replica: verify -the token, mark `empty_slots` as permanent no-op holes (discarding any local data there), fetch the -remaining missing slots from peers, then advance `commit_lsn`. Replicas never declare `Empty` -unilaterally. This is the primary recovery mechanism — it carries no write data, only the watermark -and verdicts. +quorum-lacks evidence; it never proposes past an unresolved slot. On RAFT commit each replica: mark +`empty_slots` as permanent no-op holes (discarding any local data there), fetch the remaining missing +slots from peers, then advance `commit_lsn` to the contiguous prefix bounded by `rs_commit_lsn` -- +skipping `Empty` slots but never past an unresolved `Missing` one. `client_token` is carried on the +entry but not checked against local state at apply time: `SyncRSCommitLSN` applies before the +`InternalLogin` that would establish it, so an equality-fence here would make login itself unreachable; +ordering plus the term fence on subsequent IO provide exclusivity instead. Replicas never declare +`Empty` unilaterally. This is the primary recovery mechanism — it carries no write data, only the +watermark and verdicts. --- diff --git a/docs/craft/subtasks.md b/docs/craft/subtasks.md index 7348bdf..db39529 100644 --- a/docs/craft/subtasks.md +++ b/docs/craft/subtasks.md @@ -132,8 +132,8 @@ and enforce single-writer exclusivity without data flowing through the RAFT log. **SyncRSCommitLSN:** - RAFT entry carries `{rs_commit_lsn, client_token, empty_slots[]}` - **Leader pre-resolution:** before proposing `N`, the leader resolves every unresolved slot ≤ `N`: fetch it from any holder, or record an `Empty` verdict on quorum-lacks evidence (leader counts itself; non-responders never count); it must not propose past an unresolved slot -- On apply: verify token; mark `empty_slots` Empty, **discarding any local data held there** (reconciliation); if behind, `fetch_data()` the remaining missing slots from peers; then `commit_lsn = rs_commit_lsn`. **Apply never truncates** and replicas **never declare Empty unilaterally** -- Peer catch-up (`CraftPeerFetcher::fetch_from_peer`) is **timeout-bounded**: every call passes `peer_fetch_timeout_ms` (`home_blks_config.fbs`, default 5000ms); a peer that misses the deadline is treated as a hard failure, same as any other fetch failure (best-effort — `commit_lsn` still advances, unresolved LSNs stay missing). The interface only carries the deadline; enforcing it against a real wire call is S9's (the transport's) +- On apply: `client_token` is carried on the entry but NOT checked against local state -- `SyncRSCommitLSN` applies before the `InternalLogin` that would establish it, so an equality-fence here would make login itself unreachable; ordering plus the term fence on subsequent IO provide exclusivity instead. Mark `empty_slots` Empty, **discarding any local data held there** (reconciliation); if behind, `fetch_data()` the remaining missing slots from peers; then advance `commit_lsn` to the contiguous prefix bounded by `rs_commit_lsn` (skipping `Empty` slots, never past an unresolved `Missing` one). **Apply never truncates** and replicas **never declare Empty unilaterally** +- Peer catch-up (`CraftPeerFetcher::fetch_from_peer`) is **timeout-bounded**: every call passes `peer_fetch_timeout_ms` (`home_blks_config.fbs`, default 5000ms); a peer that misses the deadline is treated as a hard failure, same as any other fetch failure (best-effort — unresolved LSNs stay missing and `commit_lsn` stalls just below the first one). The interface only carries the deadline; enforcing it against a real wire call is S9's (the transport's) - `append(sync_to, client_token)` proposes this entry via RAFT - Triggers: periodic every N LSNs (configurable via `home_blks_config.fbs`, default 128), watchdog, login, **client-requested (after a failed sub-quorum write)** — the client's `Resolve` RPC lands on `CraftReplDev::request_resolution(term, upto)`, which runs this same leader pre-resolution and returns the Empty verdicts ≤ `upto` (`craft::resolution_result`). The client broadcasts it to every member (it cannot know the leader mid-session); a follower returns `NOT_LEADER` diff --git a/src/include/homeblks/home_blocks.hpp b/src/include/homeblks/home_blocks.hpp index 0afcd98..31a247d 100644 --- a/src/include/homeblks/home_blocks.hpp +++ b/src/include/homeblks/home_blocks.hpp @@ -69,7 +69,7 @@ using volume_handle = std::shared_ptr< volume >; // 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, - EMPTY_SLOT); + EMPTY_SLOT, WRONG_TOKEN, INVALID_ENTRY); ENUM(volume_state, uint32_t, INIT, // created, not yet online diff --git a/src/lib/craft/craft_repl_dev.cpp b/src/lib/craft/craft_repl_dev.cpp index 013eef2..f8ff28b 100644 --- a/src/lib/craft/craft_repl_dev.cpp +++ b/src/lib/craft/craft_repl_dev.cpp @@ -14,6 +14,7 @@ *********************************************************************************/ #include "craft_repl_dev.hpp" +#include "../coro_helpers.hpp" #include #include @@ -23,9 +24,30 @@ #include // iomanager singleton, reactor_regex #include // value_awaitable: lock-free completion-before-suspend-safe bridge +#include +#include +#include + namespace homeblocks { // ─── Journal entry on-disk format ───────────────────────────────────────────── + +namespace { +// fetch_data's contract is one entry per requested LSN (never one that wasn't asked for, never +// repeated). Returns the first response LSN that violates it (unrequested or duplicated), or nullopt if +// every entry matches exactly one requested LSN. Erasing from `pending` as we go catches duplicates for +// free: a repeated lsn finds nothing left to erase the second time. +std::optional< int64_t > validate_fetch_response(std::vector< int64_t > const& requested, + std::vector< JournalSlot > const& response) { + std::unordered_set< int64_t > pending{requested.begin(), requested.end()}; + for (auto const& slot : response) { + if (pending.erase(slot.lsn) == 0) return slot.lsn; + } + return std::nullopt; +} +} // namespace + +// ─── HomeStore journal backend ──────────────────────────────────────────────── // // 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 @@ -162,6 +184,33 @@ class HomeStoreCraftJournalBackend : public CraftJournalBackend { co_return ok(); } + // Reads the raw local entry back off the log store -- the exact bytes write_slot wrote, + // header + serialized blkid -- and hands the blkid to free_data. Never goes through + // read_slot/JournalSlot: that type is wire-shared with craft::JournalSlot for peer fetch_data + // responses and deliberately carries no blkid (meaningless to a remote peer). + async_status free_slot(int64_t lsn) override { + homestore::log_buffer buf; + try { + buf = logstore_->read_sync(static_cast< homestore::logstore_seq_num_t >(lsn)); + } catch (std::exception const& e) { + LOGE("free_slot: read_sync failed lsn={}: {}", lsn, e.what()); + co_return std::unexpected(std::make_error_condition(std::errc::io_error)); + } + if (buf.size() < sizeof(CraftJournalEntry)) { + LOGE("free_slot: entry truncated lsn={} size={}", lsn, buf.size()); + co_return std::unexpected(std::make_error_condition(std::errc::io_error)); + } + CraftJournalEntry hdr{}; + std::memcpy(&hdr, buf.bytes(), sizeof(CraftJournalEntry)); + if (hdr.all_zeros) co_return ok(); + + homestore::multi_blk_id blkid{}; + blkid.deserialize(sisl::blob{buf.bytes() + sizeof(CraftJournalEntry), + buf.size() - static_cast< uint32_t >(sizeof(CraftJournalEntry))}, + true /* copy */); + co_return co_await free_data(blkid); + } + private: shared< homestore::home_log_store > logstore_; uint64_t vol_ordinal_; @@ -177,19 +226,10 @@ unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore:: 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 ──────────────────────────────────────────── +// ─── get_rs_commit_lsn ──────────────────────────────────────────── // 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 */) { - 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(uint64_t /* term */, bool /* is_login */) { craft::lsn_pair pair{}; { @@ -291,7 +331,9 @@ async_result< craft::lsn_pair > CraftReplDev::write(craft::client_hdr hdr, int64 { std::lock_guard lock{missing_mu_}; - // Term check is inside the lock: state_.term is mutated by apply_internal_login (S5) under the same mutex. + // state_.term is guarded by missing_mu_ like the rest of state_ -- read it under the same lock + // used for the gap-marking below rather than unlocked, now that apply_internal_login (22887) + // actually mutates it from the RAFT commit thread. 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)); @@ -436,39 +478,50 @@ async_status CraftReplDev::append(int64_t /* sync_to */, uint64_t /* client_toke // 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. +// The missing_mu_ lock is held only for the up-front classification pass below, dropped before any +// 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 for the whole batch. // // 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()); + enum class SlotKind { Empty, Present, Absent }; - for (int64_t lsn : lsns) { - enum class SlotKind { Empty, Present, Absent }; - SlotKind kind; - { - std::lock_guard lk{missing_mu_}; + std::vector< SlotKind > kinds; + kinds.reserve(lsns.size()); + { + std::lock_guard lk{missing_mu_}; + for (int64_t lsn : lsns) { if (empty_lsns_.contains(lsn)) { - kind = SlotKind::Empty; + kinds.push_back(SlotKind::Empty); } else if (lsn >= 0 && lsn <= state_.last_append_lsn && !missing_lsns_.contains(lsn)) { - kind = SlotKind::Present; + kinds.push_back(SlotKind::Present); } else { - kind = SlotKind::Absent; + kinds.push_back(SlotKind::Absent); } } + } + + std::vector< JournalSlot > result; + result.reserve(lsns.size()); - if (kind == SlotKind::Empty) { + for (size_t i = 0; i < lsns.size(); ++i) { + const int64_t lsn = lsns[i]; + switch (kinds[i]) { + case SlotKind::Empty: result.push_back(JournalSlot{.lsn = lsn, .is_empty = true}); - } else if (kind == SlotKind::Present) { + break; + case 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)); + break; + } + case SlotKind::Absent: + break; // omit from result (not-present-here) } - // Absent: omit from result (not-present-here) } co_return result; @@ -476,24 +529,238 @@ async_result< std::vector< JournalSlot > > CraftReplDev::fetch_data(std::vector< // ─── RAFT listener ──────────────────────────────────────────────────────────── -void CraftReplDev::CraftRaftListener::on_commit(int64_t lsn, sisl::blob const& /* header */, - sisl::blob const& /* key */, +void CraftReplDev::CraftRaftListener::on_commit(int64_t lsn, sisl::blob const& header, sisl::blob const& key, std::vector< homestore::multi_blk_id > const& /* blkids */, cintrusive< homestore::repl_req_ctx >& /* ctx */) { - // S5 will parse the entry type from `header` and dispatch to - // owner_->apply_sync_rs_commit_lsn() or owner_->apply_internal_login(). - LOGD("CraftRaftListener::on_commit lsn={} (entry dispatch not yet implemented)", lsn); + if (header.size() < sizeof(CraftEntryHeader)) { + LOGE("on_commit lsn={} header too small ({} bytes)", lsn, header.size()); + return; + } + const auto* entry_hdr = reinterpret_cast< const CraftEntryHeader* >(header.cbytes()); + + switch (entry_hdr->type) { + case CraftEntryType::SyncRSCommitLSN: { + if (key.size() < sizeof(SyncRSCommitLSNPayload)) { + LOGE("on_commit lsn={} SyncRSCommitLSN key too small ({} bytes)", lsn, key.size()); + return; + } + const auto* payload = reinterpret_cast< const SyncRSCommitLSNPayload* >(key.cbytes()); + auto empty_slots = parse_empty_slots(key); + if (!empty_slots) { + LOGE("on_commit lsn={} SyncRSCommitLSN malformed empty_slots", lsn); + return; + } + // apply_sync_rs_commit_lsn co_awaits peer fetch + journal writes; on_commit itself is a synchronous + // HomeStore callback, so fire-and-forget it. + // + // Lifetime: on_commit itself only touches the raw `owner_` pointer, which is safe since HomeStore + // never calls on_commit on a dead device. The DETACHED coroutine this dispatches into is a separate + // concern -- apply_sync_rs_commit_lsn opens with `auto self = shared_from_this()`, so the coroutine + // frame holds a strong reference across every co_await, keeping CraftReplDev alive even if every + // external owner (e.g. a volume-removal path) drops its shared_ptr mid-apply. Requires every + // CraftReplDev to be owned via shared_ptr + // + // FIXME: KNOWN GAP (not yet fixed): detaching here also breaks strict RAFT apply ordering. + // on_commit returns to HomeStore as soon as this coroutine hits its first co_await, so + // HomeStore can call on_commit for the NEXT committed entry -- a synchronous InternalLogin, or + // another detached SyncRSCommitLSN -- before this one's effects are fully applied. + // No individual field access races (missing_mu_ still guards every access), but replicas can end up + // applying entries in different effective orders depending on async completion timing, which + // violates the determinism RAFT relies on for replicas to converge. See the commit_lsn advance at + // the tail of apply_sync_rs_commit_lsn and the client_token overwrite in apply_internal_login for + // the two mutation points this exposes. Real fix: one per-device serialized apply queue that both + // entry types funnel through, processing one entry's full effect (including all its co_awaits) + // before starting the next -- not independent detached tasks. + detail::detach( + owner_->apply_sync_rs_commit_lsn(payload->rs_commit_lsn, payload->client_token, std::move(*empty_slots))); + break; + } + case CraftEntryType::InternalLogin: { + // Fixed-size payload, no variable trailing data (unlike SyncRSCommitLSN) -- exact-size check. + if (key.size() != sizeof(InternalLoginPayload)) { + LOGE("on_commit lsn={} InternalLogin key wrong size ({} bytes)", lsn, key.size()); + return; + } + const auto* login_payload = reinterpret_cast< const InternalLoginPayload* >(key.cbytes()); + // Pure in-memory state transition (no co_await) -- called directly, not detached. + owner_->apply_internal_login(login_payload->client_token, login_payload->term); + break; + } + default: + LOGE("on_commit lsn={} unrecognized CraftEntryType={}", lsn, static_cast< uint8_t >(entry_hdr->type)); + break; + } } // ─── RAFT apply helpers (S5 implements) ────────────────────────────────────── +// +// apply_sync_rs_commit_lsn (22886): empty_slots is range-checked against rs_commit_lsn first -- the only +// all-or-nothing gate on this apply. SyncRSCommitLSN verdicts are only ever defined for slots the leader +// pre-resolved up to rs_commit_lsn (S5). client_token is NOT checked against the current session. Past the +// range check, every step is best-effort +// forward progress: empty_slots are reconciled and the newly-spanned range is marked missing, catch-up +// attempts to fill in what it can from a peer, and last_append_lsn advances regardless of whether catch-up +// fully succeeded -- mirroring truncate()'s invariant that apply never reverts the watermark, only advances +// it. commit_lsn is different: it's the local contiguous prefix (CRAFT-Design), so it only advances up to +// the first still-unresolved Missing slot, skipping over Empty ones, even though rs_commit_lsn itself is a +// watermark the whole replica set already agreed on. A peer's fetch_data response gets its own +// all-or-nothing check (validate_fetch_response): unlike the range check above, this one can't gate the +// whole apply (gap marking and last_append_lsn already advanced by the time the response arrives), so a +// malformed response is instead treated exactly like a failed fetch + +async_status CraftReplDev::apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint64_t client_token, + std::vector< int64_t > empty_slots) { + // Lives in the coroutine frame across every co_await below -- see the lifetime comment at the + // on_commit call site (detail::detach) for why this is required. + auto self = shared_from_this(); + + // Validated before any state is touched -- an out-of-range verdict means the entry itself cannot be + // trusted, not that this one slot should be skipped, so it gates the entire apply. + for (int64_t lsn : empty_slots) { + if (lsn < 0 || lsn > rs_commit_lsn) { + LOGE("apply_sync_rs_commit_lsn: empty_slots lsn={} out of range [0, {}] -- rejecting entire apply", lsn, + rs_commit_lsn); + co_return std::unexpected(make_error_condition(volume_error::INVALID_ENTRY)); + } + } + + std::vector< int64_t > to_free; + std::vector< int64_t > to_fetch; + uint64_t term; + { + std::lock_guard lk{missing_mu_}; + // client_token is NOT gated against state_.client_token here. Per the login sequence (CRAFT-Design), + // SyncRSCommitLSN applies BEFORE InternalLogin (which sets state_.client_token), so an equality-fence + // here would veto the very entry that carries login's own Empty verdicts, + // and would also veto every post-restart watchdog SyncRSCommitLSN, since state_ is + // in-memory-only and client_token resets to 0 across a restart. craft_client's reference + // (MemCraftReplica::cold_apply_sync) discards the parameter outright for the same reason. + // Exclusivity comes from RAFT's commit ordering plus the term fence every other IO already + // checks (see apply_internal_login's header comment), not from an equality check here. + term = state_.term; + + for (int64_t lsn : empty_slots) { + if (missing_lsns_.erase(lsn)) { to_free.push_back(lsn); } + } + empty_lsns_.insert(empty_slots.begin(), empty_slots.end()); + + // Everything newly spanned by this advance that isn't Empty-verdicted is a gap until catch-up + // (below) resolves it -- same idiom write() uses for gaps opened by an out-of-order dlsn. + for (int64_t lsn = state_.last_append_lsn + 1; lsn <= rs_commit_lsn; ++lsn) { + if (!empty_lsns_.contains(lsn)) missing_lsns_.insert(lsn); + } + state_.last_append_lsn = std::max(state_.last_append_lsn, rs_commit_lsn); -void CraftReplDev::apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint64_t /* client_token */, - std::vector< int64_t > /* empty_slots */) { - LOGD("apply_sync_rs_commit_lsn rs_commit_lsn={} (not yet implemented)", rs_commit_lsn); + for (int64_t lsn : missing_lsns_) { + if (lsn <= rs_commit_lsn) to_fetch.push_back(lsn); + } + } + + if (!to_free.empty()) { + for (int64_t lsn : to_free) { + if (auto fr = co_await journal_->free_slot(lsn); !fr) { + LOGE("apply_sync_rs_commit_lsn: free_slot failed lsn={}: {} -- blocks may leak", lsn, + fr.error().message()); + } + } + } + + if (!to_fetch.empty()) { + if (peer_fetcher_ == nullptr) { + LOGW("apply_sync_rs_commit_lsn: {} lsn(s) missing but no peer_fetcher_ wired -- leaving as missing", + to_fetch.size()); + } else if (auto fetched = co_await peer_fetcher_->fetch_data(to_fetch, peer_fetch_timeout_ms_); !fetched) { + LOGE("apply_sync_rs_commit_lsn: fetch_data failed: {} -- leaving {} lsn(s) as missing", + fetched.error().message(), to_fetch.size()); + } else if (auto bad_lsn = validate_fetch_response(to_fetch, *fetched); bad_lsn) { + // fetch_data's contract is one entry per requested LSN (never one we didn't ask for, never + // repeated) -- any deviation means the response itself can't be trusted, so none of it is + // applied (same outcome as a fetch failure) rather than cherry-picking the entries that look + // fine from a peer that has already proven unreliable. + LOGE("apply_sync_rs_commit_lsn: peer response lsn={} not requested (or duplicated) -- rejecting " + "entire batch, leaving {} lsn(s) as missing", + *bad_lsn, to_fetch.size()); + } else { + for (auto& slot : *fetched) { + if (slot.is_empty) { + std::lock_guard lk{missing_mu_}; + empty_lsns_.insert(slot.lsn); + missing_lsns_.erase(slot.lsn); + continue; + } + // HS_DATA_LINKED, same as write(): allocate blocks and write the payload before + // journalling the block reference. all_zeros slots carry no data and skip alloc. + homestore::multi_blk_id blkid{}; + bool blkid_allocated = false; + if (!slot.all_zeros) { + auto alloc_res = co_await journal_->alloc_write_data(slot.data, slot.len_bytes); + if (!alloc_res) { + LOGE("apply_sync_rs_commit_lsn: alloc_write_data failed lsn={}: {} -- leaving as missing", + slot.lsn, alloc_res.error().message()); + continue; + } + blkid = *alloc_res; + blkid_allocated = true; + } + + // FIXME: We need to address the case when blkid is not set. How would write_slot handle that? + auto res = co_await journal_->write_slot(slot.lsn, term, slot.lba_off_bytes, slot.len_bytes, blkid, + slot.all_zeros); + if (!res) { + LOGE("apply_sync_rs_commit_lsn: write_slot failed lsn={}: {} -- leaving as missing", slot.lsn, + res.error().message()); + if (blkid_allocated) { + detail::detach([self, blkid, lsn = slot.lsn]() -> async_status { + if (auto fr = co_await self->journal_->free_data(blkid); !fr) + LOGE("apply_sync_rs_commit_lsn: free_data failed after write_slot failure lsn={}: {}", + lsn, fr.error().message()); + co_return ok(); + }()); + } + continue; + } + std::lock_guard lk{missing_mu_}; + missing_lsns_.erase(slot.lsn); + } + } + } + + // commit_lsn (CRAFT-Design) is the LOCAL CONTIGUOUS prefix, distinct from rs_commit_lsn (the + // replica-set-wide watermark RAFT already agreed on): it must skip over Empty slots but never + // advance past an unresolved Missing one, even if catch-up above left holes below rs_commit_lsn. + // Mirrors craft_client's reference MemCraftReplica::apply_up_to. + // + // 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. + { + 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; + } + } + LOGT("apply_sync_rs_commit_lsn ok rs_commit_lsn={} client_token={}", rs_commit_lsn, client_token); + co_return ok(); } +// ─── InternalLogin apply (S5 / SDSTOR-22887) ───────────────────────────────── +// +// Pure in-memory state transition -- no journal I/O, no peer fetch -- so this stays synchronous +// (unlike apply_sync_rs_commit_lsn) and on_commit calls it directly rather than via detail::detach(). +// "Enforce single-writer exclusivity" needs no explicit rejection here: every other RPC's term-fence +// check (STALE_TERM on mismatch) already does that. Overwriting state_.term is what invalidates any +// existing session -- a caller still presenting the old term is fenced out on its very next call. + void CraftReplDev::apply_internal_login(uint64_t client_token, uint64_t term) { - LOGD("apply_internal_login client_token={} term={} (not yet implemented)", client_token, term); + std::lock_guard lk{missing_mu_}; + state_.client_token = client_token; // opaque id, no ordering semantics -- plain overwrite + // term is RAFT-ordered in practice (the leader always proposes strictly increasing terms), but + // guard against regression the same way commit_lsn/last_append_lsn already do rather than trusting + // log order blindly. + state_.term = std::max(state_.term, term); + LOGD("apply_internal_login client_token={} term={}", client_token, state_.term); } } // namespace homeblocks diff --git a/src/lib/craft/craft_repl_dev.hpp b/src/lib/craft/craft_repl_dev.hpp index ca8a54f..f33fc0a 100644 --- a/src/lib/craft/craft_repl_dev.hpp +++ b/src/lib/craft/craft_repl_dev.hpp @@ -20,8 +20,10 @@ #include #include +#include #include #include +#include #include namespace homestore { @@ -84,6 +86,13 @@ class CraftJournalBackend { // 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; + // TODO: Need to revisit this if this func can be avoided + // Reads the already-committed local entry at lsn and, if it isn't all_zeros, frees the blkid + // it references via free_data. Local-only by design: unlike read_slot/JournalSlot (the + // wire-shared type used to answer a peer's fetch_data), a blkid has no meaning off this + // replica, so this never needs to leave the local backend. Used by apply_sync_rs_commit_lsn's + // to_free path to reclaim blocks under an entry a later SyncRSCommitLSN verdicts Empty. + virtual async_status free_slot(int64_t lsn) = 0; virtual ~CraftJournalBackend() = default; }; @@ -104,7 +113,13 @@ unique< CraftJournalBackend > make_homestore_journal_backend(shared< homestore:: class CraftPeerFetcher { public: virtual async_result< craft::lsn_pair > get_rs_commit_lsn(uint64_t term, bool is_login) = 0; - virtual async_result< std::vector< JournalSlot > > fetch_data(const std::vector< int64_t >& lsns) = 0; + // `timeout_ms` is the deadline this call must complete within (CraftReplDev passes + // peer_fetch_timeout_ms_, set from home_blks_config.fbs's peer_fetch_timeout_ms). A real transport + // (S9) must treat a missed deadline as a hard failure, same as an unreachable peer -- this interface + // only carries the contract; there's nothing to enforce yet since today's only implementations are + // direct function calls (production is unwired, tests call synchronously). + virtual async_result< std::vector< JournalSlot > > fetch_data(const std::vector< int64_t >& lsns, + uint32_t timeout_ms) = 0; virtual ~CraftPeerFetcher() = default; }; @@ -114,9 +129,22 @@ class CraftPeerFetcher { // (write, read, login, truncate, ...) on top of a HomeStore log store and // index. Non-CRAFT volumes are unaffected. -class CraftReplDev { -public: +class CraftReplDev : public std::enable_shared_from_this< CraftReplDev > { +#ifdef _PRERELEASE + // Lets test_craft_raft_entries.cpp call apply_sync_rs_commit_lsn (private) directly, so it can assert + // on the exact result rather than only on-commit's discarded fire-and-forget outcome. + friend class CraftRaftEntriesTest; +#endif + + // Private -- see create() below. shared_from_this() (used by apply_sync_rs_commit_lsn's detached + // coroutine) requires the object to already be owned by a shared_ptr, so construction is gated behind + // create() rather than exposed directly. explicit CraftReplDev(volume_id_t vol_id, unique< CraftJournalBackend > journal); + +public: + static shared< CraftReplDev > create(volume_id_t vol_id, unique< CraftJournalBackend > journal) { + return shared< CraftReplDev >(new CraftReplDev(vol_id, std::move(journal))); + } ~CraftReplDev() = default; // ── client-facing ────────────────────────────────────────────────────── @@ -173,9 +201,6 @@ class CraftReplDev { // ── internal / peer API (server-to-server; NEVER reachable over the client wire) ── - // Return {commit_lsn, last_append_lsn} for the local partition. - async_result< craft::lsn_pair > get_lsns(volume_id_t vol_id); - // Callee side of the GetRSCommitLSN broadcast -- matches craft::craft_peer::get_rs_commit_lsn's // shape (craft_client's include/craft/peer.hpp) so a future wire-decoded request has somewhere // to pass {term, is_login}. is_login=true is meant to quiesce prior-session writes before @@ -220,11 +245,23 @@ class CraftReplDev { std::lock_guard lk{missing_mu_}; return state_.commit_lsn; } + uint64_t client_token() const { + std::lock_guard lk{missing_mu_}; + return state_.client_token; + } + uint64_t term() const { + std::lock_guard lk{missing_mu_}; + return state_.term; + } // Wires the server-to-server peer channel used by apply_sync_rs_commit_lsn catch-up. // Called by CraftConnector (S9) after construction; tests inject a mock. void set_peer_fetcher(CraftPeerFetcher* f) { peer_fetcher_ = f; } + // Overrides the deadline passed to fetch_from_peer (default mirrors home_blks_config.fbs). + // 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; } + #ifdef _PRERELEASE // Seeds partition watermarks and the missing set directly, bypassing write(). // Only compiled when _PRERELEASE is defined; never present in production binaries. @@ -236,6 +273,9 @@ class CraftReplDev { void seed_empty(std::initializer_list< int64_t > empty); // Seeds the session term so tests can exercise write() with a non-zero term without a full login. void seed_term(uint64_t term); + // Exposes the RAFT listener so tests can drive on_commit() directly -- raft_listener_ has no other + // accessor (production wiring into HomeStore's repl_dev happens elsewhere). + homestore::repl_dev_listener& test_listener() { return raft_listener_; } #endif private: @@ -292,23 +332,32 @@ class CraftReplDev { void on_config_rollback(int64_t) override {} private: + // Back-pointer to the owning CraftReplDev -- raft_listener_ is a value member of CraftReplDev + // (see its declaration below), so this can never dangle CraftReplDev* owner_; }; - // Called from CraftRaftListener::on_commit after deserialising the entry type. - void apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint64_t client_token, std::vector< int64_t > empty_slots); + // Called from CraftRaftListener::on_commit after deserialising the entry type. Detached (fire-and-forget) + // from on_commit since that HomeStore callback is synchronous but catch-up here needs to co_await peer + // fetch + journal writes. + async_status apply_sync_rs_commit_lsn(int64_t rs_commit_lsn, uint64_t client_token, + std::vector< int64_t > empty_slots); void apply_internal_login(uint64_t client_token, uint64_t term); volume_id_t vol_id_; unique< CraftJournalBackend > journal_; CraftPartitionState state_; - std::set< int64_t > missing_lsns_; // gaps between commit_lsn and last_append_lsn - 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_ + // TODO: Can this be replaced with boost::icl::interval_set? Particularly helpful when a write + // comes in with a huge gap -- gap-fill loops (write(), apply_sync_rs_commit_lsn()) currently + // 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_ 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 }; diff --git a/src/lib/craft/tests/CMakeLists.txt b/src/lib/craft/tests/CMakeLists.txt index 43b293a..cb9644f 100644 --- a/src/lib/craft/tests/CMakeLists.txt +++ b/src/lib/craft/tests/CMakeLists.txt @@ -15,7 +15,7 @@ target_link_libraries(test_craft_truncate add_test(NAME CraftTruncateTest COMMAND test_craft_truncate) -# Unit tests for CraftReplDev::get_lsns(), get_rs_commit_lsn(), and fetch_data() (S6). +# Unit tests for CraftReplDev::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 @@ -80,3 +80,18 @@ target_link_libraries(test_craft_journal_slot_wire ) add_test(NAME CraftJournalSlotWireTest COMMAND test_craft_journal_slot_wire) + +# Unit tests for CraftReplDev::apply_sync_rs_commit_lsn and the on_commit SyncRSCommitLSN dispatch +# (S5: SDSTOR-22886). Same pattern: compile craft_repl_dev.cpp directly. +add_executable(test_craft_raft_entries) +target_sources(test_craft_raft_entries PRIVATE + test_craft_raft_entries.cpp + ../craft_repl_dev.cpp +) +target_compile_definitions(test_craft_raft_entries PRIVATE _PRERELEASE) +target_link_libraries(test_craft_raft_entries + ${COMMON_TEST_DEPS} + -rdynamic +) + +add_test(NAME CraftRaftEntriesTest COMMAND test_craft_raft_entries) diff --git a/src/lib/craft/tests/mock_journal_backend.hpp b/src/lib/craft/tests/mock_journal_backend.hpp new file mode 100644 index 0000000..fb1321b --- /dev/null +++ b/src/lib/craft/tests/mock_journal_backend.hpp @@ -0,0 +1,45 @@ +/********************************************************************************* + * 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. + * + *********************************************************************************/ +#pragma once + +#include "craft/craft_repl_dev.hpp" + +// Shared free_slot body for CraftJournalBackend test mocks that record write_slot calls into a +// std::map< int64_t, JournalSlot > slots member. test_craft_write.cpp, test_craft_raft_entries.cpp, +// and test_craft_peer_exchange.cpp all need identical behavior here: read back the recorded slot, +// skip it if all_zeros (nothing was ever allocated), otherwise delegate to the mock's own free_data +// so any per-test free_data call counters still fire. +namespace homeblocks { + +// Shared read_slot body for the same mocks: look up the recorded slot by lsn, or +// no_such_file_or_directory if write_slot was never called for it. +template < typename Mock > +async_result< JournalSlot > mock_read_slot(Mock& mock, int64_t lsn) { + auto it = mock.slots.find(lsn); + if (it == mock.slots.end()) + co_return std::unexpected(std::make_error_condition(std::errc::no_such_file_or_directory)); + co_return it->second; +} + +template < typename Mock > +async_status mock_free_slot(Mock& mock, int64_t lsn) { + auto it = mock.slots.find(lsn); + if (it == mock.slots.end()) + co_return std::unexpected(std::make_error_condition(std::errc::no_such_file_or_directory)); + if (it->second.all_zeros) co_return ok(); + co_return co_await mock.free_data(homestore::multi_blk_id{}); +} + +} // namespace homeblocks \ No newline at end of file diff --git a/src/lib/craft/tests/test_craft_peer_exchange.cpp b/src/lib/craft/tests/test_craft_peer_exchange.cpp index 11598ba..9ecb8bc 100644 --- a/src/lib/craft/tests/test_craft_peer_exchange.cpp +++ b/src/lib/craft/tests/test_craft_peer_exchange.cpp @@ -13,8 +13,7 @@ * *********************************************************************************/ -// Unit tests for CraftReplDev::get_lsns(), get_rs_commit_lsn(), and fetch_data() -// (S6: Peer Data Exchange APIs). +// Unit tests for CraftReplDev::get_rs_commit_lsn() and fetch_data() (S6: Peer Data Exchange APIs). // // fetch_data() implements a four-way response per slot: // present+data : slot in journal, all_zeros=false @@ -35,6 +34,7 @@ #include "craft/craft_repl_dev.hpp" #include "coro_helpers.hpp" +#include "mock_journal_backend.hpp" SISL_LOGGING_DEF(HOMEBLOCKS_LOG_MODS) SISL_LOGGING_INIT(HOMEBLOCKS_LOG_MODS) @@ -65,14 +65,13 @@ class MockCraftJournalBackend : public CraftJournalBackend { async_result< JournalSlot > read_slot(int64_t lsn) override { if (fail_on_read && *fail_on_read == lsn) co_return std::unexpected(std::make_error_condition(std::errc::io_error)); - auto it = slots.find(lsn); - if (it == slots.end()) - co_return std::unexpected(std::make_error_condition(std::errc::no_such_file_or_directory)); - co_return it->second; + co_return co_await mock_read_slot(*this, lsn); } async_status truncate_to(int64_t) override { co_return ok(); } async_status free_data(homestore::multi_blk_id) override { co_return ok(); } + + async_status free_slot(int64_t lsn) override { return mock_free_slot(*this, lsn); } }; // ── test fixture ───────────────────────────────────────────────────────────── @@ -82,10 +81,9 @@ class CraftPeerExchangeTest : public ::testing::Test { void SetUp() override { auto mock = std::make_unique< MockCraftJournalBackend >(); journal_ = mock.get(); - dev_ = std::make_unique< CraftReplDev >(volume_id_t{}, std::move(mock)); + dev_ = CraftReplDev::create(volume_id_t{}, std::move(mock)); } - auto do_get_lsns() { return homeblocks::detail::sync_get(dev_->get_lsns(volume_id_t{})); } auto do_get_rs_commit_lsn() { return homeblocks::detail::sync_get(dev_->get_rs_commit_lsn(0, false)); } auto do_fetch_data(std::vector< int64_t > lsns) { return homeblocks::detail::sync_get(dev_->fetch_data(std::move(lsns))); @@ -97,41 +95,29 @@ class CraftPeerExchangeTest : public ::testing::Test { } MockCraftJournalBackend* journal_{nullptr}; - std::unique_ptr< CraftReplDev > dev_; + std::shared_ptr< CraftReplDev > dev_; }; -// ── get_lsns / get_rs_commit_lsn ───────────────────────────────────────────── +// ── get_rs_commit_lsn ──────────────────────────────────────────────────────── // Fresh device: both watermarks default to -1 (uninitialized sentinel). -TEST_F(CraftPeerExchangeTest, GetLsnsDefaultState) { - auto r = do_get_lsns(); +TEST_F(CraftPeerExchangeTest, GetRsCommitLsnDefaultState) { + auto r = do_get_rs_commit_lsn(); ASSERT_TRUE(r.has_value()); EXPECT_EQ(r->commit_lsn, -1); EXPECT_EQ(r->last_append_lsn, -1); } -// After seeding, get_lsns reflects both watermarks correctly. -TEST_F(CraftPeerExchangeTest, GetLsnsAfterSeed) { +// After seeding, get_rs_commit_lsn reflects both watermarks correctly. +TEST_F(CraftPeerExchangeTest, GetRsCommitLsnAfterSeed) { dev_->seed_lsns(50, {30, 40}); dev_->seed_commit_lsn(25); - auto r = do_get_lsns(); + auto r = do_get_rs_commit_lsn(); ASSERT_TRUE(r.has_value()); EXPECT_EQ(r->commit_lsn, 25); EXPECT_EQ(r->last_append_lsn, 50); } -// get_rs_commit_lsn is an alias of get_lsns; both must return the same snapshot. -TEST_F(CraftPeerExchangeTest, GetRsCommitLsnMatchesGetLsns) { - dev_->seed_lsns(100, {}); - dev_->seed_commit_lsn(80); - auto lsns_r = do_get_lsns(); - auto rs_r = do_get_rs_commit_lsn(); - ASSERT_TRUE(lsns_r.has_value()); - ASSERT_TRUE(rs_r.has_value()); - EXPECT_EQ(lsns_r->commit_lsn, rs_r->commit_lsn); - EXPECT_EQ(lsns_r->last_append_lsn, rs_r->last_append_lsn); -} - // ── fetch_data ──────────────────────────────────────────────────────────────── // Requesting zero LSNs returns an empty result vector, not an error. diff --git a/src/lib/craft/tests/test_craft_raft_entries.cpp b/src/lib/craft/tests/test_craft_raft_entries.cpp new file mode 100644 index 0000000..afdde18 --- /dev/null +++ b/src/lib/craft/tests/test_craft_raft_entries.cpp @@ -0,0 +1,591 @@ +/********************************************************************************* + * 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. + * + *********************************************************************************/ + +// Unit tests for CraftReplDev's two RAFT entry applies -- SyncRSCommitLSN (S5 / SDSTOR-22886) and +// InternalLogin (S5 / SDSTOR-22887) -- and their on_commit dispatch. +// +// SyncRSCommitLSN tests verify: +// - client_token is carried on the entry for observability only -- it is NOT gated against local state, +// regardless of what InternalLogin has (or hasn't) established, since SyncRSCommitLSN applies before +// the InternalLogin that would set it (see the inline comment at the call site for why) +// - empty_slots are range-validated against rs_commit_lsn and reconciled into empty_lsns_/missing_lsns_ +// - commit_lsn/last_append_lsn advance directly when there's no gap to catch up on +// - fetch_data is invoked with exactly the missing LSNs when behind, and its response is persisted +// - a peer response naming an unrequested or duplicate LSN is rejected as a whole batch +// - catch-up is best-effort: a failed fetch, a failed write_slot, or no peer_fetcher_ at all leaves the +// affected LSN(s) missing; commit_lsn stalls just below the first unresolved Missing slot (Empty +// slots are skipped over), never advancing past it regardless of rs_commit_lsn +// - commit_lsn never decrements +// - on_commit parses a real serialized SyncRSCommitLSN entry and dispatches correctly (and rejects +// malformed header/key blobs without touching state) +// +// InternalLogin tests verify: +// - on_commit dispatches to apply_internal_login, which sets client_token/term +// - a wrong-size key (too short or too long) is rejected, state untouched +// - a second InternalLogin replaces client_token outright but never regresses term +// - once applied, write()'s term-fence check reflects the new term end-to-end +// - InternalLogin establishing a new client_token has no bearing on SyncRSCommitLSN applies -- old or +// new token, the apply proceeds the same either way +// +// This TU defines SISL_LOGGING_DEF for the homeblocks module because it compiles craft_repl_dev.cpp +// directly (same pattern as test_craft_truncate.cpp). + +#include +#include +#include +#include + +#include "craft/craft_repl_dev.hpp" +#include "coro_helpers.hpp" +#include "mock_journal_backend.hpp" + +SISL_LOGGING_DEF(HOMEBLOCKS_LOG_MODS) +SISL_LOGGING_INIT(HOMEBLOCKS_LOG_MODS) + +namespace homeblocks { +namespace { + +// ── journal mock ────────────────────────────────────────────────────────────── +// +// Backed by a std::map so tests can inspect exactly which LSNs got persisted during catch-up. +// fail_on_write optionally injects an I/O error for a specific LSN. + +class MockCraftJournalBackend : public CraftJournalBackend { +public: + std::map< int64_t, JournalSlot > slots; + std::optional< int64_t > fail_on_write; + + async_result< homestore::multi_blk_id > alloc_write_data(sisl::sg_list const&, lba_count_t) override { + co_return homestore::multi_blk_id{}; + } + + async_status write_slot(int64_t lsn, uint64_t /* term */, lba_t lba, lba_count_t len, + homestore::multi_blk_id /* blkid */, bool all_zeros) override { + if (fail_on_write && *fail_on_write == lsn) + co_return std::unexpected(std::make_error_condition(std::errc::io_error)); + slots[lsn] = JournalSlot{.lsn = lsn, .all_zeros = all_zeros, .lba_off_bytes = lba, .len_bytes = len}; + co_return ok(); + } + + async_result< JournalSlot > read_slot(int64_t lsn) override { return mock_read_slot(*this, lsn); } + + async_status truncate_to(int64_t) override { co_return ok(); } + + async_status free_data(homestore::multi_blk_id) override { co_return ok(); } + + async_status free_slot(int64_t lsn) override { return mock_free_slot(*this, lsn); } + + bool has_slot(int64_t lsn) const { return slots.count(lsn) > 0; } +}; + +// ── peer fetcher mock ───────────────────────────────────────────────────────── +// +// Records the LSN list it was last called with; returns a programmable response or an injected error. + +class MockCraftPeerFetcher : public CraftPeerFetcher { +public: + std::vector< int64_t > last_requested; + uint32_t last_timeout_ms{0}; + std::vector< JournalSlot > response; + bool should_fail{false}; + + async_result< craft::lsn_pair > get_rs_commit_lsn(uint64_t /* term */, bool /* is_login */) override { + co_return craft::lsn_pair{}; + } + + async_result< std::vector< JournalSlot > > fetch_data(const std::vector< int64_t >& lsns, + uint32_t timeout_ms) override { + last_requested = lsns; + last_timeout_ms = timeout_ms; + if (should_fail) co_return std::unexpected(std::make_error_condition(std::errc::io_error)); + co_return response; + } +}; + +// ── wire-format helpers for the on_commit dispatch tests ───────────────────── + +sisl::blob as_blob(std::vector< uint8_t >& buf) { return sisl::blob{buf.data(), static_cast< uint32_t >(buf.size())}; } + +std::vector< uint8_t > make_header(CraftEntryType type) { + std::vector< uint8_t > buf(sizeof(CraftEntryHeader)); + reinterpret_cast< CraftEntryHeader* >(buf.data())->type = type; + return buf; +} + +std::vector< uint8_t > make_sync_rs_commit_lsn_key(int64_t rs_commit_lsn, uint64_t client_token, + const std::vector< int64_t >& empty_slots) { + std::vector< uint8_t > buf(sync_rs_commit_lsn_key_size(empty_slots.size())); + serialize_sync_rs_commit_lsn(buf.data(), rs_commit_lsn, client_token, empty_slots); + return buf; +} + +std::vector< uint8_t > make_internal_login_key(uint64_t client_token, uint64_t term) { + std::vector< uint8_t > buf(sizeof(InternalLoginPayload)); + auto* p = reinterpret_cast< InternalLoginPayload* >(buf.data()); + p->client_token = client_token; + p->term = term; + return buf; +} + +} // namespace + +// craft_repl_dev.hpp friends this exact type (homeblocks::CraftRaftEntriesTest) so it can call the +// private apply_sync_rs_commit_lsn directly -- it must NOT sit in the anonymous namespace above, or it +// would be a distinct, unrelated type from the friend's perspective. + +// ── test fixture ───────────────────────────────────────────────────────────── + +class CraftRaftEntriesTest : public ::testing::Test { +protected: + void SetUp() override { + auto mock = std::make_unique< MockCraftJournalBackend >(); + journal_ = mock.get(); + dev_ = CraftReplDev::create(volume_id_t{}, std::move(mock)); + } + + auto do_apply(int64_t rs_commit_lsn, uint64_t client_token, std::vector< int64_t > empty_slots = {}) { + return homeblocks::detail::sync_get( + dev_->apply_sync_rs_commit_lsn(rs_commit_lsn, client_token, std::move(empty_slots))); + } + + MockCraftJournalBackend* journal_{nullptr}; + MockCraftPeerFetcher fetcher_; + std::shared_ptr< CraftReplDev > dev_; +}; + +namespace { + +// ── client_token is not gated ───────────────────────────────────────────────── + +// client_token is carried for observability only -- a mismatch must NOT block the apply. See the inline +// comment in apply_sync_rs_commit_lsn (under missing_mu_) for why: SyncRSCommitLSN applies before the +// InternalLogin that would establish state_.client_token, so an equality-fence here would make login +// itself unreachable. +TEST_F(CraftRaftEntriesTest, ClientTokenMismatchDoesNotBlockApply) { + dev_->seed_lsns(5, {3}); + auto r = do_apply(/*rs_commit_lsn=*/100, /*client_token=*/999); + + ASSERT_TRUE(r.has_value()); + EXPECT_EQ(dev_->last_append_lsn(), 100); + EXPECT_EQ(dev_->commit_lsn(), 2); // stalls at lsn=3, still missing -- no peer_fetcher_ wired +} + +// ── empty_slots range validation ────────────────────────────────────────────── + +// A negative LSN in empty_slots is nonsensical for a SyncRSCommitLSN verdict -- reject the whole apply, +// the same all-or-nothing gate as a token mismatch, before any state is touched. +TEST_F(CraftRaftEntriesTest, RejectsEmptySlotWithNegativeLSN) { + dev_->seed_lsns(5, {3}); + auto r = do_apply(/*rs_commit_lsn=*/10, /*client_token=*/0, /*empty_slots=*/{-1}); + + ASSERT_FALSE(r.has_value()); + EXPECT_EQ(r.error(), make_error_condition(volume_error::INVALID_ENTRY)); + EXPECT_EQ(dev_->commit_lsn(), -1); + EXPECT_EQ(dev_->last_append_lsn(), 5); + EXPECT_EQ(dev_->missing_count(), 1u); +} + +// An empty_slots entry above rs_commit_lsn names a slot the leader never pre-resolved (S5 only resolves +// up to the LSN it proposes) -- reject the whole apply rather than let it poison empty_lsns_ for a slot +// that hasn't even been reached yet. +TEST_F(CraftRaftEntriesTest, RejectsEmptySlotAboveRSCommitLSN) { + dev_->seed_lsns(5, {3}); + auto r = do_apply(/*rs_commit_lsn=*/5, /*client_token=*/0, /*empty_slots=*/{6}); + + ASSERT_FALSE(r.has_value()); + EXPECT_EQ(r.error(), make_error_condition(volume_error::INVALID_ENTRY)); + EXPECT_EQ(dev_->commit_lsn(), -1); + EXPECT_EQ(dev_->last_append_lsn(), 5); + EXPECT_EQ(dev_->missing_count(), 1u); +} + +// ── empty_slots reconciliation ──────────────────────────────────────────────── + +TEST_F(CraftRaftEntriesTest, EmptySlotsReconciled) { + dev_->seed_lsns(5, {3}); + auto r = do_apply(/*rs_commit_lsn=*/5, /*client_token=*/0, /*empty_slots=*/{3}); + + ASSERT_TRUE(r.has_value()); + EXPECT_TRUE(dev_->is_empty_slot(3)); + EXPECT_FALSE(dev_->is_missing(3)); + EXPECT_EQ(dev_->commit_lsn(), 5); +} + +// An empty_slots entry can also fall inside the range this same apply newly opens (rather than being +// an already-missing LSN from before) -- it must end up ONLY in empty_lsns_, not re-added to +// missing_lsns_ by the gap-marking step that runs right after reconciliation. +TEST_F(CraftRaftEntriesTest, EmptySlotWithinNewGapRangeNotDoubleTracked) { + dev_->seed_lsns(0, {}); + auto r = do_apply(/*rs_commit_lsn=*/5, /*client_token=*/0, /*empty_slots=*/{3}); + + ASSERT_TRUE(r.has_value()); + EXPECT_TRUE(dev_->is_empty_slot(3)); + EXPECT_FALSE(dev_->is_missing(3)); + // The rest of the newly-opened gap range (1, 2, 4, 5) is still missing -- no peer_fetcher_ wired. + EXPECT_TRUE(dev_->is_missing(1)); + EXPECT_TRUE(dev_->is_missing(2)); + EXPECT_TRUE(dev_->is_missing(4)); + EXPECT_TRUE(dev_->is_missing(5)); + EXPECT_EQ(dev_->missing_count(), 4u); + EXPECT_EQ(dev_->commit_lsn(), 0); // stalls at lsn=1, still missing -- no peer_fetcher_ wired +} + +// ── watermark advance ────────────────────────────────────────────────────────── + +// last_append_lsn already covers rs_commit_lsn: nothing to fetch, commit_lsn advances directly. +TEST_F(CraftRaftEntriesTest, NoGapAdvancesDirectly) { + dev_->set_peer_fetcher(&fetcher_); + dev_->seed_lsns(10, {}); + + auto r = do_apply(/*rs_commit_lsn=*/5, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_EQ(dev_->commit_lsn(), 5); + EXPECT_TRUE(fetcher_.last_requested.empty()); +} + +TEST_F(CraftRaftEntriesTest, CommitLsnNeverDecrements) { + dev_->seed_lsns(10, {}); + dev_->seed_commit_lsn(8); + + auto r = do_apply(/*rs_commit_lsn=*/5, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_EQ(dev_->commit_lsn(), 8); + EXPECT_EQ(dev_->last_append_lsn(), 10); +} + +// ── catch-up ─────────────────────────────────────────────────────────────────── + +// Behind rs_commit_lsn: fetch_data is called with exactly the missing LSNs, and its response +// (one present slot, one Empty slot) is persisted/marked correctly. +TEST_F(CraftRaftEntriesTest, BehindWithPeerFetcherAppliesFetchedSlots) { + dev_->set_peer_fetcher(&fetcher_); + dev_->seed_lsns(0, {}); + fetcher_.response = { + JournalSlot{.lsn = 1, .lba_off_bytes = 10, .len_bytes = 4}, + JournalSlot{.lsn = 2, .is_empty = true}, + }; + + auto r = do_apply(/*rs_commit_lsn=*/2, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_EQ(fetcher_.last_requested, (std::vector< int64_t >{1, 2})); + EXPECT_TRUE(journal_->has_slot(1)); + EXPECT_TRUE(dev_->is_empty_slot(2)); + EXPECT_FALSE(dev_->is_missing(1)); + EXPECT_FALSE(dev_->is_missing(2)); + EXPECT_EQ(dev_->commit_lsn(), 2); + EXPECT_EQ(dev_->last_append_lsn(), 2); +} + +// set_peer_fetch_timeout_ms() threads the configured deadline through to fetch_data verbatim. +TEST_F(CraftRaftEntriesTest, BehindPassesConfiguredTimeoutToPeerFetcher) { + dev_->set_peer_fetcher(&fetcher_); + dev_->set_peer_fetch_timeout_ms(1234); + dev_->seed_lsns(0, {}); + fetcher_.response = {JournalSlot{.lsn = 1, .lba_off_bytes = 10, .len_bytes = 4}}; + + auto r = do_apply(/*rs_commit_lsn=*/1, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_EQ(fetcher_.last_timeout_ms, 1234u); +} + +// fetch_data's contract is one entry per requested LSN. A response naming an LSN we never asked for (a +// buggy/misbehaving peer) can't be partially trusted -- since gap-marking already advanced by this point +// in the apply, this can't gate the whole apply the way the upfront empty_slots check does, but it CAN +// still refuse the batch: none of the response is applied (same outcome as a fetch failure), and commit_lsn +// stalls at the first still-missing lsn (best-effort forward progress, not a jump to rs_commit_lsn). +TEST_F(CraftRaftEntriesTest, BehindRejectsPeerResponseWithUnrequestedLSN) { + dev_->set_peer_fetcher(&fetcher_); + dev_->seed_lsns(0, {}); + fetcher_.response = { + JournalSlot{.lsn = 1, .lba_off_bytes = 10, .len_bytes = 4}, JournalSlot{.lsn = 2, .is_empty = true}, + JournalSlot{.lsn = 99, .is_empty = true}, // never requested -- only 1 and 2 were + }; + + auto r = do_apply(/*rs_commit_lsn=*/2, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_FALSE(journal_->has_slot(1)); + EXPECT_FALSE(dev_->is_empty_slot(2)); + EXPECT_FALSE(dev_->is_empty_slot(99)); + EXPECT_EQ(dev_->missing_count(), 2u); // 1 and 2 both remain missing + EXPECT_EQ(dev_->commit_lsn(), 0); // stalls at lsn=1, still missing +} + +// A duplicate entry for an actually-requested LSN is just as much a contract violation as an +// unrequested one (validate_fetch_response catches both the same way) -- same whole-batch rejection. +TEST_F(CraftRaftEntriesTest, BehindRejectsPeerResponseWithDuplicateLSN) { + dev_->set_peer_fetcher(&fetcher_); + dev_->seed_lsns(0, {}); + fetcher_.response = { + JournalSlot{.lsn = 1, .lba_off_bytes = 10, .len_bytes = 4}, + JournalSlot{.lsn = 1, .lba_off_bytes = 20, .len_bytes = 4}, // duplicate + }; + + auto r = do_apply(/*rs_commit_lsn=*/1, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_FALSE(journal_->has_slot(1)); + EXPECT_EQ(dev_->missing_count(), 1u); + EXPECT_EQ(dev_->commit_lsn(), 0); // stalls at lsn=1, still missing +} + +// fetch_data fails outright: commit_lsn stalls at the first missing lsn (best-effort, not a jump to +// rs_commit_lsn); every spanned LSN remains missing. +TEST_F(CraftRaftEntriesTest, BehindFetchFailsCommitLsnStallsAtFirstMissing) { + dev_->set_peer_fetcher(&fetcher_); + dev_->seed_lsns(0, {}); + fetcher_.should_fail = true; + + auto r = do_apply(/*rs_commit_lsn=*/3, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_EQ(dev_->commit_lsn(), 0); // stalls at lsn=1, still missing + EXPECT_EQ(dev_->last_append_lsn(), 3); + EXPECT_EQ(dev_->missing_count(), 3u); +} + +// No peer_fetcher_ wired at all (S9 not wired yet): same best-effort outcome as a fetch failure. +TEST_F(CraftRaftEntriesTest, BehindNoPeerFetcherCommitLsnStallsAtFirstMissing) { + dev_->seed_lsns(0, {}); + + auto r = do_apply(/*rs_commit_lsn=*/2, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_EQ(dev_->commit_lsn(), 0); // stalls at lsn=1, still missing + EXPECT_EQ(dev_->missing_count(), 2u); +} + +// A fetched slot's write_slot fails: that LSN alone stays missing; the rest of catch-up still applies, +// and commit_lsn advances up to (but not past) it. +TEST_F(CraftRaftEntriesTest, WriteSlotFailureDuringCatchupLeavesLsnMissing) { + dev_->set_peer_fetcher(&fetcher_); + dev_->seed_lsns(0, {}); + fetcher_.response = { + JournalSlot{.lsn = 1, .lba_off_bytes = 1, .len_bytes = 4}, + JournalSlot{.lsn = 2, .lba_off_bytes = 2, .len_bytes = 4}, + }; + journal_->fail_on_write = 2; + + auto r = do_apply(/*rs_commit_lsn=*/2, /*client_token=*/0); + + ASSERT_TRUE(r.has_value()); + EXPECT_TRUE(journal_->has_slot(1)); + EXPECT_FALSE(journal_->has_slot(2)); + EXPECT_FALSE(dev_->is_missing(1)); + EXPECT_TRUE(dev_->is_missing(2)); + EXPECT_EQ(dev_->commit_lsn(), 1); // lsn=1 resolved; stalls at lsn=2, still missing +} + +// ── on_commit dispatch ───────────────────────────────────────────────────────── + +TEST_F(CraftRaftEntriesTest, OnCommitDispatchesSyncRSCommitLSN) { + auto header_buf = make_header(CraftEntryType::SyncRSCommitLSN); + auto key_buf = make_sync_rs_commit_lsn_key(/*rs_commit_lsn=*/7, /*client_token=*/0, /*empty_slots=*/{}); + cintrusive< homestore::repl_req_ctx > ctx{}; + + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key_buf), {}, ctx); + + // last_append_lsn advances unconditionally + EXPECT_EQ(dev_->last_append_lsn(), 7); +} + +TEST_F(CraftRaftEntriesTest, OnCommitRejectsHeaderTooSmall) { + std::vector< uint8_t > empty_header; + auto key_buf = make_sync_rs_commit_lsn_key(7, 0, {}); + cintrusive< homestore::repl_req_ctx > ctx{}; + + dev_->test_listener().on_commit(1, as_blob(empty_header), as_blob(key_buf), {}, ctx); + + EXPECT_EQ(dev_->commit_lsn(), -1); // untouched +} + +TEST_F(CraftRaftEntriesTest, OnCommitRejectsMalformedSyncRSCommitLSNKey) { + auto header_buf = make_header(CraftEntryType::SyncRSCommitLSN); + std::vector< uint8_t > short_key(sizeof(SyncRSCommitLSNPayload) - 1, 0); + cintrusive< homestore::repl_req_ctx > ctx{}; + + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(short_key), {}, ctx); + + EXPECT_EQ(dev_->commit_lsn(), -1); // untouched +} + +// Distinct from the too-short case above: this key is large enough for the fixed prefix (and even +// carries 2 real trailing slots), but lies about how many follow -- parse_empty_slots's exact-size +// check (not on_commit's coarser size check) is what rejects it. +TEST_F(CraftRaftEntriesTest, OnCommitRejectsMismatchedEmptySlotsCount) { + auto header_buf = make_header(CraftEntryType::SyncRSCommitLSN); + auto key_buf = make_sync_rs_commit_lsn_key(7, 0, {10, 20}); + reinterpret_cast< SyncRSCommitLSNPayload* >(key_buf.data())->num_empty_slots = 5; + cintrusive< homestore::repl_req_ctx > ctx{}; + + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key_buf), {}, ctx); + + EXPECT_EQ(dev_->commit_lsn(), -1); // untouched +} + +TEST_F(CraftRaftEntriesTest, OnCommitIgnoresUnrecognizedEntryType) { + std::vector< uint8_t > header_buf(sizeof(CraftEntryHeader)); + reinterpret_cast< CraftEntryHeader* >(header_buf.data())->type = static_cast< CraftEntryType >(99); + std::vector< uint8_t > key_buf; + cintrusive< homestore::repl_req_ctx > ctx{}; + + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key_buf), {}, ctx); + + EXPECT_EQ(dev_->commit_lsn(), -1); // untouched +} + +// ── InternalLogin apply (SDSTOR-22887) ──────────────────────────────────────── + +TEST_F(CraftRaftEntriesTest, OnCommitDispatchesInternalLogin) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + auto key_buf = make_internal_login_key(/*client_token=*/42, /*term=*/5); + cintrusive< homestore::repl_req_ctx > ctx{}; + + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key_buf), {}, ctx); + + EXPECT_EQ(dev_->client_token(), 42u); + EXPECT_EQ(dev_->term(), 5u); +} + +TEST_F(CraftRaftEntriesTest, OnCommitRejectsInternalLoginWrongSize) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + std::vector< uint8_t > short_key(sizeof(InternalLoginPayload) - 1, 0); + cintrusive< homestore::repl_req_ctx > ctx{}; + + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(short_key), {}, ctx); + + EXPECT_EQ(dev_->client_token(), 0u); // untouched + EXPECT_EQ(dev_->term(), 0u); // untouched +} + +// Unlike SyncRSCommitLSN's coarser "at least the fixed prefix" check (it has variable trailing data), +// InternalLoginPayload never does -- on_commit's check is an exact-size `!=`, so a key that's too LARGE +// must be rejected just as much as one that's too small. +TEST_F(CraftRaftEntriesTest, OnCommitRejectsInternalLoginKeyTooLarge) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + std::vector< uint8_t > long_key(sizeof(InternalLoginPayload) + 1, 0); + cintrusive< homestore::repl_req_ctx > ctx{}; + + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(long_key), {}, ctx); + + EXPECT_EQ(dev_->client_token(), 0u); // untouched + EXPECT_EQ(dev_->term(), 0u); // untouched +} + +// "A second InternalLogin invalidates any existing session before establishing the new one" (ticket) -- +// the second apply's values must win outright, not merge with the first's. Driven through on_commit +// (apply_internal_login itself is private, and TEST_F bodies live in a class derived from +// CraftRaftEntriesTest -- friendship doesn't propagate to it, so the public dispatch path is used here). +TEST_F(CraftRaftEntriesTest, SecondInternalLoginReplacesSession) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + cintrusive< homestore::repl_req_ctx > ctx{}; + + auto key1 = make_internal_login_key(/*client_token=*/1, /*term=*/1); + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key1), {}, ctx); + auto key2 = make_internal_login_key(/*client_token=*/2, /*term=*/2); + dev_->test_listener().on_commit(2, as_blob(header_buf), as_blob(key2), {}, ctx); + + EXPECT_EQ(dev_->client_token(), 2u); + EXPECT_EQ(dev_->term(), 2u); +} + +TEST_F(CraftRaftEntriesTest, InternalLoginTermNeverRegresses) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + cintrusive< homestore::repl_req_ctx > ctx{}; + + auto key1 = make_internal_login_key(/*client_token=*/1, /*term=*/5); + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key1), {}, ctx); + auto key2 = make_internal_login_key(/*client_token=*/2, /*term=*/3); + dev_->test_listener().on_commit(2, as_blob(header_buf), as_blob(key2), {}, ctx); + + EXPECT_EQ(dev_->term(), 5u); +} + +// client_token has no ordering semantics (it's an opaque id, unlike term) -- it's a plain overwrite even +// when the accompanying term regresses and is guarded. Pins down the intentional asymmetry between the +// two fields so it doesn't read as an oversight to a future reader. +TEST_F(CraftRaftEntriesTest, InternalLoginClientTokenOverwrittenEvenWhenTermRegresses) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + cintrusive< homestore::repl_req_ctx > ctx{}; + + auto key1 = make_internal_login_key(/*client_token=*/1, /*term=*/5); + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key1), {}, ctx); + auto key2 = make_internal_login_key(/*client_token=*/99, /*term=*/3); // lower term, different token + dev_->test_listener().on_commit(2, as_blob(header_buf), as_blob(key2), {}, ctx); + + EXPECT_EQ(dev_->term(), 5u); // guarded against regression + EXPECT_EQ(dev_->client_token(), 99u); // overwritten regardless +} + +// Happy-path regression check for the write()-term-check-under-lock reorg: a matching term still +// succeeds normally (only the mismatch path changed). +TEST_F(CraftRaftEntriesTest, WriteSucceedsWithMatchingTermAfterInternalLogin) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + auto key_buf = make_internal_login_key(/*client_token=*/1, /*term=*/5); + cintrusive< homestore::repl_req_ctx > ctx{}; + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key_buf), {}, ctx); + + auto r = homeblocks::detail::sync_get( + dev_->write(craft::client_hdr{.term = 5, .commit_lsn = -1, .all_committed_lsn = -1}, /*dlsn=*/1, + /*addr=*/0, /*len=*/0, {}, /*all_zeros=*/true)); + + ASSERT_TRUE(r.has_value()); +} + +// End-to-end: once InternalLogin moves state_.term forward, a write() still presenting the old term is +// fenced out on its very next call -- the "invalidation" this ticket calls for, and the regression test +// for write()'s term-check-under-lock fix. +TEST_F(CraftRaftEntriesTest, WriteRejectsStaleTermAfterInternalLogin) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + auto key_buf = make_internal_login_key(/*client_token=*/1, /*term=*/5); + cintrusive< homestore::repl_req_ctx > ctx{}; + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key_buf), {}, ctx); + + auto r = homeblocks::detail::sync_get( + dev_->write(craft::client_hdr{.term = 4, .commit_lsn = -1, .all_committed_lsn = -1}, /*dlsn=*/1, + /*addr=*/0, /*len=*/0, {}, /*all_zeros=*/true)); + + ASSERT_FALSE(r.has_value()); + EXPECT_EQ(r.error(), make_error_condition(volume_error::STALE_TERM)); +} + +// InternalLogin establishing a new client_token has no bearing on SyncRSCommitLSN applies -- neither the +// stale/old token nor the newly-established one gates the apply; both succeed identically. Guards against +// reintroducing an equality-fence keyed off InternalLogin's client_token. +TEST_F(CraftRaftEntriesTest, SyncRSCommitLSNAppliesRegardlessOfInternalLoginToken) { + auto header_buf = make_header(CraftEntryType::InternalLogin); + auto key_buf = make_internal_login_key(/*client_token=*/7, /*term=*/1); + cintrusive< homestore::repl_req_ctx > ctx{}; + dev_->test_listener().on_commit(1, as_blob(header_buf), as_blob(key_buf), {}, ctx); + + auto stale = do_apply(/*rs_commit_lsn=*/5, /*client_token=*/0); // the old default -- would have been a mismatch + ASSERT_TRUE(stale.has_value()); + + auto fresh = do_apply(/*rs_commit_lsn=*/5, /*client_token=*/7); // the token InternalLogin just set + ASSERT_TRUE(fresh.has_value()); + EXPECT_EQ(dev_->last_append_lsn(), 5); +} + +} // namespace +} // namespace homeblocks + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/src/lib/craft/tests/test_craft_truncate.cpp b/src/lib/craft/tests/test_craft_truncate.cpp index 894f242..1d4e67b 100644 --- a/src/lib/craft/tests/test_craft_truncate.cpp +++ b/src/lib/craft/tests/test_craft_truncate.cpp @@ -64,6 +64,10 @@ class MockCraftJournalBackend : public CraftJournalBackend { co_return ok(); } async_status free_data(homestore::multi_blk_id) override { co_return ok(); } + + async_status free_slot(int64_t) override { + co_return std::unexpected(std::make_error_condition(std::errc::not_supported)); + } }; // ── test fixture ───────────────────────────────────────────────────────────── @@ -73,13 +77,13 @@ class CraftTruncateTest : public ::testing::Test { void SetUp() override { auto mock = std::make_unique< MockCraftJournalBackend >(); journal_ = mock.get(); - dev_ = std::make_unique< CraftReplDev >(volume_id_t{}, std::move(mock)); + dev_ = CraftReplDev::create(volume_id_t{}, std::move(mock)); } auto do_truncate(int64_t lsn) { return homeblocks::detail::sync_get(dev_->truncate(lsn)); } MockCraftJournalBackend* journal_{nullptr}; - std::unique_ptr< CraftReplDev > dev_; + std::shared_ptr< CraftReplDev > dev_; }; // ── tests ───────────────────────────────────────────────────────────────────── diff --git a/src/lib/craft/tests/test_craft_write.cpp b/src/lib/craft/tests/test_craft_write.cpp index 3c57db0..a258bab 100644 --- a/src/lib/craft/tests/test_craft_write.cpp +++ b/src/lib/craft/tests/test_craft_write.cpp @@ -31,6 +31,7 @@ #include "craft/craft_repl_dev.hpp" #include "coro_helpers.hpp" +#include "mock_journal_backend.hpp" SISL_LOGGING_DEF(HOMEBLOCKS_LOG_MODS) SISL_LOGGING_INIT(HOMEBLOCKS_LOG_MODS) @@ -66,12 +67,7 @@ class MockCraftJournalBackend : public CraftJournalBackend { co_return ok(); } - async_result< JournalSlot > read_slot(int64_t lsn) override { - auto it = slots.find(lsn); - if (it == slots.end()) - co_return std::unexpected(std::make_error_condition(std::errc::no_such_file_or_directory)); - co_return it->second; - } + async_result< JournalSlot > read_slot(int64_t lsn) override { return mock_read_slot(*this, lsn); } async_status truncate_to(int64_t) override { co_return ok(); } async_status free_data(homestore::multi_blk_id) override { @@ -79,6 +75,8 @@ class MockCraftJournalBackend : public CraftJournalBackend { co_return ok(); } + async_status free_slot(int64_t lsn) override { return mock_free_slot(*this, lsn); } + bool has_slot(int64_t lsn) const { return slots.contains(lsn); } size_t slot_count() const { return slots.size(); } }; @@ -90,7 +88,7 @@ class CraftWriteTest : public ::testing::Test { void SetUp() override { auto mock = std::make_unique< MockCraftJournalBackend >(); journal_ = mock.get(); - dev_ = std::make_unique< CraftReplDev >(volume_id_t{}, std::move(mock)); + dev_ = CraftReplDev::create(volume_id_t{}, std::move(mock)); } auto do_write(uint64_t term, int64_t lsn, bool all_zeros = true) { @@ -108,7 +106,7 @@ class CraftWriteTest : public ::testing::Test { } MockCraftJournalBackend* journal_{nullptr}; - std::unique_ptr< CraftReplDev > dev_; + std::shared_ptr< CraftReplDev > dev_; }; // Each lsn arrives exactly one step ahead: no gap, no missing entries after each write. diff --git a/src/lib/home_blks_config.fbs b/src/lib/home_blks_config.fbs index f0d5ae9..56ad7d4 100644 --- a/src/lib/home_blks_config.fbs +++ b/src/lib/home_blks_config.fbs @@ -24,6 +24,10 @@ table HomeBlksSettings{ // how often (in appended LSNs) the leader auto-proposes a SyncRSCommitLSN entry; sync_rs_commit_lsn_interval: uint32 = 128; + + // deadline (in milliseconds) for a server-to-server peer fetch_data() call; + // a peer that misses it is treated as unreachable, same as a hard failure. + peer_fetch_timeout_ms: uint32 = 5000; } root_type HomeBlksSettings;