SDSTOR-23112: fix snapshot shard state when sealed_lsn exceeds snapshot LSN cutoff

conanfile.py

@@ -10,7 +10,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "4.2.4"
+    version = "4.2.5"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeStore"

src/lib/homestore_backend/heap_chunk_selector.cpp

@@ -365,8 +365,6 @@ void HeapChunkSelector::switch_chunks_for_pg(const pg_id_t pg_id, const chunk_nu
     std::unique_lock lk(pg_chunk_collection->mtx);
     auto& pg_chunks = pg_chunk_collection->m_pg_chunks;
 
-    // LOGDEBUGMOD(homeobject, "gc: before switch chunks for pg_id={}, pg_chunks={}", pg_chunks);
-
     if (sisl_unlikely(pg_chunks[v_chunk_id]->get_chunk_id() == new_chunk_id)) {
         // this might happens when crash recovery. the crash happens after pg metablk is updated but before gc task
         // metablk is destroyed.

src/lib/homestore_backend/pg_blob_iterator.cpp

@@ -24,14 +24,37 @@ HSHomeObject::PGBlobIterator::PGBlobIterator(HSHomeObject& home_obj, homestore::
     if (upto_lsn != 0) {
         // Iterate all shards and its blobs which have lsn <= upto_lsn
         for (auto& shard : pg->shards_) {
-            if (shard->info.create_lsn <= upto_lsn) {
-                auto v_chunk_id = home_obj_.get_shard_v_chunk_id(shard->info.id);
-                shard_list_.emplace_back(shard->info, v_chunk_id.value());
+            auto shard_info = shard->info;
+            if (shard_info.create_lsn <= upto_lsn) {
+                auto v_chunk_id = home_obj_.get_shard_v_chunk_id(shard_info.id);
+                RELEASE_ASSERT(v_chunk_id.has_value(), "v_chunk_id not found for shard_id={}", shard_info.id);
+
+                /*
+                 * Snapshot metadata may reflect the leader's latest shard state, even when that
+                 * state transition happened after the snapshot LSN cutoff. If such a shard is
+                 * treated as sealed during snapshot apply, the receiver may release the backing
+                 * chunk too early. That chunk can then be reclaimed or relocated before log replay
+                 * catches up, leaving later writes to use stale physical-chunk information and
+                 * eventually fail during commit.
+                 *
+                 * To keep snapshot state consistent with the requested cutoff, convert any shard
+                 * whose sealed_lsn is beyond upto_lsn back to OPEN for iteration purposes and follower will received
+                 * and commit the seal_shard log for this shard after the snapshot apply is completed, where the shard
+                 * state and sealed_lsn will be updated to the correct value.
+                 */
+
+                if (shard_info.state == ShardInfo::State::SEALED && shard_info.sealed_lsn > upto_lsn) {
+                    shard_info.state = ShardInfo::State::OPEN;
+                    shard_info.sealed_lsn = INT64_MAX;
+                }
+
+                shard_list_.emplace_back(shard_info, v_chunk_id.value());
             }
         }
         // Sort shard list by <vchunkid, lsn> to ensure open shards positioned after sealed shards within each chunk
         std::ranges::sort(shard_list_, [](const ShardEntry& a, const ShardEntry& b) {
-            return a.v_chunk_num != b.v_chunk_num ? a.v_chunk_num < b.v_chunk_num : a.info.create_lsn < b.info.create_lsn;
+            return a.v_chunk_num != b.v_chunk_num ? a.v_chunk_num < b.v_chunk_num
+                                                  : a.info.create_lsn < b.info.create_lsn;
         });
     }
 }

src/lib/homestore_backend/tests/homeobj_misc_tests.cpp

@@ -139,9 +139,16 @@ TEST_F(HomeObjectFixture, PGBlobIterator) {
         auto shard_msg = GetSizePrefixedResyncShardMetaData(meta_data.cbytes() + sizeof(SyncMessageHeader));
         ASSERT_EQ(shard_msg->shard_id(), shard->info.id);
         ASSERT_EQ(shard_msg->pg_id(), pg->pg_info_.id);
-        ASSERT_EQ(shard_msg->state(), static_cast< uint8_t >(shard->info.state));
+        // A shard sealed after the snapshot LSN is downgraded to OPEN for snapshot consistency.
+        auto expected_state = shard->info.state;
+        auto expected_sealed_lsn = shard->info.sealed_lsn;
+        if (expected_state == ShardInfo::State::SEALED && shard->info.sealed_lsn > snp_lsn) {
+            expected_state = ShardInfo::State::OPEN;
+            expected_sealed_lsn = static_cast< uint64_t >(INT64_MAX);
+        }
+        ASSERT_EQ(shard_msg->state(), static_cast< uint8_t >(expected_state));
         ASSERT_EQ(shard_msg->created_lsn(), shard->info.create_lsn);
-        ASSERT_EQ(shard_msg->sealed_lsn(), shard->info.sealed_lsn);
+        ASSERT_EQ(shard_msg->sealed_lsn(), expected_sealed_lsn);
         ASSERT_EQ(shard_msg->created_time(), shard->info.created_time);
         ASSERT_EQ(shard_msg->last_modified_time(), shard->info.last_modified_time);
         ASSERT_EQ(shard_msg->total_capacity_bytes(), shard->info.total_capacity_bytes);
@@ -316,6 +323,74 @@ TEST_F(HomeObjectFixture, PGBlobIteratorGCTombstoneDetection) {
     pg_iter->stop();
 }
 
+// Verifies that a shard sealed after the snapshot LSN cutoff is presented as OPEN to the snapshot
+// receiver (Case A), while a shard sealed at or before the cutoff remains SEALED (Case B).
+//
+// Without this conversion a receiver applying the snapshot could release the shard's backing chunk
+// too early, before log replay has a chance to replay the seal_shard entry, leading to stale
+// physical-chunk references and write failures.
+TEST_F(HomeObjectFixture, PGBlobIteratorSealedLsnCutoff) {
+    constexpr pg_id_t pg_id{1};
+    create_pg(pg_id);
+    auto shard_1_info = create_shard(pg_id, 64 * Mi, "shard1");
+    auto shard_2_info = create_shard(pg_id, 64 * Mi, "shard2");
+
+    // Seal shard_1 after shard_2 is created so that shard_1.sealed_lsn > shard_2.create_lsn.
+    seal_shard(shard_1_info.id);
+
+    auto pg = _obj_inst->get_hs_pg(pg_id);
+    ASSERT_TRUE(pg != nullptr);
+
+    // Identify shards by id to be independent of sort order in shard_list_.
+    Shard* pg_shard_1 = nullptr;
+    Shard* pg_shard_2 = nullptr;
+    for (auto& s : pg->shards_) {
+        if (s->info.id == shard_1_info.id) pg_shard_1 = s.get();
+        if (s->info.id == shard_2_info.id) pg_shard_2 = s.get();
+    }
+    ASSERT_TRUE(pg_shard_1 != nullptr && pg_shard_2 != nullptr);
+    ASSERT_EQ(pg_shard_1->info.state, ShardInfo::State::SEALED);
+    // Guarantee that sealing happened after shard_2 was created.
+    ASSERT_GT(pg_shard_1->info.sealed_lsn, pg_shard_2->info.create_lsn);
+
+    // Case A: snp_lsn falls between shard_1.create_lsn and shard_1.sealed_lsn.
+    // shard_1 must be downgraded to OPEN so the receiver does not release its chunk prematurely.
+    {
+        auto snp_lsn = pg_shard_2->info.create_lsn;
+        ASSERT_GT(snp_lsn, pg_shard_1->info.create_lsn);
+        ASSERT_LT(snp_lsn, pg_shard_1->info.sealed_lsn);
+
+        auto pg_iter = std::make_shared< HSHomeObject::PGBlobIterator >(*_obj_inst, pg->pg_info_.replica_set_uuid, snp_lsn);
+        ASSERT_EQ(pg_iter->shard_list_.size(), 2u);
+
+        auto it1 = std::find_if(pg_iter->shard_list_.begin(), pg_iter->shard_list_.end(),
+                                [&](const auto& e) { return e.info.id == shard_1_info.id; });
+        ASSERT_NE(it1, pg_iter->shard_list_.end());
+        EXPECT_EQ(it1->info.state, ShardInfo::State::OPEN);
+        EXPECT_EQ(it1->info.sealed_lsn, static_cast< uint64_t >(INT64_MAX));
+
+        auto it2 = std::find_if(pg_iter->shard_list_.begin(), pg_iter->shard_list_.end(),
+                                [&](const auto& e) { return e.info.id == shard_2_info.id; });
+        ASSERT_NE(it2, pg_iter->shard_list_.end());
+        EXPECT_EQ(it2->info.state, ShardInfo::State::OPEN);
+    }
+
+    // Case B: snp_lsn >= shard_1.sealed_lsn.
+    // shard_1 was fully sealed within the snapshot range and must remain SEALED.
+    {
+        auto snp_lsn = pg_shard_1->info.sealed_lsn;
+
+        auto pg_iter = std::make_shared< HSHomeObject::PGBlobIterator >(*_obj_inst, pg->pg_info_.replica_set_uuid, snp_lsn);
+        ASSERT_EQ(pg_iter->shard_list_.size(), 2u);
+
+        auto it1 = std::find_if(pg_iter->shard_list_.begin(), pg_iter->shard_list_.end(),
+                                [&](const auto& e) { return e.info.id == shard_1_info.id; });
+        ASSERT_NE(it1, pg_iter->shard_list_.end());
+        EXPECT_EQ(it1->info.state, ShardInfo::State::SEALED);
+        EXPECT_EQ(it1->info.sealed_lsn, pg_shard_1->info.sealed_lsn);
+    }
+}
+
 TEST_F(HomeObjectFixture, SnapshotReceiveHandler) {
     constexpr uint64_t snp_lsn = 1;
     constexpr uint64_t num_shards_per_pg = 3;