add missing flush inblockstore compact (#411)

- Bugfix: Flush the last block before closing the last new block written to during blockstore compact. UE-291196
- Feature: Drop unreachable CAS data during GC pass. UE-291196

1 files changed, 67 insertions, 11 deletions

diff --git a/src/zenstore/blockstore.cpp b/src/zenstore/blockstore.cpp
index 86dbcc971..5081ae65d 100644
--- a/src/zenstore/blockstore.cpp
+++ b/src/zenstore/blockstore.cpp
@@ -1004,6 +1004,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 
 	uint32_t		 NewBlockIndex = 0;
 	MovedChunksArray MovedChunks;
+	ChunkIndexArray	 ScrubbedChunks;
 
 	uint64_t AddedSize	 = 0;
 	uint64_t RemovedSize = 0;
@@ -1032,14 +1033,15 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 
 	auto ReportChanges = [&]() -> bool {
 		bool Continue = true;
-		if (!MovedChunks.empty() || RemovedSize > 0)
+		if (!MovedChunks.empty() || !ScrubbedChunks.empty() || RemovedSize > 0)
 		{
-			Continue = ChangeCallback(MovedChunks, RemovedSize > AddedSize ? RemovedSize - AddedSize : 0);
+			Continue = ChangeCallback(MovedChunks, ScrubbedChunks, RemovedSize > AddedSize ? RemovedSize - AddedSize : 0);
 			DeletedSize += RemovedSize;
 			RemovedSize = 0;
 			AddedSize	= 0;
 			MovedCount += MovedChunks.size();
 			MovedChunks.clear();
+			ScrubbedChunks.clear();
 		}
 		return Continue;
 	};
@@ -1068,6 +1070,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 						 LogPrefix,
 						 m_BlocksBasePath,
 						 BlockIndex);
+				ScrubbedChunks.insert(ScrubbedChunks.end(), KeepChunkIndexes.begin(), KeepChunkIndexes.end());
 				return true;
 			}
 			if (!It->second)
@@ -1076,6 +1079,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 						 LogPrefix,
 						 m_BlocksBasePath,
 						 BlockIndex);
+				ScrubbedChunks.insert(ScrubbedChunks.end(), KeepChunkIndexes.begin(), KeepChunkIndexes.end());
 				return true;
 			}
 			OldBlockFile = It->second;
@@ -1115,6 +1119,7 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 						ChunkLocation.Size,
 						OldBlockFile->GetPath(),
 						OldBlockSize);
+					ScrubbedChunks.push_back(ChunkIndex);
 					continue;
 				}
 
@@ -1128,7 +1133,11 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 
 				if ((WriteOffset + ChunkView.GetSize()) > m_MaxBlockSize)
 				{
-					TargetFileBuffer.reset();
+					if (TargetFileBuffer)
+					{
+						TargetFileBuffer->Flush();
+						TargetFileBuffer.reset();
+					}
 					if (NewBlockFile)
 					{
 						ZEN_ASSERT_SLOW(NewBlockFile->IsOpen());
@@ -1260,6 +1269,12 @@ BlockStore::CompactBlocks(const BlockStoreCompactState&	  CompactState,
 		return true;
 	});
 
+	if (TargetFileBuffer)
+	{
+		TargetFileBuffer->Flush();
+		TargetFileBuffer.reset();
+	}
+
 	if (NewBlockFile)
 	{
 		ZEN_ASSERT_SLOW(NewBlockFile->IsOpen());
@@ -1853,7 +1868,7 @@ TEST_CASE("blockstore.compact.blocks")
 		Store.CompactBlocks(
 			State,
 			Alignment,
-			[&](const BlockStore::MovedChunksArray&, uint64_t) {
+			[&](const BlockStore::MovedChunksArray&, const BlockStore::ChunkIndexArray&, uint64_t) {
 				CHECK(false);
 				return true;
 			},
@@ -1878,9 +1893,10 @@ TEST_CASE("blockstore.compact.blocks")
 		Store.CompactBlocks(
 			State,
 			Alignment,
-			[&](const BlockStore::MovedChunksArray& Moved, uint64_t Removed) {
+			[&](const BlockStore::MovedChunksArray& Moved, const BlockStore::ChunkIndexArray& Scrubbed, uint64_t Removed) {
 				RemovedSize += Removed;
 				CHECK(Moved.empty());
+				CHECK(Scrubbed.empty());
 				return true;
 			},
 			[]() { return 0; });
@@ -1903,9 +1919,10 @@ TEST_CASE("blockstore.compact.blocks")
 		Store.CompactBlocks(
 			State,
 			Alignment,
-			[&](const BlockStore::MovedChunksArray& Moved, uint64_t Removed) {
+			[&](const BlockStore::MovedChunksArray& Moved, const BlockStore::ChunkIndexArray& Scrubbed, uint64_t Removed) {
 				RemovedSize += Removed;
 				CHECK(Moved.empty());
+				CHECK(Scrubbed.empty());
 				return true;
 			},
 			[]() { return 0; });
@@ -1913,7 +1930,7 @@ TEST_CASE("blockstore.compact.blocks")
 		CHECK_LE(Store.TotalSize(), 1088);
 		CHECK_GT(Store.TotalSize(), 0);
 	}
-	SUBCASE("keep everthing")
+	SUBCASE("keep everything")
 	{
 		Store.Flush(true);
 
@@ -1926,7 +1943,7 @@ TEST_CASE("blockstore.compact.blocks")
 		Store.CompactBlocks(
 			State,
 			Alignment,
-			[&](const BlockStore::MovedChunksArray&, uint64_t) {
+			[&](const BlockStore::MovedChunksArray&, const BlockStore::ChunkIndexArray&, uint64_t) {
 				CHECK(false);
 				return true;
 			},
@@ -1957,8 +1974,9 @@ TEST_CASE("blockstore.compact.blocks")
 		Store.CompactBlocks(
 			State,
 			Alignment,
-			[&](const BlockStore::MovedChunksArray& Moved, uint64_t Removed) {
+			[&](const BlockStore::MovedChunksArray& Moved, const BlockStore::ChunkIndexArray& Scrubbed, uint64_t Removed) {
 				CHECK(Moved.empty());
+				CHECK(Scrubbed.empty());
 				RemovedSize += Removed;
 				return true;
 			},
@@ -1992,7 +2010,8 @@ TEST_CASE("blockstore.compact.blocks")
 		Store.CompactBlocks(
 			State,
 			Alignment,
-			[&](const BlockStore::MovedChunksArray& Moved, uint64_t Removed) {
+			[&](const BlockStore::MovedChunksArray& Moved, const BlockStore::ChunkIndexArray& Scrubbed, uint64_t Removed) {
+				CHECK(Scrubbed.empty());
 				for (const auto& Move : Moved)
 				{
 					const BlockStoreLocation& OldLocation = State.GetLocation(Move.first);
@@ -2069,7 +2088,8 @@ TEST_CASE("blockstore.compact.blocks")
 		Store.CompactBlocks(
 			State,
 			Alignment,
-			[&](const BlockStore::MovedChunksArray& Moved, uint64_t Removed) {
+			[&](const BlockStore::MovedChunksArray& Moved, const BlockStore::ChunkIndexArray& Scrubbed, uint64_t Removed) {
+				CHECK(Scrubbed.empty());
 				for (const auto& Move : Moved)
 				{
 					const BlockStoreLocation& OldLocation = State.GetLocation(Move.first);
@@ -2104,6 +2124,42 @@ TEST_CASE("blockstore.compact.blocks")
 		}
 		CHECK_LT(Store.TotalSize(), PreSize);
 	}
+	SUBCASE("scrub")
+	{
+		Store.Flush(true);
+
+		BlockStoreCompactState State;
+		for (const BlockStoreLocation& Location : ChunkLocations)
+		{
+			State.IncludeBlock(Location.BlockIndex);
+			CHECK(State.AddKeepLocation(Location));
+		}
+		State.IncludeBlock(0);
+		State.IncludeBlock(999);
+		std::vector<size_t> ExpectedScrubbedIndexes;
+		ExpectedScrubbedIndexes.push_back(ChunkLocations.size() + 0);
+		State.AddKeepLocation(BlockStoreLocation{.BlockIndex = 0, .Offset = 2000, .Size = 322});
+		ExpectedScrubbedIndexes.push_back(ChunkLocations.size() + 1);
+		State.AddKeepLocation(BlockStoreLocation{.BlockIndex = 0, .Offset = 10, .Size = 3220});
+		ExpectedScrubbedIndexes.push_back(ChunkLocations.size() + 2);
+		State.AddKeepLocation(BlockStoreLocation{.BlockIndex = 999, .Offset = 2, .Size = 40});
+
+		std::vector<size_t> ScrubbedIndexes;
+
+		Store.CompactBlocks(
+			State,
+			Alignment,
+			[&](const BlockStore::MovedChunksArray&, const BlockStore::ChunkIndexArray& ScrubbedArray, uint64_t) {
+				ScrubbedIndexes.insert(ScrubbedIndexes.end(), ScrubbedArray.begin(), ScrubbedArray.end());
+				return true;
+			},
+			[]() {
+				CHECK(false);
+				return 0;
+			});
+		std::sort(ScrubbedIndexes.begin(), ScrubbedIndexes.end());
+		CHECK_EQ(ExpectedScrubbedIndexes, ScrubbedIndexes);
+	}
 }
 
 #endif