Skip to content

move seal_shard shard state change from pre_commit to commit #457

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
xiaoxichen merged 1 commit into from
Aug 13, 2026
Merged

move seal_shard shard state change from pre_commit to commit #457

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion conanfile.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@

class HomeObjectConan(ConanFile):
name = "homeobject"
version = "4.3.0"
version = "4.3.1"

homepage = "https://github.com/eBay/HomeObject"
description = "Blob Store built on HomeStore"
Expand Down
81 changes: 19 additions & 62 deletions src/lib/homestore_backend/hs_shard_manager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -397,42 +397,10 @@ bool HSHomeObject::on_shard_message_pre_commit(int64_t lsn, sisl::blob const& he
return false;
}

#ifdef _PRERELEASE
if (msg_header->msg_type == ReplicationMessageType::SEAL_SHARD_MSG) {
// Pause SEAL pre_commit at function entry, before state=SEALED. Test thread can race
// _put_blob while shard is still OPEN; sealed_lsn guard rejects the late blob on commit.
iomgr_flip::instance()->callback_flip("pause_seal_pre_commit");
}
#endif

switch (msg_header->msg_type) {
case ReplicationMessageType::SEAL_SHARD_MSG: {
auto decoded_shard_sb =
decode_shard_sb(r_cast< const shard_info_superblk* >(header.cbytes() + sizeof(ReplicationMessageHeader)));
RELEASE_ASSERT(decoded_shard_sb.has_value(), "failed to decode shard superblk in pre_commit SEAL_SHARD_MSG");
auto& [shard_info, p_chunk_id_unused, v_chunk_id_unused] = decoded_shard_sb.value();

{
std::scoped_lock lock_guard(_shard_lock);
auto iter = _shard_map.find(shard_info.id);
RELEASE_ASSERT(iter != _shard_map.end(), "shardID=0x{:x}, pg={}, shard=0x{:x}, shard does not exist",
shard_info.id, (shard_info.id >> homeobject::shard_width),
(shard_info.id & homeobject::shard_mask));
auto& state = (*iter->second)->info.state;
// we just change the state to SEALED, so that it will fail the later coming put_blob on this shard and will
// be easy for rollback.
// the update of superblk will be done in on_shard_message_commit;
if (state == ShardInfo::State::OPEN) {
state = ShardInfo::State::SEALED;
} else {
SLOGW(tid, shard_info.id, "try to seal an unopened shard");
}
}
}
default: {
break;
}
}
// SEAL_SHARD state transition is done in on_shard_message_commit, not here.
// Reason: create+seal can land in the same raft batch; create only inserts the shard into
// _shard_map on commit, so seal pre_commit would fail to find the shard.
// Late put_blob is rejected by the sealed_lsn guard in on_blob_put_commit.
return true;
}

Expand All @@ -459,27 +427,9 @@ void HSHomeObject::on_shard_message_rollback(int64_t lsn, sisl::blob const& head
break;
}
case ReplicationMessageType::SEAL_SHARD_MSG: {
auto decoded_shard_sb =
decode_shard_sb(r_cast< const shard_info_superblk* >(header.cbytes() + sizeof(ReplicationMessageHeader)));
RELEASE_ASSERT(decoded_shard_sb.has_value(), "failed to decode shard superblk in rollback SEAL_SHARD_MSG");
auto& [shard_info, p_chunk_id_unused, v_chunk_id_unused] = decoded_shard_sb.value();
{
std::scoped_lock lock_guard(_shard_lock);
auto iter = _shard_map.find(shard_info.id);
RELEASE_ASSERT(iter != _shard_map.end(), "shardID=0x{:x}, pg={}, shard=0x{:x}, shard does not exist",
shard_info.id, (shard_info.id >> homeobject::shard_width),
(shard_info.id & homeobject::shard_mask));
auto& state = (*iter->second)->info.state;
// we just change the state to SEALED, since it will be easy for rollback
// the update of superblk will be done in on_shard_message_commit;
if (state == ShardInfo::State::SEALED) {
state = ShardInfo::State::OPEN;
SLOGD(tid, shard_info.id, "rollback seal shard message, lsn={}", lsn);
} else {
SLOGW(tid, shard_info.id,
"try to rollback seal_shard message , but the shard state is not sealed, lsn= {}", lsn);
}
}
// Seal no longer mutates shard state in pre_commit, so rollback has nothing to revert
// in the shard map. Just fail the proposer's promise.
SLOGD(tid, msg_header->shard_id, "rollback seal shard message, type={}, lsn={}", msg_header->msg_type, lsn);
// TODO:set a proper error code
if (ctx) { ctx->promise_.setValue(folly::makeUnexpected(ShardError(ShardErrorCode::RETRY_REQUEST))); }

Expand Down Expand Up @@ -532,6 +482,10 @@ void HSHomeObject::on_shard_message_commit(int64_t lsn, sisl::blob const& h, sha

#ifdef _PRERELEASE
if (header->msg_type == ReplicationMessageType::SEAL_SHARD_MSG) {
// Pause SEAL commit at function entry, before state=SEALED / sealed_lsn update.
// Test thread can race _put_blob while shard is still OPEN; sealed_lsn guard rejects
// the late blob when this gate is released and seal commit finishes.
iomgr_flip::instance()->callback_flip("pause_seal_commit");
// Wait for CREATE_SHARD (next log) to be in log store before SEAL releases its vchunk.
// Polled via get_last_append_lsn() so it doesn't rely on pre_commit signals. Armed on the
// repro follower only; no-op on leader and other followers.
Expand Down Expand Up @@ -567,19 +521,22 @@ void HSHomeObject::on_shard_message_commit(int64_t lsn, sisl::blob const& h, sha
RELEASE_ASSERT(decoded_shard_sb.has_value(), "failed to decode shard superblk in commit SEAL_SHARD_MSG");
auto& [shard_info, p_chunk_id_unused, v_chunk_id_unused] = decoded_shard_sb.value();

ShardInfo::State state;
{
std::scoped_lock lock_guard(_shard_lock);
auto iter = _shard_map.find(shard_info.id);
RELEASE_ASSERT(iter != _shard_map.end(), "shardID=0x{:x}, pg={}, shard=0x{:x}, shard does not exist",
shard_info.id, (shard_info.id >> homeobject::shard_width),
(shard_info.id & homeobject::shard_mask));
state = (*iter->second)->info.state;
auto& state = (*iter->second)->info.state;
// State transition moved here from pre_commit so create+seal in the same raft batch works:
// create commit adds the shard to the map before seal commit runs.
if (state == ShardInfo::State::OPEN) {
state = ShardInfo::State::SEALED;
} else {
SLOGW(tid, shard_info.id, "try to seal an unopened shard, current_state={}", static_cast< int >(state));
}
}

RELEASE_ASSERT(state == ShardInfo::State::SEALED,
"try to commit SEAL_SHARD_MSG but shard state is not sealed. shardID={}", shard_info.id);

// Corner case:
// Assume cp_lsn = dc_lsn = 10.
// lsn 11: put_blob (blob -> pba in chunk-1)
Expand Down
50 changes: 24 additions & 26 deletions src/lib/homestore_backend/tests/hs_gc_tests.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1094,25 +1094,25 @@ TEST_F(HomeObjectFixture, StalePChunkRouteAfterGC) {
}

// ===================================================================================================
// StaleBlobRouteAfterSealAndGC: PUT_BLOB races with SEAL_SHARD pre_commit; sealed_lsn guard rejects it.
// StaleBlobRouteAfterSealAndGC: PUT_BLOB races with SEAL_SHARD commit; sealed_lsn guard rejects it.
//
// Production scenario :
// A PUT_BLOB whose admission check passed (shard OPEN) should be rejected if the shard gets
// sealed before the blob is committed. The sealed_lsn guard in on_blob_put_commit must catch it.
//
// Exact sequence modelled (single replica, leader):
// ① SEAL_SHARD pre_commit fires and PAUSES before changing state=SEALED
// (flip "pause_seal_pre_commit"). At this point shard state is still OPEN.
// ① SEAL_SHARD commit fires and PAUSES before changing state=SEALED / sealed_lsn
// (flip "pause_seal_commit"). At this point shard state is still OPEN.
// ② _put_blob is called in the test thread. get_blk_alloc_hints sees state==OPEN → passes,
// blk is allocated on pchunk_A. The put is async (raft not yet committed).
// ③ Gate releases → state = SEALED → SEAL_SHARD commit → sealed_lsn = X.
// ③ Gate releases → state = SEALED → sealed_lsn = X.
// ④ PUT_BLOB commit (lsn = X+1): on_blob_put_commit checks lsn(X+1) >= sealed_lsn(X) → reject.
// The allocated blk is freed; the blob does NOT land in the pg index.
//
// Verification: the late blob is absent from the index; bulk blobs are still readable.
//
// Runs on the leader replica only; no multi-replica complexity needed.
// Pause point: flip "pause_seal_pre_commit". Compiled out of release builds.
// Pause point: flip "pause_seal_commit". Compiled out of release builds.
// ===================================================================================================
TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) {
const pg_id_t pg_id = 1;
Expand Down Expand Up @@ -1141,23 +1141,22 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) {
std::map< pg_id_t, blob_id_t > pg_blob_id{{pg_id, 0}};
put_blobs(shards, num_blobs_per_shard, pg_blob_id);

// ---- arm the flip on the leader: pause SEAL pre_commit before state=SEALED ----
// ---- arm the flip on the leader: pause SEAL commit before state=SEALED ----
if (i_am_leader) {
auto dont_care = m_fc.create_condition("", flip::Operator::DONT_CARE, (int)0);
flip::FlipFrequency freq;
freq.set_count(3); // 3 replicas all call callback_flip; count must be >= num_replicas
freq.set_percent(100);
m_fc.inject_callback_flip< void >("pause_seal_pre_commit", {dont_care}, freq, std::function< void() >([&]() {
LOGI(
"[StaleBlobRouteAfterSealAndGC] pausing SEAL pre_commit BEFORE lock");
std::unique_lock< std::mutex > lk(repro2_mtx);
repro2_blocked.store(true);
repro2_cv.notify_all();
repro2_cv.wait(lk, [&] { return repro2_released.load(); });
LOGI("[StaleBlobRouteAfterSealAndGC] resuming SEAL pre_commit");
}));
LOGINFO("[StaleBlobRouteAfterSealAndGC] armed pause_seal_pre_commit on leader replica={}",
g_helper->replica_num());
m_fc.inject_callback_flip< void >(
"pause_seal_commit", {dont_care}, freq, std::function< void() >([&]() {
LOGI("[StaleBlobRouteAfterSealAndGC] pausing SEAL commit BEFORE state update");
std::unique_lock< std::mutex > lk(repro2_mtx);
repro2_blocked.store(true);
repro2_cv.notify_all();
repro2_cv.wait(lk, [&] { return repro2_released.load(); });
LOGI("[StaleBlobRouteAfterSealAndGC] resuming SEAL commit");
}));
LOGINFO("[StaleBlobRouteAfterSealAndGC] armed pause_seal_commit on leader replica={}", g_helper->replica_num());
}

g_helper->sync(); // make sure flip is armed on all replicas before proceeding
Expand All @@ -1166,26 +1165,26 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) {
blob_id_t late_blob_id [[maybe_unused]] = INVALID_UINT64_ID;
if (i_am_leader) {
// 1. Start seal_shard in a background thread so it runs concurrently.
// seal_shard will hit the gate in pre_commit and pause there.
// seal_shard will hit the gate in commit and pause there.
auto tid = generateRandomTraceId();
bool seal_ok = false;
std::thread seal_thread([&]() {
auto r = std::move(_obj_inst->shard_manager()->seal_shard(shard1.id, tid)).get();
seal_ok = r.hasValue();
});

// 2. Wait until pre_commit is paused (shard state is still OPEN).
// 2. Wait until commit is paused (shard state is still OPEN).
{
std::unique_lock< std::mutex > lk(repro2_mtx);
if (!repro2_cv.wait_for(lk, std::chrono::seconds(30), [&] { return repro2_blocked.load(); })) {
repro2_released.store(true); // avoid deadlock if gate never fires
repro2_cv.notify_all();
seal_thread.join();
m_fc.remove_flip("pause_seal_pre_commit");
FAIL() << "SEAL pre_commit never reached the pause point";
m_fc.remove_flip("pause_seal_commit");
FAIL() << "SEAL commit never reached the pause point";
}
}
LOGINFO("[StaleBlobRouteAfterSealAndGC] leader sees SEAL pre_commit paused; shard state=OPEN; "
LOGINFO("[StaleBlobRouteAfterSealAndGC] leader sees SEAL commit paused; shard state=OPEN; "
"calling _put_blob with shard still OPEN");

// 3. Call _put_blob in a background thread: shard state is OPEN → get_blk_alloc_hints
Expand All @@ -1199,9 +1198,8 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) {
b.hasValue() ? "admitted" : "rejected");
});

// 4. Release the gate: state = SEALED, seal pre_commit returns → raft commits seal.
// After seal commit, sealed_lsn = lsn_seal. Then put_blob commit fires and
// on_blob_put_commit checks lsn(put) >= sealed_lsn → rejects.
// 4. Release the gate: state = SEALED, seal commit returns with sealed_lsn = lsn_seal.
// Then put_blob commit fires and on_blob_put_commit checks lsn(put) >= sealed_lsn → rejects.
{
std::unique_lock< std::mutex > lk(repro2_mtx);
repro2_released.store(true);
Expand All @@ -1212,7 +1210,7 @@ TEST_F(HomeObjectFixture, StaleBlobRouteAfterSealAndGC) {
// 5. Wait for both background threads.
blob_thread.join();
seal_thread.join();
m_fc.remove_flip("pause_seal_pre_commit");
m_fc.remove_flip("pause_seal_commit");
ASSERT_TRUE(seal_ok) << "seal_shard failed";

EXPECT_TRUE(blob_rejected)
Expand Down
Loading