Fix curruption of disk cache bucket index on GC (#448)

* make sure we hold the index lock when reading payload data in reclaim space
* don't use index snapshot when updating index in reclaim space
* check that things have not moved under our feet
* don't touch m_Payloads without a lock
* start write block index on the highest block index
* we don't need to bump writeblockindex when stopping write to a block, we will bump appropriately when we start a new block
* changelog

1 files changed, 32 insertions, 26 deletions

diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index e6383c3a1..715704c2e 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -436,10 +436,10 @@ CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
 
 		RwLock::SharedLockScope ___(m_LocationMapLock);
 		Stopwatch				Timer;
-		const auto				____ = MakeGuard([&Timer, &WriteBlockTimeUs, &WriteBlockLongestTimeUs] {
+		const auto				____ = MakeGuard([&Timer, &ReadBlockTimeUs, &ReadBlockLongestTimeUs] {
 			 uint64_t ElapsedUs = Timer.GetElapsedTimeUs();
-			 WriteBlockTimeUs += ElapsedUs;
-			 WriteBlockLongestTimeUs = std::max(ElapsedUs, WriteBlockLongestTimeUs);
+			 ReadBlockTimeUs += ElapsedUs;
+			 ReadBlockLongestTimeUs = std::max(ElapsedUs, ReadBlockLongestTimeUs);
 		 });
 		LocationMap					 = m_LocationMap;
 		Locations					 = m_Locations;
@@ -496,41 +496,47 @@ CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
 		[&](const BlockStore::MovedChunksArray& MovedChunks, const BlockStore::ChunkIndexArray& RemovedChunks) {
 			std::vector<CasDiskIndexEntry> LogEntries;
 			LogEntries.reserve(MovedChunks.size() + RemovedChunks.size());
-			for (const auto& Entry : MovedChunks)
-			{
-				size_t					  ChunkIndex  = Entry.first;
-				const BlockStoreLocation& NewLocation = Entry.second;
-				const IoHash&			  ChunkHash	  = ChunkIndexToChunkHash[ChunkIndex];
-				LogEntries.push_back({.Key = ChunkHash, .Location = {NewLocation, m_PayloadAlignment}});
-			}
-			for (const size_t ChunkIndex : RemovedChunks)
-			{
-				const IoHash&				  ChunkHash		  = ChunkIndexToChunkHash[ChunkIndex];
-				const BlockStoreDiskLocation& OldDiskLocation = Locations[LocationMap[ChunkHash]];
-				LogEntries.push_back({.Key = ChunkHash, .Location = OldDiskLocation, .Flags = CasDiskIndexEntry::kTombstone});
-				DeletedChunks.push_back(ChunkHash);
-			}
-
-			m_CasLog.Append(LogEntries);
-			m_CasLog.Flush();
 			{
 				RwLock::ExclusiveLockScope __(m_LocationMapLock);
 				Stopwatch				   Timer;
 				const auto				   ____ = MakeGuard([&] {
 					uint64_t ElapsedUs = Timer.GetElapsedTimeUs();
-					ReadBlockTimeUs += ElapsedUs;
-					ReadBlockLongestTimeUs = std::max(ElapsedUs, ReadBlockLongestTimeUs);
+					WriteBlockTimeUs += ElapsedUs;
+					WriteBlockLongestTimeUs = std::max(ElapsedUs, WriteBlockLongestTimeUs);
 				});
-				for (const CasDiskIndexEntry& Entry : LogEntries)
+				for (const auto& Entry : MovedChunks)
 				{
-					if (Entry.Flags & CasDiskIndexEntry::kTombstone)
+					size_t					  ChunkIndex	= Entry.first;
+					const BlockStoreLocation& NewLocation	= Entry.second;
+					const IoHash&			  ChunkHash		= ChunkIndexToChunkHash[ChunkIndex];
+					size_t					  LocationIndex = m_LocationMap[ChunkHash];
+					BlockStoreDiskLocation&	  Location		= m_Locations[LocationIndex];
+					if (Locations[LocationMap[ChunkHash]] != Location)
 					{
-						m_LocationMap.erase(Entry.Key);
+						// Entry has been updated while GC was running, ignore the move
 						continue;
 					}
-					m_Locations[m_LocationMap[Entry.Key]] = Entry.Location;
+					Location = {NewLocation, m_PayloadAlignment};
+					LogEntries.push_back({.Key = ChunkHash, .Location = Location});
+				}
+				for (const size_t ChunkIndex : RemovedChunks)
+				{
+					const IoHash&				  ChunkHash		= ChunkIndexToChunkHash[ChunkIndex];
+					size_t						  LocationIndex = m_LocationMap[ChunkHash];
+					const BlockStoreDiskLocation& Location		= Locations[LocationIndex];
+					if (Locations[LocationMap[ChunkHash]] != Location)
+					{
+						// Entry has been updated while GC was running, ignore the delete
+						continue;
+					}
+					LogEntries.push_back({.Key = ChunkHash, .Location = Location, .Flags = CasDiskIndexEntry::kTombstone});
+					m_LocationMap.erase(ChunkHash);
+					DeletedChunks.push_back(ChunkHash);
 				}
 			}
+
+			m_CasLog.Append(LogEntries);
+			m_CasLog.Flush();
 		},
 		[&GcCtx]() { return GcCtx.CollectSmallObjects(); });