Skip to content

Fix snapshot receiver progress on replay #453

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
koujl merged 1 commit into from
Jul 29, 2026
Merged

Fix snapshot receiver progress on replay #453

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.2.6"
version = "4.2.7"

homepage = "https://github.com/eBay/HomeObject"
description = "Blob Store built on HomeStore"
Expand Down
6 changes: 6 additions & 0 deletions src/lib/homestore_backend/hs_homeobject.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -667,8 +667,14 @@ class HSHomeObject : public HomeObjectImpl {
const int64_t snp_lsn;
const pg_id_t pg_id;
shared< BlobIndexTable > index_table;

// Progress and accounting fields
uint64_t cur_shard_complete_blobs{0};
uint64_t cur_shard_complete_bytes{0};
bool cur_shard_completed{false};
std::shared_mutex progress_lock;
snapshot_progress progress;

SnapshotContext(int64_t lsn, pg_id_t pg_id) : snp_lsn{lsn}, pg_id{pg_id} {}
};

Expand Down
35 changes: 31 additions & 4 deletions src/lib/homestore_backend/snapshot_receive_handler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,19 @@ int HSHomeObject::SnapshotReceiveHandler::process_shard_snapshot_data(ResyncShar
shard_sb->v_chunk_id = shard_meta.vchunk_id();
// Now let's create local shard
home_obj_.local_create_shard(shard_sb->info, shard_sb->v_chunk_id);
{
std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock);
if (ctx_->shard_cursor == shard_meta.shard_id()) {
ctx_->progress.complete_blobs -= ctx_->cur_shard_complete_blobs;
ctx_->progress.complete_bytes -= ctx_->cur_shard_complete_bytes;
if (ctx_->cur_shard_completed) { ctx_->progress.complete_shards--; }
}
ctx_->cur_shard_complete_blobs = 0;
ctx_->cur_shard_complete_bytes = 0;
ctx_->cur_shard_completed = false;
ctx_->progress.cur_batch_blobs = 0;
ctx_->progress.cur_batch_bytes = 0;
}
ctx_->shard_cursor = shard_meta.shard_id();
ctx_->cur_batch_num = 0;
return 0;
Expand All @@ -117,6 +130,13 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob
std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock);
ctx_->progress.complete_blobs -= ctx_->progress.cur_batch_blobs;
ctx_->progress.complete_bytes -= ctx_->progress.cur_batch_bytes;
ctx_->cur_shard_complete_blobs -= ctx_->progress.cur_batch_blobs;
ctx_->cur_shard_complete_bytes -= ctx_->progress.cur_batch_bytes;
ctx_->progress.cur_batch_blobs = 0;
ctx_->progress.cur_batch_bytes = 0;
} else {
std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock);
// A new batch must not inherit the prior batch's rollback contribution.
ctx_->progress.cur_batch_blobs = 0;
ctx_->progress.cur_batch_bytes = 0;
}
Expand All @@ -136,6 +156,7 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob
std::vector< std::shared_ptr< sisl::io_blob_safe > > data_bufs;

auto skipped_blobs = 0;
bool found_unexpected_corruption = false;
for (unsigned int i = 0; i < data_blobs.blob_list()->size(); i++) {
const auto blob = data_blobs.blob_list()->Get(i);

Expand All @@ -146,6 +167,9 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob
continue;
}

// Count the logical batch size even when an existing blob is re-committed during replay.
total_bytes += blob->data()->size();

auto start = Clock::now();

#ifdef _PRERELEASE
Expand Down Expand Up @@ -188,7 +212,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob
LOGE("Blob verification failed for blob_id={}", blob->blob_id());
std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock);
ctx_->progress.error_count++;
return BLOB_DATA_CORRUPTED;
found_unexpected_corruption = true;
break;
}
} else {
LOGW("find corrupted_blobs={} in shardID=0x{:x}, pg={}, shard=0x{:x}", blob->blob_id(), ctx_->shard_cursor,
Expand Down Expand Up @@ -271,9 +296,9 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob
LOGD("Persisted blob_id={} in {}us", blob_id, duration);
return std::error_code{};
}));
total_bytes += data_size;
}
auto ec = collect_all_futures(futs).get();
if (found_unexpected_corruption) { return BLOB_DATA_CORRUPTED; }
// when there is a allocation failure it breaks the while loop earlier.
auto all_io_submitted = (futs.size() + skipped_blobs == data_blobs.blob_list()->size());

Expand All @@ -298,8 +323,9 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob
ctx_->progress.cur_batch_bytes = total_bytes;
ctx_->progress.complete_blobs += ctx_->progress.cur_batch_blobs;
ctx_->progress.complete_bytes += ctx_->progress.cur_batch_bytes;
ctx_->cur_shard_complete_blobs += ctx_->progress.cur_batch_blobs;
ctx_->cur_shard_complete_bytes += ctx_->progress.cur_batch_bytes;
}

if (is_last_batch) {
// Release chunk for sealed shard
ShardInfo::State state;
Expand All @@ -313,7 +339,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob
}
{
std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock);
ctx_->progress.complete_shards++;
if (!ctx_->cur_shard_completed) { ctx_->progress.complete_shards++; }
ctx_->cur_shard_completed = true;
}
// We only update the snp info superblk on completion of each shard, since resumption is also shard-level
update_snp_info_sb(ctx_->shard_cursor == ctx_->shard_list.front());
Expand Down
71 changes: 68 additions & 3 deletions src/lib/homestore_backend/tests/homeobj_misc_tests.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,7 +436,8 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) {
}
auto pg_entry =
CreateResyncPGMetaDataDirect(builder, pg_id, &uuid, pg->pg_info_.size, pg->pg_info_.expected_member_num,
pg->pg_info_.chunk_size, blob_seq_num, num_shards_per_pg, &members, &shard_ids);
pg->pg_info_.chunk_size, blob_seq_num, num_shards_per_pg, &members, &shard_ids,
blob_seq_num /* total_blobs_to_transfer */);
builder.Finish(pg_entry);
auto pg_meta = GetResyncPGMetaData(builder.GetBufferPointer());
auto ret = handler->process_pg_snapshot_data(*pg_meta);
Expand All @@ -450,6 +451,7 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) {
std::uniform_int_distribution<> random_bytes_dis(1, 16 * 1024);

blob_id_t cur_blob_id{0};
uint64_t expected_complete_blobs{0};
for (uint64_t i = 1; i <= num_shards_per_pg; i++) {
LOGINFO("TESTING: applying meta for shard {}", i);
ASSERT_TRUE(handler->is_valid_obj_id(objId(HSHomeObject::get_sequence_num_from_shard_id(i), 0)));
Expand Down Expand Up @@ -503,11 +505,12 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) {
for (uint64_t j = 1; j <= num_batches_per_shard; j++) {
ASSERT_TRUE(handler->is_valid_obj_id(objId(HSHomeObject::get_sequence_num_from_shard_id(shard.id), j)));

// Don't test unexpected corruption on the last batch, since for simplicity we're not simulating resending
// Keep the final batch clean so each shard completes even if a prior batch needs a resend.
bool is_corrupted_batch =
j < num_batches_per_shard && corrupt_dis(gen) <= unexpected_corrupted_batch_percentage;
LOGINFO("TESTING: applying blobs for shard {} batch {}, is_corrupted {}", shard.id, j, is_corrupted_batch);
std::vector< flatbuffers::Offset< ResyncBlobData > > blob_entries;
auto const batch_start_blob_id = cur_blob_id;
for (uint64_t k = 0; k < num_blobs_per_batch; k++) {
auto blob_state = corrupt_dis(gen) <= corrupted_blob_percentage ? ResyncBlobState::CORRUPTED
: ResyncBlobState::NORMAL;
Expand Down Expand Up @@ -560,9 +563,46 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) {
ret = handler->process_blobs_snapshot_data(*blob_batch, j, j == num_batches_per_shard);
if (is_corrupted_batch) {
ASSERT_NE(ret, 0);
builder.Reset();
std::vector< flatbuffers::Offset< ResyncBlobData > > retry_blob_entries;
for (blob_id_t retry_blob_id = batch_start_blob_id;
retry_blob_id < batch_start_blob_id + num_blobs_per_batch; retry_blob_id++) {
auto retry_blob = build_blob(retry_blob_id);
const auto aligned_hdr_size =
sisl::round_up(sizeof(HSHomeObject::BlobHeader), _obj_inst->_data_block_size);
sisl::io_blob_safe retry_blob_raw(aligned_hdr_size + retry_blob.body.size(), io_align);
HSHomeObject::BlobHeader retry_hdr;
retry_hdr.type = HSHomeObject::DataHeader::data_type_t::BLOB_INFO;
retry_hdr.shard_id = shard.id;
retry_hdr.blob_id = retry_blob_id;
retry_hdr.hash_algorithm = HSHomeObject::BlobHeader::HashAlgorithm::CRC32;
retry_hdr.blob_size = retry_blob.body.size();
retry_hdr.user_key_size = retry_blob.user_key.size();
retry_hdr.object_offset = retry_blob.object_off;
retry_hdr.data_offset = aligned_hdr_size;
if (!retry_blob.user_key.empty()) {
std::memcpy(retry_hdr.user_key, retry_blob.user_key.data(), retry_blob.user_key.size());
}
_obj_inst->compute_blob_payload_hash(retry_hdr.hash_algorithm, retry_blob.body.cbytes(),
retry_blob.body.size(), retry_hdr.hash,
HSHomeObject::BlobHeader::blob_max_hash_len);
retry_hdr.seal();
std::memcpy(retry_blob_raw.bytes(), &retry_hdr, sizeof(HSHomeObject::BlobHeader));
std::memcpy(retry_blob_raw.bytes() + retry_hdr.data_offset, retry_blob.body.cbytes(),
retry_blob.body.size());
std::vector retry_data(retry_blob_raw.bytes(), retry_blob_raw.bytes() + retry_blob_raw.size());
retry_blob_entries.push_back(CreateResyncBlobDataDirect(
builder, retry_blob_id, static_cast< uint8_t >(ResyncBlobState::NORMAL), &retry_data));
blob_map[retry_blob_id] = std::make_tuple< Blob, bool >(std::move(retry_blob), false);
}
builder.Finish(CreateResyncBlobDataBatchDirect(builder, &retry_blob_entries, j == num_batches_per_shard));
auto retry_blob_batch = GetResyncBlobDataBatch(builder.GetBufferPointer());
ASSERT_EQ(handler->process_blobs_snapshot_data(*retry_blob_batch, j, j == num_batches_per_shard), 0);
} else {
ASSERT_EQ(ret, 0);
}
expected_complete_blobs += num_blobs_per_batch;
ASSERT_EQ(handler->ctx_->progress.complete_blobs, expected_complete_blobs);
builder.Reset();
ASSERT_EQ(handler->get_shard_cursor(), shard.id);
ASSERT_EQ(handler->get_next_shard(),
Expand Down Expand Up @@ -593,7 +633,10 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandler) {
ASSERT_EQ(v.value(), exVchunk->m_v_chunk_id.value());
ASSERT_TRUE(_obj_inst->chunk_selector()->is_chunk_available(pg_id, v.value()));
}
ASSERT_EQ(handler->ctx_->progress.complete_shards, i);
}
ASSERT_EQ(handler->ctx_->progress.total_blobs, blob_seq_num);
ASSERT_EQ(handler->ctx_->progress.total_shards, num_shards_per_pg);
}

// Test: verify the case (https://jirap.corp.ebay.com/browse/SDSTOR-23023) when the snapshot receiver re-delivers a blob
Expand Down Expand Up @@ -692,6 +735,14 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandlerAllocatorResyncAfterCrash) {
}
b.Finish(CreateResyncBlobDataBatchDirect(b, &blob_entries, is_last));
};
auto batch_bytes = [&](blob_id_t start_blob_id, blob_id_t end_blob_id) {
uint64_t total_bytes{0};
for (blob_id_t blob_id = start_blob_id; blob_id < end_blob_id; blob_id++) {
auto blob = build_blob(blob_id);
total_bytes += sisl::round_up(sizeof(HSHomeObject::BlobHeader), _obj_inst->_data_block_size) + blob.body.size();
}
return total_bytes;
};

// ---- First delivery: partial batch (blobs 0..num_blobs_partial-1), not the last batch ----
// Simulates the pre-crash state: these blobs are written to disk, committed, and indexed.
Expand All @@ -702,6 +753,8 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandlerAllocatorResyncAfterCrash) {
ASSERT_EQ(handler->process_blobs_snapshot_data(*GetResyncBlobDataBatch(builder.GetBufferPointer()), 1 /*batch_num*/,
false /*is_last_batch*/),
0);
ASSERT_EQ(handler->ctx_->progress.complete_blobs, num_blobs_partial);
ASSERT_EQ(handler->ctx_->progress.complete_bytes, batch_bytes(0, num_blobs_partial));
builder.Reset();

// Confirm only the partial set is readable and indexed
Expand Down Expand Up @@ -750,11 +803,23 @@ TEST_F(HomeObjectFixture, SnapshotReceiveHandlerAllocatorResyncAfterCrash) {
// new blobs are then allocated strictly above them.
// Blobs num_blobs_partial..num_blobs_total-1 are new → normal write path.
LOGINFO("TEST: second (full) delivery after crash, blobs [0, {})", num_blobs_total);
handler->ctx_->cur_batch_num = 0; // reset so batch_num=1 is accepted again
// Replaying shard metadata resets the shard-local contribution before batch 1 is redelivered.
auto replay_shard_entry = CreateResyncShardMetaData(
builder, shard_id, pg_id, static_cast< uint8_t >(ShardInfo::State::OPEN), snp_lsn /*create_lsn*/,
get_time_since_epoch_ms() /*created_time*/, get_time_since_epoch_ms() /*modified_time*/, 1024 * Mi /*capacity*/,
exVchunk->m_v_chunk_id.value(), 0 /*meta*/, 0 /*sealed_lsn*/);
builder.Finish(replay_shard_entry);
ASSERT_EQ(handler->process_shard_snapshot_data(*GetResyncShardMetaData(builder.GetBufferPointer())), 0);
ASSERT_EQ(handler->ctx_->progress.complete_blobs, 0);
ASSERT_EQ(handler->ctx_->progress.complete_bytes, 0);
builder.Reset();
make_blob_batch(builder, 0, num_blobs_total, true /*is_last*/);
ASSERT_EQ(handler->process_blobs_snapshot_data(*GetResyncBlobDataBatch(builder.GetBufferPointer()), 1 /*batch_num*/,
true /*is_last_batch*/),
0);
ASSERT_EQ(handler->ctx_->progress.complete_blobs, num_blobs_total);
ASSERT_EQ(handler->ctx_->progress.complete_bytes, batch_bytes(0, num_blobs_total));
ASSERT_EQ(handler->ctx_->progress.complete_shards, 1);
builder.Reset();

// ---- Verify allocator watermarks advanced beyond the partial set ----
Expand Down
Loading