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

5 files changed, 130 insertions, 17 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
diff --git a/src/zenstore/buildstore/buildstore.cpp b/src/zenstore/buildstore/buildstore.cpp
index 6eb01dfc4..41c747e08 100644
--- a/src/zenstore/buildstore/buildstore.cpp
+++ b/src/zenstore/buildstore/buildstore.cpp
@@ -1237,7 +1237,9 @@ public:
 						m_Store.m_MetadataBlockStore.CompactBlocks(
 							BlockCompactState,
 							m_Store.m_Config.MetadataBlockStoreAlignement,
-							[&](const BlockStore::MovedChunksArray& MovedArray, uint64_t FreedDiskSpace) {
+							[&](const BlockStore::MovedChunksArray& MovedArray,
+								const BlockStore::ChunkIndexArray&	ScrubbedArray,
+								uint64_t							FreedDiskSpace) {
 								std::vector<MetadataDiskEntry> MovedEntries;
 								MovedEntries.reserve(MovedArray.size());
 								RwLock::ExclusiveLockScope _(m_Store.m_Lock);
@@ -1264,7 +1266,27 @@ public:
 										}
 									}
 								}
+
+								for (size_t Scrubbed : ScrubbedArray)
+								{
+									const IoHash& Key = BlockCompactStateKeys[Scrubbed];
+									if (auto It = m_Store.m_BlobLookup.find(Key); It != m_Store.m_BlobLookup.end())
+									{
+										const BlobIndex Index = It->second;
+
+										if (MetadataIndex Meta = m_Store.m_BlobEntries[Index].Metadata; Meta)
+										{
+											MovedEntries.push_back(
+												MetadataDiskEntry{.Entry = m_Store.m_MetadataEntries[Meta], .BlobHash = Key});
+											MovedEntries.back().Entry.Flags |= MetadataEntry::kTombStone;
+											m_Store.m_MetadataEntries[Meta]		  = {};
+											m_Store.m_BlobEntries[Index].Metadata = {};
+										}
+									}
+								}
+
 								m_Store.m_MetadatalogFile.Append(MovedEntries);
+
 								Stats.RemovedDisk += FreedDiskSpace;
 								if (Ctx.IsCancelledFlag.load())
 								{
diff --git a/src/zenstore/cache/cachedisklayer.cpp b/src/zenstore/cache/cachedisklayer.cpp
index d26c6dfd2..f76ad5c7d 100644
--- a/src/zenstore/cache/cachedisklayer.cpp
+++ b/src/zenstore/cache/cachedisklayer.cpp
@@ -2952,10 +2952,12 @@ public:
 						m_Bucket.m_BlockStore.CompactBlocks(
 							BlockCompactState,
 							m_Bucket.m_Configuration.PayloadAlignment,
-							[&](const BlockStore::MovedChunksArray& MovedArray, uint64_t FreedDiskSpace) {
+							[&](const BlockStore::MovedChunksArray& MovedArray,
+								const BlockStore::ChunkIndexArray&	ScrubbedArray,
+								uint64_t							FreedDiskSpace) {
 								std::vector<DiskIndexEntry> MovedEntries;
 								MovedEntries.reserve(MovedArray.size());
-								RwLock::ExclusiveLockScope _(m_Bucket.m_IndexLock);
+								RwLock::ExclusiveLockScope IndexLock(m_Bucket.m_IndexLock);
 								for (const std::pair<size_t, BlockStoreLocation>& Moved : MovedArray)
 								{
 									size_t		  ChunkIndex = Moved.first;
@@ -2977,6 +2979,24 @@ public:
 										MovedEntries.push_back({.Key = Key, .Location = Payload.Location});
 									}
 								}
+
+								for (size_t ScrubbedIndex : ScrubbedArray)
+								{
+									const IoHash& Key = BlockCompactStateKeys[ScrubbedIndex];
+
+									if (auto It = m_Bucket.m_Index.find(Key); It != m_Bucket.m_Index.end())
+									{
+										BucketPayload& Payload	= m_Bucket.m_Payloads[It->second];
+										DiskLocation   Location = Payload.Location;
+
+										m_Bucket.RemoveMemCachedData(IndexLock, Payload);
+										m_Bucket.RemoveMetaData(IndexLock, Payload);
+
+										Location.Flags |= DiskLocation::kTombStone;
+										MovedEntries.push_back(DiskIndexEntry{.Key = Key, .Location = Location});
+									}
+								}
+
 								m_Bucket.m_SlogFile.Append(MovedEntries);
 								Stats.RemovedDisk += FreedDiskSpace;
 								if (Ctx.IsCancelledFlag.load())
diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index 730bdf143..75562176e 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -718,7 +718,9 @@ public:
 						m_CasContainerStrategy.m_BlockStore.CompactBlocks(
 							BlockCompactState,
 							m_CasContainerStrategy.m_PayloadAlignment,
-							[&](const BlockStore::MovedChunksArray& MovedArray, uint64_t FreedDiskSpace) {
+							[&](const BlockStore::MovedChunksArray& MovedArray,
+								const BlockStore::ChunkIndexArray&	ScrubbedArray,
+								uint64_t							FreedDiskSpace) {
 								std::vector<CasDiskIndexEntry> MovedEntries;
 								RwLock::ExclusiveLockScope	   _(m_CasContainerStrategy.m_LocationMapLock);
 								for (const std::pair<size_t, BlockStoreLocation>& Moved : MovedArray)
@@ -743,6 +745,18 @@ public:
 										MovedEntries.push_back(CasDiskIndexEntry{.Key = Key, .Location = Location});
 									}
 								}
+								for (size_t ScrubbedIndex : ScrubbedArray)
+								{
+									const IoHash& Key = BlockCompactStateKeys[ScrubbedIndex];
+									if (auto It = m_CasContainerStrategy.m_LocationMap.find(Key);
+										It != m_CasContainerStrategy.m_LocationMap.end())
+									{
+										BlockStoreDiskLocation& Location = m_CasContainerStrategy.m_Locations[It->second];
+										MovedEntries.push_back(
+											CasDiskIndexEntry{.Key = Key, .Location = Location, .Flags = CasDiskIndexEntry::kTombstone});
+										m_CasContainerStrategy.m_LocationMap.erase(It);
+									}
+								}
 								m_CasContainerStrategy.m_CasLog.Append(MovedEntries);
 								Stats.RemovedDisk += FreedDiskSpace;
 								if (Ctx.IsCancelledFlag.load())
diff --git a/src/zenstore/include/zenstore/blockstore.h b/src/zenstore/include/zenstore/blockstore.h
index b0713fea1..8cbcad11b 100644
--- a/src/zenstore/include/zenstore/blockstore.h
+++ b/src/zenstore/include/zenstore/blockstore.h
@@ -127,7 +127,8 @@ public:
 	typedef std::vector<std::pair<size_t, BlockStoreLocation>> MovedChunksArray;
 	typedef std::vector<size_t>								   ChunkIndexArray;
 
-	typedef std::function<bool(const MovedChunksArray& MovedChunks, uint64_t FreedDiskSpace)>			 CompactCallback;
+	typedef std::function<bool(const MovedChunksArray& MovedChunks, const ChunkIndexArray& ScrubbedChunks, uint64_t FreedDiskSpace)>
+																										 CompactCallback;
 	typedef std::function<uint64_t()>																	 ClaimDiskReserveCallback;
 	typedef std::function<bool(size_t ChunkIndex, const void* Data, uint64_t Size)>						 IterateChunksSmallSizeCallback;
 	typedef std::function<bool(size_t ChunkIndex, BlockStoreFile& File, uint64_t Offset, uint64_t Size)> IterateChunksLargeSizeCallback;
@@ -173,7 +174,7 @@ public:
 	void CompactBlocks(
 		const BlockStoreCompactState&	CompactState,
 		uint32_t						PayloadAlignment,
-		const CompactCallback&			ChangeCallback		= [](const MovedChunksArray&, uint64_t) { return true; },
+		const CompactCallback&			ChangeCallback = [](const MovedChunksArray&, const ChunkIndexArray&, uint64_t) { return true; },
 		const ClaimDiskReserveCallback& DiskReserveCallback = []() { return 0; },
 		std::string_view				LogPrefix			= {});