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

4 files changed, 93 insertions, 75 deletions

diff --git a/src/zenserver/cache/cachedisklayer.cpp b/src/zenserver/cache/cachedisklayer.cpp
index fafc6bbee..894676d6a 100644
--- a/src/zenserver/cache/cachedisklayer.cpp
+++ b/src/zenserver/cache/cachedisklayer.cpp
@@ -860,7 +860,7 @@ ZenCacheDiskLayer::CacheBucket::Flush()
 		Payloads	= m_Payloads;
 		AccessTimes = m_AccessTimes;
 	}
-	SaveManifest(MakeManifest(std::move(Index), std::move(AccessTimes), std::move(Payloads)));
+	SaveManifest(MakeManifest(std::move(Index), std::move(AccessTimes), Payloads));
 }
 
 void
@@ -878,7 +878,9 @@ ZenCacheDiskLayer::CacheBucket::SaveManifest(CbObject&& Manifest)
 }
 
 CbObject
-ZenCacheDiskLayer::CacheBucket::MakeManifest(IndexMap&& Index, std::vector<AccessTime>&& AccessTimes, std::vector<BucketPayload>&& Payloads)
+ZenCacheDiskLayer::CacheBucket::MakeManifest(IndexMap&&						   Index,
+											 std::vector<AccessTime>&&		   AccessTimes,
+											 const std::vector<BucketPayload>& Payloads)
 {
 	using namespace std::literals;
 
@@ -1405,7 +1407,7 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 			Payloads	= m_Payloads;
 			AccessTimes = m_AccessTimes;
 		}
-		SaveManifest(MakeManifest(std::move(Index), std::move(AccessTimes), std::move(Payloads)));
+		SaveManifest(MakeManifest(std::move(Index), std::move(AccessTimes), Payloads));
 	});
 
 	m_SlogFile.Flush();
@@ -1464,6 +1466,7 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 
 	std::vector<DiskIndexEntry>		 ExpiredStandaloneEntries;
 	IndexMap						 Index;
+	std::vector<BucketPayload>		 Payloads;
 	BlockStore::ReclaimSnapshotState BlockStoreState;
 	{
 		bool Expected = false;
@@ -1474,8 +1477,7 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 		}
 		auto FlushingGuard = MakeGuard([&] { m_IsFlushing.store(false); });
 
-		std::vector<AccessTime>	   AccessTimes;
-		std::vector<BucketPayload> Payloads;
+		std::vector<AccessTime> AccessTimes;
 		{
 			ZEN_TRACE_CPU("Z$::Disk::Bucket::CollectGarbage::State");
 			RwLock::SharedLockScope IndexLock(m_IndexLock);
@@ -1502,7 +1504,7 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 			{
 				if (auto It = Index.find(Key); It != Index.end())
 				{
-					const BucketPayload& Payload = m_Payloads[It->second];
+					const BucketPayload& Payload = Payloads[It->second];
 					DiskIndexEntry		 Entry	 = {.Key = It->first, .Location = Payload.Location};
 					if (Entry.Location.Flags & DiskLocation::kStandaloneFile)
 					{
@@ -1522,7 +1524,7 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 				m_SlogFile.Append(ExpiredStandaloneEntries);
 			}
 		}
-		SaveManifest(MakeManifest(std::move(Index), std::move(AccessTimes), std::move(Payloads)));
+		SaveManifest(MakeManifest(std::move(Index), std::move(AccessTimes), Payloads));
 	}
 
 	if (GcCtx.IsDeletionMode())
@@ -1600,15 +1602,17 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 
 	std::vector<IoHash> TotalChunkHashes;
 	TotalChunkHashes.reserve(TotalChunkCount);
-	for (const auto& Entry : Index)
 	{
-		const DiskLocation& Location = m_Payloads[Entry.second].Location;
-
-		if (Location.Flags & DiskLocation::kStandaloneFile)
+		for (const auto& Entry : Index)
 		{
-			continue;
+			const DiskLocation& Location = Payloads[Entry.second].Location;
+
+			if (Location.Flags & DiskLocation::kStandaloneFile)
+			{
+				continue;
+			}
+			TotalChunkHashes.push_back(Entry.first);
 		}
-		TotalChunkHashes.push_back(Entry.first);
 	}
 
 	if (TotalChunkHashes.empty())
@@ -1626,7 +1630,7 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 
 	GcCtx.FilterCids(TotalChunkHashes, [&](const IoHash& ChunkHash, bool Keep) {
 		auto				KeyIt		 = Index.find(ChunkHash);
-		const DiskLocation& DiskLocation = m_Payloads[KeyIt->second].Location;
+		const DiskLocation& DiskLocation = Payloads[KeyIt->second].Location;
 		BlockStoreLocation	Location	 = DiskLocation.GetBlockLocation(m_PayloadAlignment);
 		size_t				ChunkIndex	 = ChunkLocations.size();
 		ChunkLocations.push_back(Location);
@@ -1661,48 +1665,51 @@ ZenCacheDiskLayer::CacheBucket::CollectGarbage(GcContext& GcCtx)
 		[&](const BlockStore::MovedChunksArray& MovedChunks, const BlockStore::ChunkIndexArray& RemovedChunks) {
 			std::vector<DiskIndexEntry> 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];
-				const BucketPayload&	  OldPayload	  = m_Payloads[Index[ChunkHash]];
-				const DiskLocation&		  OldDiskLocation = OldPayload.Location;
-				LogEntries.push_back(
-					{.Key = ChunkHash, .Location = DiskLocation(NewLocation, m_PayloadAlignment, OldDiskLocation.GetFlags())});
-			}
-			for (const size_t ChunkIndex : RemovedChunks)
-			{
-				const IoHash&		 ChunkHash		 = ChunkIndexToChunkHash[ChunkIndex];
-				const BucketPayload& OldPayload		 = m_Payloads[Index[ChunkHash]];
-				const DiskLocation&	 OldDiskLocation = OldPayload.Location;
-				LogEntries.push_back({.Key		= ChunkHash,
-									  .Location = DiskLocation(OldDiskLocation.GetBlockLocation(m_PayloadAlignment),
-															   m_PayloadAlignment,
-															   OldDiskLocation.GetFlags() | DiskLocation::kTombStone)});
-				DeletedChunks.insert(ChunkHash);
-			}
-
-			m_SlogFile.Append(LogEntries);
-			m_SlogFile.Flush();
 			{
 				RwLock::ExclusiveLockScope __(m_IndexLock);
 				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 DiskIndexEntry& Entry : LogEntries)
+				for (const auto& Entry : MovedChunks)
 				{
-					if (Entry.Location.GetFlags() & DiskLocation::kTombStone)
+					size_t					  ChunkIndex   = Entry.first;
+					const BlockStoreLocation& NewLocation  = Entry.second;
+					const IoHash&			  ChunkHash	   = ChunkIndexToChunkHash[ChunkIndex];
+					size_t					  PayloadIndex = m_Index[ChunkHash];
+					BucketPayload&			  Payload	   = m_Payloads[PayloadIndex];
+					if (Payloads[Index[ChunkHash]].Location != m_Payloads[PayloadIndex].Location)
 					{
-						m_Index.erase(Entry.Key);
+						// Entry has been updated while GC was running, ignore the move
 						continue;
 					}
-					m_Payloads[m_Index[Entry.Key]].Location = Entry.Location;
+					Payload.Location = DiskLocation(NewLocation, m_PayloadAlignment, Payload.Location.GetFlags());
+					LogEntries.push_back({.Key = ChunkHash, .Location = Payload.Location});
+				}
+				for (const size_t ChunkIndex : RemovedChunks)
+				{
+					const IoHash&		 ChunkHash	  = ChunkIndexToChunkHash[ChunkIndex];
+					size_t				 PayloadIndex = m_Index[ChunkHash];
+					const BucketPayload& Payload	  = m_Payloads[PayloadIndex];
+					if (Payloads[Index[ChunkHash]].Location != Payload.Location)
+					{
+						// Entry has been updated while GC was running, ignore the delete
+						continue;
+					}
+					const DiskLocation& OldDiskLocation = Payload.Location;
+					LogEntries.push_back({.Key		= ChunkHash,
+										  .Location = DiskLocation(OldDiskLocation.GetBlockLocation(m_PayloadAlignment),
+																   m_PayloadAlignment,
+																   OldDiskLocation.GetFlags() | DiskLocation::kTombStone)});
+					m_Index.erase(ChunkHash);
+					DeletedChunks.insert(ChunkHash);
 				}
 			}
+
+			m_SlogFile.Append(LogEntries);
+			m_SlogFile.Flush();
 		},
 		[&]() { return GcCtx.CollectSmallObjects(); });
 }
diff --git a/src/zenserver/cache/cachedisklayer.h b/src/zenserver/cache/cachedisklayer.h
index c4bedfee8..62163100d 100644
--- a/src/zenserver/cache/cachedisklayer.h
+++ b/src/zenserver/cache/cachedisklayer.h
@@ -32,6 +32,8 @@ struct DiskLocation
 		this->Location.BlockLocation = BlockStoreDiskLocation(Location, PayloadAlignment);
 	}
 
+	inline bool operator!=(const DiskLocation& Rhs) const { return memcmp(&Location, &Rhs.Location, sizeof(Location)) != 0; }
+
 	inline BlockStoreLocation GetBlockLocation(uint64_t PayloadAlignment) const
 	{
 		ZEN_ASSERT(!(Flags & kStandaloneFile));
@@ -231,7 +233,7 @@ private:
 		uint64_t ReadIndexFile(const std::filesystem::path& IndexPath, uint32_t& OutVersion);
 		uint64_t ReadLog(const std::filesystem::path& LogPath, uint64_t LogPosition);
 		void	 OpenLog(const bool IsNew);
-		CbObject MakeManifest(IndexMap&& Index, std::vector<AccessTime>&& AccessTimes, std::vector<BucketPayload>&& Payloads);
+		CbObject MakeManifest(IndexMap&& Index, std::vector<AccessTime>&& AccessTimes, const std::vector<BucketPayload>& Payloads);
 		void	 SaveManifest(CbObject&& Manifest);
 		CacheValueDetails::ValueDetails GetValueDetails(const IoHash& Key, size_t Index) const;
 		// These locks are here to avoid contention on file creation, therefore it's sufficient
diff --git a/src/zenstore/blockstore.cpp b/src/zenstore/blockstore.cpp
index f99b0bc4a..968e919d6 100644
--- a/src/zenstore/blockstore.cpp
+++ b/src/zenstore/blockstore.cpp
@@ -171,6 +171,7 @@ BlockStore::Initialize(const std::filesystem::path& BlocksBasePath, uint64_t Max
 
 	if (std::filesystem::is_directory(m_BlocksBasePath))
 	{
+		uint32_t						   NextBlockIndex = 0;
 		std::vector<std::filesystem::path> FoldersToScan;
 		FoldersToScan.push_back(m_BlocksBasePath);
 		size_t FolderOffset = 0;
@@ -202,10 +203,15 @@ BlockStore::Initialize(const std::filesystem::path& BlocksBasePath, uint64_t Max
 					m_TotalSize.fetch_add(BlockFile->FileSize(), std::memory_order::relaxed);
 					m_ChunkBlocks[BlockIndex] = BlockFile;
 					FoundBlocks[BlockIndex]	  = BlockFile->FileSize();
+					if (BlockIndex >= NextBlockIndex)
+					{
+						NextBlockIndex = (BlockIndex + 1) & (m_MaxBlockCount - 1);
+					}
 				}
 			}
 			++FolderOffset;
 		}
+		m_WriteBlockIndex.store(NextBlockIndex, std::memory_order_release);
 	}
 	else
 	{
@@ -363,14 +369,11 @@ BlockStore::Flush(bool ForceNewBlock)
 		RwLock::ExclusiveLockScope _(m_InsertLock);
 		if (m_CurrentInsertOffset > 0)
 		{
-			uint32_t WriteBlockIndex = m_WriteBlockIndex.load(std::memory_order_acquire);
-			WriteBlockIndex			 = (WriteBlockIndex + 1) & (m_MaxBlockCount - 1);
 			if (m_WriteBlock)
 			{
 				m_WriteBlock->Flush();
 			}
-			m_WriteBlock = nullptr;
-			m_WriteBlockIndex.store(WriteBlockIndex, std::memory_order_release);
+			m_WriteBlock		  = nullptr;
 			m_CurrentInsertOffset = 0;
 		}
 		return;
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(); });