Merge branch 'main' into de/zen-service-command

1 files changed, 402 insertions, 240 deletions

diff --git a/src/zenstore/cache/cachedisklayer.cpp b/src/zenstore/cache/cachedisklayer.cpp
index 61552fafc..15a1c9650 100644
--- a/src/zenstore/cache/cachedisklayer.cpp
+++ b/src/zenstore/cache/cachedisklayer.cpp
@@ -14,6 +14,7 @@
 #include <zencore/trace.h>
 #include <zencore/workthreadpool.h>
 #include <zencore/xxhash.h>
+#include <zenutil/parallelwork.h>
 #include <zenutil/referencemetadata.h>
 #include <zenutil/workerpools.h>
 
@@ -195,34 +196,33 @@ namespace cache::impl {
 		return true;
 	}
 
-	bool MoveAndDeleteDirectory(const std::filesystem::path& Dir)
+	std::filesystem::path MoveDroppedDirectory(const std::filesystem::path& Dir)
 	{
 		int DropIndex = 0;
 		do
 		{
-			if (!std::filesystem::exists(Dir))
+			if (!IsDir(Dir))
 			{
-				return false;
+				return {};
 			}
 
 			std::string			  DroppedName		= fmt::format("[dropped]{}({})", Dir.filename().string(), DropIndex);
 			std::filesystem::path DroppedBucketPath = Dir.parent_path() / DroppedName;
-			if (std::filesystem::exists(DroppedBucketPath))
+			if (IsDir(DroppedBucketPath))
 			{
 				DropIndex++;
 				continue;
 			}
 
 			std::error_code Ec;
-			std::filesystem::rename(Dir, DroppedBucketPath, Ec);
+			RenameDirectory(Dir, DroppedBucketPath, Ec);
 			if (!Ec)
 			{
-				DeleteDirectories(DroppedBucketPath);
-				return true;
+				return DroppedBucketPath;
 			}
-			// TODO: Do we need to bail at some point?
 			zen::Sleep(100);
-		} while (true);
+		} while (DropIndex < 10);
+		return {};
 	}
 }  // namespace cache::impl
 
@@ -318,10 +318,10 @@ private:
 #pragma pack(4)
 	struct ManifestData
 	{
-		uint32_t   RawSize;	   // 4
-		AccessTime Timestamp;  // 4
-		IoHash	   RawHash;	   // 20
-		IoHash	   Key;		   // 20
+		uint32_t RawSize;			 // 4
+		uint32_t SecondsSinceEpoch;	 // 4
+		IoHash	 RawHash;			 // 20
+		IoHash	 Key;				 // 20
 	};
 #pragma pack(pop)
 
@@ -603,7 +603,7 @@ BucketManifestSerializer::ReadSidecarFile(RwLock::ExclusiveLockScope&								  B
 
 			ZenCacheDiskLayer::CacheBucket::BucketPayload& PayloadEntry = Payloads[PlIndex];
 
-			AccessTimes[PlIndex] = Entry->Timestamp;
+			AccessTimes[PlIndex].SetSecondsSinceEpoch(Entry->SecondsSinceEpoch);
 
 			if (Entry->RawSize && Entry->RawHash != IoHash::Zero)
 			{
@@ -630,6 +630,16 @@ BucketManifestSerializer::WriteSidecarFile(RwLock::SharedLockScope&,
 {
 	ZEN_TRACE_CPU("Z$::WriteSidecarFile");
 
+	ZEN_DEBUG("writing store sidecar for '{}'", SidecarPath);
+	const uint64_t EntryCount = Index.size();
+	Stopwatch	   Timer;
+	const auto	   _ = MakeGuard([&] {
+		ZEN_INFO("wrote store sidecar for '{}' containing {} entries in {}",
+				 SidecarPath,
+				 EntryCount,
+				 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+	});
+
 	BucketMetaHeader Header;
 	Header.EntryCount  = m_ManifestEntryCount;
 	Header.LogPosition = SnapshotLogPosition;
@@ -647,43 +657,44 @@ BucketManifestSerializer::WriteSidecarFile(RwLock::SharedLockScope&,
 
 	SidecarFile.Write(&Header, sizeof Header, 0);
 
-	// TODO: make this batching for better performance
 	{
 		uint64_t WriteOffset = sizeof Header;
 
-		//		BasicFileWriter SidecarWriter(SidecarFile, 128 * 1024);
+		const size_t MaxManifestDataBufferCount = (512u * 1024u) / sizeof(ManifestData);
 
-		std::vector<ManifestData> ManifestDataBuffer;
-		const size_t			  MaxManifestDataBufferCount = Min(Index.size(), 8192u);  // 512 Kb
-		ManifestDataBuffer.reserve(MaxManifestDataBufferCount);
+		std::vector<ManifestData> ManifestDataBuffer(Min(m_ManifestEntryCount, MaxManifestDataBufferCount));
+		auto					  WriteIt = ManifestDataBuffer.begin();
 		for (auto& Kv : Index)
 		{
-			const IoHash&	   Key	   = Kv.first;
-			const PayloadIndex PlIndex = Kv.second;
+			ManifestData& Data = *WriteIt++;
 
-			IoHash	 RawHash = IoHash::Zero;
-			uint32_t RawSize = 0;
+			const PayloadIndex PlIndex = Kv.second;
+			Data.Key				   = Kv.first;
+			Data.SecondsSinceEpoch	   = AccessTimes[PlIndex].GetSecondsSinceEpoch();
 
 			if (const MetaDataIndex MetaIndex = Payloads[PlIndex].MetaData)
 			{
-				RawHash = MetaDatas[MetaIndex].RawHash;
-				RawSize = MetaDatas[MetaIndex].RawSize;
+				Data.RawHash = MetaDatas[MetaIndex].RawHash;
+				Data.RawSize = MetaDatas[MetaIndex].RawSize;
+			}
+			else
+			{
+				Data.RawHash = IoHash::Zero;
+				Data.RawSize = 0;
 			}
 
-			ManifestDataBuffer.emplace_back(
-				ManifestData{.RawSize = RawSize, .Timestamp = AccessTimes[PlIndex], .RawHash = RawHash, .Key = Key});
-			if (ManifestDataBuffer.size() == MaxManifestDataBufferCount)
+			if (WriteIt == ManifestDataBuffer.end())
 			{
-				const uint64_t WriteSize = sizeof(ManifestData) * ManifestDataBuffer.size();
+				uint64_t WriteSize = std::distance(ManifestDataBuffer.begin(), WriteIt) * sizeof(ManifestData);
 				SidecarFile.Write(ManifestDataBuffer.data(), WriteSize, WriteOffset);
 				WriteOffset += WriteSize;
-				ManifestDataBuffer.clear();
-				ManifestDataBuffer.reserve(MaxManifestDataBufferCount);
+				WriteIt = ManifestDataBuffer.begin();
 			}
 		}
-		if (ManifestDataBuffer.size() > 0)
+		if (WriteIt != ManifestDataBuffer.begin())
 		{
-			SidecarFile.Write(ManifestDataBuffer.data(), sizeof(ManifestData) * ManifestDataBuffer.size(), WriteOffset);
+			uint64_t WriteSize = std::distance(ManifestDataBuffer.begin(), WriteIt) * sizeof(ManifestData);
+			SidecarFile.Write(ManifestDataBuffer.data(), WriteSize, WriteOffset);
 		}
 	}
 
@@ -740,6 +751,16 @@ ZenCacheDiskLayer::CacheBucket::CacheBucket(GcManager&				   Gc,
 
 ZenCacheDiskLayer::CacheBucket::~CacheBucket()
 {
+	try
+	{
+		m_SlogFile.Flush();
+		m_SlogFile.Close();
+		m_BlockStore.Close();
+	}
+	catch (const std::exception& Ex)
+	{
+		ZEN_ERROR("~CacheBucket() failed with: ", Ex.what());
+	}
 	m_Gc.RemoveGcReferencer(*this);
 }
 
@@ -813,12 +834,13 @@ ZenCacheDiskLayer::CacheBucket::OpenOrCreate(std::filesystem::path BucketDir, bo
 }
 
 void
-ZenCacheDiskLayer::CacheBucket::WriteIndexSnapshotLocked(bool FlushLockPosition, const std::function<uint64_t()>& ClaimDiskReserveFunc)
+ZenCacheDiskLayer::CacheBucket::WriteIndexSnapshotLocked(uint64_t						  LogPosition,
+														 bool							  ResetLog,
+														 const std::function<uint64_t()>& ClaimDiskReserveFunc)
 {
 	ZEN_TRACE_CPU("Z$::Bucket::WriteIndexSnapshot");
 
-	const uint64_t LogCount = FlushLockPosition ? 0 : m_SlogFile.GetLogCount();
-	if (m_LogFlushPosition == LogCount)
+	if (m_LogFlushPosition == LogPosition)
 	{
 		return;
 	}
@@ -835,7 +857,7 @@ ZenCacheDiskLayer::CacheBucket::WriteIndexSnapshotLocked(bool FlushLockPosition,
 
 	namespace fs = std::filesystem;
 
-	fs::path IndexPath = cache::impl::GetIndexPath(m_BucketDir, m_BucketName);
+	const fs::path IndexPath = cache::impl::GetIndexPath(m_BucketDir, m_BucketName);
 
 	try
 	{
@@ -867,66 +889,70 @@ ZenCacheDiskLayer::CacheBucket::WriteIndexSnapshotLocked(bool FlushLockPosition,
 			throw std::system_error(Ec, fmt::format("failed to create new snapshot file in '{}'", m_BucketDir));
 		}
 
-		{
-			// This is in a separate scope just to ensure IndexWriter goes out
-			// of scope before the file is flushed/closed, in order to ensure
-			// all data is written to the file
-			BasicFileWriter IndexWriter(ObjectIndexFile, 128 * 1024);
+		const uint64_t IndexLogPosition = ResetLog ? 0 : LogPosition;
 
-			cache::impl::CacheBucketIndexHeader Header = {.EntryCount		= EntryCount,
-														  .LogPosition		= LogCount,
-														  .PayloadAlignment = gsl::narrow<uint32_t>(m_Configuration.PayloadAlignment)};
+		cache::impl::CacheBucketIndexHeader Header = {.EntryCount		= EntryCount,
+													  .LogPosition		= IndexLogPosition,
+													  .PayloadAlignment = gsl::narrow<uint32_t>(m_Configuration.PayloadAlignment)};
 
-			Header.Checksum = cache::impl::CacheBucketIndexHeader::ComputeChecksum(Header);
-			IndexWriter.Write(&Header, sizeof(cache::impl::CacheBucketIndexHeader), 0);
+		Header.Checksum = cache::impl::CacheBucketIndexHeader::ComputeChecksum(Header);
+		ObjectIndexFile.Write(&Header, sizeof(cache::impl::CacheBucketIndexHeader), 0);
 
+		if (EntryCount > 0)
+		{
 			uint64_t IndexWriteOffset = sizeof(cache::impl::CacheBucketIndexHeader);
 
+			size_t						MaxWriteEntryCount = (512u * 1024u) / sizeof(DiskIndexEntry);
+			std::vector<DiskIndexEntry> DiskEntryBuffer(Min(m_Index.size(), MaxWriteEntryCount));
+
+			auto WriteIt = DiskEntryBuffer.begin();
 			for (auto& Entry : m_Index)
 			{
-				DiskIndexEntry IndexEntry;
-				IndexEntry.Key		= Entry.first;
-				IndexEntry.Location = m_Payloads[Entry.second].Location;
-				IndexWriter.Write(&IndexEntry, sizeof(DiskIndexEntry), IndexWriteOffset);
-
-				IndexWriteOffset += sizeof(DiskIndexEntry);
+				*WriteIt++ = {.Key = Entry.first, .Location = m_Payloads[Entry.second].Location};
+				if (WriteIt == DiskEntryBuffer.end())
+				{
+					uint64_t WriteSize = std::distance(DiskEntryBuffer.begin(), WriteIt) * sizeof(DiskIndexEntry);
+					ObjectIndexFile.Write(DiskEntryBuffer.data(), WriteSize, IndexWriteOffset);
+					IndexWriteOffset += WriteSize;
+					WriteIt = DiskEntryBuffer.begin();
+				}
 			}
 
-			IndexWriter.Flush();
+			if (WriteIt != DiskEntryBuffer.begin())
+			{
+				uint64_t WriteSize = std::distance(DiskEntryBuffer.begin(), WriteIt) * sizeof(DiskIndexEntry);
+				ObjectIndexFile.Write(DiskEntryBuffer.data(), WriteSize, IndexWriteOffset);
+			}
 		}
 
 		ObjectIndexFile.Flush();
 		ObjectIndexFile.MoveTemporaryIntoPlace(IndexPath, Ec);
 		if (Ec)
 		{
-			std::filesystem::path TempFilePath = ObjectIndexFile.GetPath();
-			ZEN_WARN("snapshot failed to rename new snapshot '{}' to '{}', reason: '{}'", TempFilePath, IndexPath, Ec.message());
+			throw std::system_error(Ec,
+									fmt::format("Snapshot failed to rename new snapshot '{}' to '{}', reason: '{}'",
+												ObjectIndexFile.GetPath(),
+												IndexPath,
+												Ec.message()));
 		}
-		else
+
+		if (ResetLog)
 		{
-			// We must only update the log flush position once the snapshot write succeeds
-			if (FlushLockPosition)
-			{
-				std::filesystem::path LogPath = cache::impl::GetLogPath(m_BucketDir, m_BucketName);
+			const std::filesystem::path LogPath = cache::impl::GetLogPath(m_BucketDir, m_BucketName);
 
-				if (std::filesystem::is_regular_file(LogPath))
+			if (IsFile(LogPath))
+			{
+				m_SlogFile.Close();
+				if (!RemoveFile(LogPath, Ec) || Ec)
 				{
-					if (!std::filesystem::remove(LogPath, Ec) || Ec)
-					{
-						ZEN_WARN("snapshot failed to clean log file '{}', removing index at '{}', reason: '{}'",
-								 LogPath,
-								 IndexPath,
-								 Ec.message());
-						std::error_code RemoveIndexEc;
-						std::filesystem::remove(IndexPath, RemoveIndexEc);
-					}
+					// This is non-critical, it only means that we will replay the events of the log over the snapshot - inefficent but in
+					// the end it will be the same result
+					ZEN_WARN("snapshot failed to clean log file '{}', reason: '{}'", LogPath, IndexPath, Ec.message());
 				}
-			}
-			if (!Ec)
-			{
-				m_LogFlushPosition = LogCount;
+				m_SlogFile.Open(LogPath, CasLogFile::Mode::kWrite);
 			}
 		}
+		m_LogFlushPosition = IndexLogPosition;
 	}
 	catch (const std::exception& Err)
 	{
@@ -939,7 +965,7 @@ ZenCacheDiskLayer::CacheBucket::ReadIndexFile(RwLock::ExclusiveLockScope&, const
 {
 	ZEN_TRACE_CPU("Z$::Bucket::ReadIndexFile");
 
-	if (!std::filesystem::is_regular_file(IndexPath))
+	if (!IsFile(IndexPath))
 	{
 		return 0;
 	}
@@ -1023,7 +1049,7 @@ ZenCacheDiskLayer::CacheBucket::ReadLog(RwLock::ExclusiveLockScope&, const std::
 {
 	ZEN_TRACE_CPU("Z$::Bucket::ReadLog");
 
-	if (!std::filesystem::is_regular_file(LogPath))
+	if (!IsFile(LogPath))
 	{
 		return 0;
 	}
@@ -1103,47 +1129,40 @@ ZenCacheDiskLayer::CacheBucket::InitializeIndexFromDisk(RwLock::ExclusiveLockSco
 
 	if (IsNew)
 	{
-		fs::remove(LogPath);
-		fs::remove(IndexPath);
-		fs::remove_all(m_BlocksBasePath);
+		RemoveFile(LogPath);
+		RemoveFile(IndexPath);
+		DeleteDirectories(m_BlocksBasePath);
 	}
 
 	CreateDirectories(m_BucketDir);
 
 	m_BlockStore.Initialize(m_BlocksBasePath, m_Configuration.MaxBlockSize, BlockStoreDiskLocation::MaxBlockIndex + 1);
 
-	if (std::filesystem::is_regular_file(IndexPath))
+	if (IsFile(IndexPath))
 	{
 		uint32_t IndexVersion = 0;
 		m_LogFlushPosition	  = ReadIndexFile(IndexLock, IndexPath, IndexVersion);
 		if (IndexVersion == 0)
 		{
 			ZEN_WARN("removing invalid index file at '{}'", IndexPath);
-			std::filesystem::remove(IndexPath);
+			RemoveFile(IndexPath);
 		}
 	}
 
 	uint64_t LogEntryCount = 0;
-	if (std::filesystem::is_regular_file(LogPath))
+	if (IsFile(LogPath))
 	{
 		if (TCasLogFile<DiskIndexEntry>::IsValid(LogPath))
 		{
 			LogEntryCount = ReadLog(IndexLock, LogPath, m_LogFlushPosition);
 		}
-		else if (fs::is_regular_file(LogPath))
+		else if (IsFile(LogPath))
 		{
 			ZEN_WARN("removing invalid log at '{}'", LogPath);
-			std::filesystem::remove(LogPath);
+			RemoveFile(LogPath);
 		}
 	}
 
-	if (IsNew || LogEntryCount > 0 || m_LogFlushPosition != 0)
-	{
-		WriteIndexSnapshot(IndexLock, /*Flush log*/ true);
-	}
-
-	m_SlogFile.Open(LogPath, CasLogFile::Mode::kWrite);
-
 	BlockStore::BlockIndexSet KnownBlocks;
 	for (const auto& Entry : m_Index)
 	{
@@ -1161,7 +1180,53 @@ ZenCacheDiskLayer::CacheBucket::InitializeIndexFromDisk(RwLock::ExclusiveLockSco
 			KnownBlocks.insert(BlockIndex);
 		}
 	}
-	m_BlockStore.SyncExistingBlocksOnDisk(KnownBlocks);
+	BlockStore::BlockIndexSet MissingBlocks = m_BlockStore.SyncExistingBlocksOnDisk(KnownBlocks);
+	m_SlogFile.Open(LogPath, CasLogFile::Mode::kWrite);
+
+	bool RemovedEntries = false;
+	if (!MissingBlocks.empty())
+	{
+		std::vector<DiskIndexEntry> MissingEntries;
+
+		for (auto& It : m_Index)
+		{
+			BucketPayload& Payload	= m_Payloads[It.second];
+			DiskLocation   Location = Payload.Location;
+			if (!Location.IsFlagSet(DiskLocation::kStandaloneFile))
+			{
+				if (MissingBlocks.contains(Location.Location.BlockLocation.GetBlockIndex()))
+				{
+					RemoveMemCachedData(IndexLock, Payload);
+					RemoveMetaData(IndexLock, Payload);
+				}
+			}
+			Location.Flags |= DiskLocation::kTombStone;
+			MissingEntries.push_back(DiskIndexEntry{.Key = It.first, .Location = Location});
+		}
+
+		ZEN_ASSERT(!MissingEntries.empty());
+
+		for (const DiskIndexEntry& Entry : MissingEntries)
+		{
+			m_Index.erase(Entry.Key);
+		}
+		m_SlogFile.Append(MissingEntries);
+		m_SlogFile.Flush();
+		{
+			std::vector<BucketPayload>	Payloads;
+			std::vector<AccessTime>		AccessTimes;
+			std::vector<BucketMetaData> MetaDatas;
+			std::vector<MemCacheData>	MemCachedPayloads;
+			IndexMap					Index;
+			CompactState(IndexLock, Payloads, AccessTimes, MetaDatas, MemCachedPayloads, Index);
+		}
+		RemovedEntries = true;
+	}
+
+	if (IsNew || LogEntryCount > 0 || m_LogFlushPosition != 0 || RemovedEntries)
+	{
+		WriteIndexSnapshot(IndexLock, m_SlogFile.GetLogCount(), /*Flush log*/ true);
+	}
 }
 
 void
@@ -1424,6 +1489,13 @@ ZenCacheDiskLayer::CacheBucket::EndGetBatch(GetBatchHandle* Batch) noexcept
 							}
 						}
 					}
+					else
+					{
+						if (m_Configuration.MemCacheSizeThreshold > 0)
+						{
+							m_MemoryMissCount++;
+						}
+					}
 				}
 			}
 
@@ -1432,7 +1504,7 @@ ZenCacheDiskLayer::CacheBucket::EndGetBatch(GetBatchHandle* Batch) noexcept
 			// Often we will find the metadata due to the thread setting the mem cached part doing it before us so it is worth
 			// checking if it is present once more before spending time fetching and setting the RawHash and RawSize in metadata
 
-			auto FillOne = [&](const DiskLocation& Location, size_t KeyIndex, IoBuffer&& Value) {
+			auto FillOne = [&](const DiskLocation& Location, size_t KeyIndex, IoBuffer&& Value, bool UsesTemporaryMemory) {
 				if (!Value)
 				{
 					return;
@@ -1455,6 +1527,12 @@ ZenCacheDiskLayer::CacheBucket::EndGetBatch(GetBatchHandle* Batch) noexcept
 					}
 				}
 
+				if (AddToMemCache || UsesTemporaryMemory)
+				{
+					// We need to own it if we want to add it to the memcache or the buffer is just a range of the block iteration buffer
+					OutValue.Value.MakeOwned();
+				}
+
 				if (SetMetaInfo)
 				{
 					// See ZenCacheDiskLayer::CacheBucket::Get - it sets the memcache part first and then if it needs to it set the
@@ -1526,35 +1604,42 @@ ZenCacheDiskLayer::CacheBucket::EndGetBatch(GetBatchHandle* Batch) noexcept
 			if (!InlineDiskLocations.empty())
 			{
 				ZEN_TRACE_CPU("Z$::Bucket::EndGetBatch::ReadInline");
-				m_BlockStore.IterateChunks(
-					std::span{begin(InlineBlockLocations), end(InlineBlockLocations)},
-					[&](uint32_t, std::span<const size_t> ChunkIndexes) -> bool {
-						// Only read into memory the IoBuffers we could potentially add to memcache
-						const uint64_t LargeChunkSizeLimit = Max(m_Configuration.MemCacheSizeThreshold, 1u * 1024u);
-						m_BlockStore.IterateBlock(
-							std::span{begin(InlineBlockLocations), end(InlineBlockLocations)},
-							ChunkIndexes,
-							[this, &FillOne, &InlineDiskLocations, &InlineKeyIndexes](size_t	  ChunkIndex,
-																					  const void* Data,
-																					  uint64_t	  Size) -> bool {
-								if (Data != nullptr)
-								{
-									FillOne(InlineDiskLocations[ChunkIndex],
-											InlineKeyIndexes[ChunkIndex],
-											IoBufferBuilder::MakeCloneFromMemory(Data, Size));
-								}
-								return true;
-							},
-							[this, &FillOne, &InlineDiskLocations, &InlineKeyIndexes](size_t		  ChunkIndex,
-																					  BlockStoreFile& File,
-																					  uint64_t		  Offset,
-																					  uint64_t		  Size) -> bool {
-								FillOne(InlineDiskLocations[ChunkIndex], InlineKeyIndexes[ChunkIndex], File.GetChunk(Offset, Size));
-								return true;
-							},
-							LargeChunkSizeLimit);
-						return true;
-					});
+				m_BlockStore.IterateChunks(std::span{begin(InlineBlockLocations), end(InlineBlockLocations)},
+										   [&](uint32_t, std::span<const size_t> ChunkIndexes) -> bool {
+											   // Up to 8KB or m_Configuration.MemCacheSizeThreshold depending on configuration
+											   const uint64_t LargeChunkSizeLimit =
+												   m_Configuration.MemCacheSizeThreshold == 0
+													   ? Min(m_Configuration.LargeObjectThreshold, 8u * 1024u)
+													   : Max(m_Configuration.MemCacheSizeThreshold, 8u * 1024u);
+
+											   m_BlockStore.IterateBlock(
+												   std::span{begin(InlineBlockLocations), end(InlineBlockLocations)},
+												   ChunkIndexes,
+												   [this, &FillOne, &InlineDiskLocations, &InlineKeyIndexes](size_t		 ChunkIndex,
+																											 const void* Data,
+																											 uint64_t	 Size) -> bool {
+													   if (Data != nullptr)
+													   {
+														   FillOne(InlineDiskLocations[ChunkIndex],
+																   InlineKeyIndexes[ChunkIndex],
+																   IoBufferBuilder::MakeFromMemory(MemoryView(Data, Size)),
+																   /*UsesTemporaryMemory*/ true);
+													   }
+													   return true;
+												   },
+												   [this, &FillOne, &InlineDiskLocations, &InlineKeyIndexes](size_t			 ChunkIndex,
+																											 BlockStoreFile& File,
+																											 uint64_t		 Offset,
+																											 uint64_t		 Size) -> bool {
+													   FillOne(InlineDiskLocations[ChunkIndex],
+															   InlineKeyIndexes[ChunkIndex],
+															   File.GetChunk(Offset, Size),
+															   /*UsesTemporaryMemory*/ false);
+													   return true;
+												   },
+												   LargeChunkSizeLimit);
+											   return true;
+										   });
 			}
 
 			if (!StandaloneDiskLocations.empty())
@@ -1564,7 +1649,7 @@ ZenCacheDiskLayer::CacheBucket::EndGetBatch(GetBatchHandle* Batch) noexcept
 				{
 					size_t				KeyIndex = StandaloneKeyIndexes[Index];
 					const DiskLocation& Location = StandaloneDiskLocations[Index];
-					FillOne(Location, KeyIndex, GetStandaloneCacheValue(Location, Batch->Keys[KeyIndex]));
+					FillOne(Location, KeyIndex, GetStandaloneCacheValue(Location, Batch->Keys[KeyIndex]), /*UsesTemporaryMemory*/ false);
 				}
 			}
 
@@ -1644,10 +1729,6 @@ ZenCacheDiskLayer::CacheBucket::EndGetBatch(GetBatchHandle* Batch) noexcept
 				else
 				{
 					m_DiskMissCount++;
-					if (m_Configuration.MemCacheSizeThreshold > 0)
-					{
-						m_MemoryMissCount++;
-					}
 				}
 			}
 		}
@@ -1918,11 +1999,13 @@ ZenCacheDiskLayer::CacheBucket::GetUsageByAccess(GcClock::TimePoint Now, GcClock
 	}
 }
 
-bool
+std::function<void()>
 ZenCacheDiskLayer::CacheBucket::Drop()
 {
 	ZEN_TRACE_CPU("Z$::Bucket::Drop");
 
+	m_Gc.RemoveGcReferencer(*this);
+
 	RwLock::ExclusiveLockScope _(m_IndexLock);
 
 	std::vector<std::unique_ptr<RwLock::ExclusiveLockScope>> ShardLocks;
@@ -1934,7 +2017,7 @@ ZenCacheDiskLayer::CacheBucket::Drop()
 	m_BlockStore.Close();
 	m_SlogFile.Close();
 
-	const bool Deleted = cache::impl::MoveAndDeleteDirectory(m_BucketDir);
+	std::filesystem::path DroppedPath = cache::impl::MoveDroppedDirectory(m_BucketDir);
 
 	m_Index.clear();
 	m_Payloads.clear();
@@ -1947,7 +2030,21 @@ ZenCacheDiskLayer::CacheBucket::Drop()
 	m_OuterCacheMemoryUsage.fetch_sub(m_MemCachedSize.load());
 	m_MemCachedSize.store(0);
 
-	return Deleted;
+	if (DroppedPath.empty())
+	{
+		return {};
+	}
+	else
+	{
+		return [DroppedPath = std::move(DroppedPath)]() {
+			std::error_code Ec;
+			(void)DeleteDirectories(DroppedPath, Ec);
+			if (Ec)
+			{
+				ZEN_WARN("Failed to clean up dropped bucket directory '{}', reason: '{}'", DroppedPath, Ec.message());
+			}
+		};
+	}
 }
 
 void
@@ -1982,6 +2079,9 @@ ZenCacheDiskLayer::CacheBucket::SaveSnapshot(const std::function<uint64_t()>& Cl
 	ZEN_TRACE_CPU("Z$::Bucket::SaveSnapshot");
 	try
 	{
+		// Be defensive regarding log position as it is written to without acquiring m_LocationMapLock
+		const uint64_t LogPosition = m_SlogFile.GetLogCount();
+
 		bool UseLegacyScheme = false;
 
 		IoBuffer							  Buffer;
@@ -1996,7 +2096,7 @@ ZenCacheDiskLayer::CacheBucket::SaveSnapshot(const std::function<uint64_t()>& Cl
 
 			{
 				RwLock::SharedLockScope IndexLock(m_IndexLock);
-				WriteIndexSnapshot(IndexLock, /*Flush log*/ false);
+				WriteIndexSnapshot(IndexLock, LogPosition, /*Flush log*/ false);
 				// Note: this copy could be eliminated on shutdown to
 				// reduce memory usage and execution time
 				Index		= m_Index;
@@ -2036,7 +2136,7 @@ ZenCacheDiskLayer::CacheBucket::SaveSnapshot(const std::function<uint64_t()>& Cl
 		else
 		{
 			RwLock::SharedLockScope IndexLock(m_IndexLock);
-			WriteIndexSnapshot(IndexLock, /*Flush log*/ false);
+			WriteIndexSnapshot(IndexLock, LogPosition, /*Flush log*/ false);
 			const uint64_t EntryCount = m_Index.size();
 			Buffer					  = ManifestWriter.MakeSidecarManifest(m_BucketId, EntryCount);
 			uint64_t SidecarSize	  = ManifestWriter.GetSidecarSize();
@@ -2146,7 +2246,7 @@ ZenCacheDiskLayer::CacheBucket::ScrubStorage(ScrubContext& Ctx)
 					RwLock::SharedLockScope ValueLock(LockForHash(HashKey));
 
 					std::error_code Ec;
-					uintmax_t		size = std::filesystem::file_size(DataFilePath.ToPath(), Ec);
+					uintmax_t		size = FileSizeFromPath(DataFilePath.ToPath(), Ec);
 					if (Ec)
 					{
 						ReportBadKey(HashKey);
@@ -2287,11 +2387,11 @@ ZenCacheDiskLayer::CacheBucket::ScrubStorage(ScrubContext& Ctx)
 					BuildPath(Path, Entry.Key);
 					fs::path				   FilePath = Path.ToPath();
 					RwLock::ExclusiveLockScope ValueLock(LockForHash(Entry.Key));
-					if (fs::is_regular_file(FilePath))
+					if (IsFile(FilePath))
 					{
 						ZEN_DEBUG("deleting bad standalone cache file '{}'", Path.ToUtf8());
 						std::error_code Ec;
-						fs::remove(FilePath, Ec);  // We don't care if we fail, we are no longer tracking this file...
+						RemoveFile(FilePath, Ec);  // We don't care if we fail, we are no longer tracking this file...
 					}
 				}
 			}
@@ -2424,7 +2524,7 @@ ZenCacheDiskLayer::CacheBucket::PutStandaloneCacheValue(const IoHash& HashKey, c
 		 if (CleanUpTempFile)
 		 {
 			 std::error_code Ec;
-			 std::filesystem::remove(DataFile.GetPath(), Ec);
+			 RemoveFile(DataFile.GetPath(), Ec);
 			 if (Ec)
 			 {
 				 ZEN_WARN("Failed to clean up temporary file '{}' for put in '{}', reason '{}'",
@@ -2452,7 +2552,7 @@ ZenCacheDiskLayer::CacheBucket::PutStandaloneCacheValue(const IoHash& HashKey, c
 	RwLock::ExclusiveLockScope ValueLock(LockForHash(HashKey));
 
 	// We do a speculative remove of the file instead of probing with a exists call and check the error code instead
-	std::filesystem::remove(FsPath, Ec);
+	RemoveFile(FsPath, Ec);
 	if (Ec)
 	{
 		if (Ec.value() != ENOENT)
@@ -2460,7 +2560,7 @@ ZenCacheDiskLayer::CacheBucket::PutStandaloneCacheValue(const IoHash& HashKey, c
 			ZEN_WARN("Failed to remove file '{}' for put in '{}', reason: '{}', retrying.", FsPath, m_BucketDir, Ec.message());
 			Sleep(100);
 			Ec.clear();
-			std::filesystem::remove(FsPath, Ec);
+			RemoveFile(FsPath, Ec);
 			if (Ec && Ec.value() != ENOENT)
 			{
 				throw std::system_error(Ec, fmt::format("Failed to remove file '{}' for put in '{}'", FsPath, m_BucketDir));
@@ -2685,7 +2785,6 @@ ZenCacheDiskLayer::CacheBucket::PutInlineCacheValue(const IoHash&		 HashKey,
 								ZEN_MEMSCOPE(GetCacheDiskTag());
 								ZEN_TRACE_CPU("Z$::Bucket::UpdateLocation");
 								DiskLocation Location(BlockStoreLocation, m_Configuration.PayloadAlignment, EntryFlags);
-								m_SlogFile.Append({.Key = HashKey, .Location = Location});
 
 								RwLock::ExclusiveLockScope IndexLock(m_IndexLock);
 								if (m_TrackedCacheKeys)
@@ -2715,6 +2814,7 @@ ZenCacheDiskLayer::CacheBucket::PutInlineCacheValue(const IoHash&		 HashKey,
 									m_AccessTimes.emplace_back(GcClock::TickCount());
 									m_Index.insert_or_assign(HashKey, EntryIndex);
 								}
+								m_SlogFile.Append({.Key = HashKey, .Location = Location});
 							});
 }
 
@@ -2731,9 +2831,10 @@ class DiskBucketStoreCompactor : public GcStoreCompactor
 	using CacheBucket	= ZenCacheDiskLayer::CacheBucket;
 
 public:
-	DiskBucketStoreCompactor(CacheBucket& Bucket, std::vector<std::pair<IoHash, uint64_t>>&& ExpiredStandaloneKeys)
+	DiskBucketStoreCompactor(CacheBucket& Bucket, std::vector<std::pair<IoHash, uint64_t>>&& ExpiredStandaloneKeys, bool FlushBucket)
 	: m_Bucket(Bucket)
 	, m_ExpiredStandaloneKeys(std::move(ExpiredStandaloneKeys))
+	, m_FlushBucket(FlushBucket)
 	{
 		m_ExpiredStandaloneKeys.shrink_to_fit();
 	}
@@ -2791,7 +2892,7 @@ public:
 					ZEN_DEBUG("GCV2: cachebucket [COMPACT] '{}': deleting standalone cache file '{}'", m_Bucket.m_BucketDir, Path.ToUtf8());
 
 					std::error_code Ec;
-					if (!fs::remove(FilePath, Ec))
+					if (!RemoveFile(FilePath, Ec))
 					{
 						continue;
 					}
@@ -2812,7 +2913,7 @@ public:
 					ZEN_DEBUG("GCV2: cachebucket [COMPACT] '{}': checking standalone cache file '{}'", m_Bucket.m_BucketDir, Path.ToUtf8());
 
 					std::error_code Ec;
-					bool			Existed = std::filesystem::is_regular_file(FilePath, Ec);
+					bool			Existed = IsFile(FilePath, Ec);
 					if (Ec)
 					{
 						ZEN_WARN("GCV2: cachebucket [COMPACT] '{}': failed checking cache payload file '{}'. Reason '{}'",
@@ -2912,10 +3013,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;
@@ -2937,6 +3040,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())
@@ -2960,6 +3081,10 @@ public:
 				}
 			}
 		}
+		if (m_FlushBucket)
+		{
+			m_Bucket.Flush();
+		}
 	}
 
 	virtual std::string GetGcName(GcCtx& Ctx) override { return m_Bucket.GetGcName(Ctx); }
@@ -2967,6 +3092,7 @@ public:
 private:
 	ZenCacheDiskLayer::CacheBucket&			 m_Bucket;
 	std::vector<std::pair<IoHash, uint64_t>> m_ExpiredStandaloneKeys;
+	bool									 m_FlushBucket = false;
 };
 
 GcStoreCompactor*
@@ -2990,24 +3116,6 @@ ZenCacheDiskLayer::CacheBucket::RemoveExpiredData(GcCtx& Ctx, GcStats& Stats)
 					 NiceBytes(Stats.FreedMemory),
 					 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
 		}
-		if (Stats.DeletedCount > 0)
-		{
-			bool Expected = false;
-			if (m_IsFlushing || !m_IsFlushing.compare_exchange_strong(Expected, true))
-			{
-				return;
-			}
-			auto FlushingGuard = MakeGuard([&] { m_IsFlushing.store(false); });
-
-			try
-			{
-				SaveSnapshot([]() { return 0; });
-			}
-			catch (const std::exception& Ex)
-			{
-				ZEN_WARN("Failed to write index and manifest after RemoveExpiredData in '{}'. Reason: '{}'", m_BucketDir, Ex.what());
-			}
-		}
 	});
 
 	const GcClock::Tick ExpireTicks = Ctx.Settings.CacheExpireTime.time_since_epoch().count();
@@ -3059,7 +3167,7 @@ ZenCacheDiskLayer::CacheBucket::RemoveExpiredData(GcCtx& Ctx, GcStats& Stats)
 			return nullptr;
 		}
 
-		if (Ctx.Settings.IsDeleteMode)
+		if (Ctx.Settings.IsDeleteMode && !ExpiredEntries.empty())
 		{
 			for (const DiskIndexEntry& Entry : ExpiredEntries)
 			{
@@ -3094,7 +3202,7 @@ ZenCacheDiskLayer::CacheBucket::RemoveExpiredData(GcCtx& Ctx, GcStats& Stats)
 		return nullptr;
 	}
 
-	return new DiskBucketStoreCompactor(*this, std::move(ExpiredStandaloneKeys));
+	return new DiskBucketStoreCompactor(*this, std::move(ExpiredStandaloneKeys), /*FlushBucket*/ Stats.DeletedCount > 0);
 }
 
 bool
@@ -3284,7 +3392,7 @@ ZenCacheDiskLayer::CacheBucket::GetReferences(const LoggerRef&	   Logger,
 						CaptureAttachments(ChunkIndex, File.GetChunk(Offset, Size).GetView());
 						return !IsCancelledFlag.load();
 					},
-					0);
+					32u * 1024);
 
 				if (Continue)
 				{
@@ -3583,20 +3691,6 @@ ZenCacheDiskLayer::~ZenCacheDiskLayer()
 	}
 }
 
-template<typename T, typename U>
-struct equal_to_2 : public eastl::binary_function<T, U, bool>
-{
-	constexpr bool operator()(const T& a, const U& b) const { return a == b; }
-
-	template<typename T_ = T,
-			 typename U_ = U,
-			 typename	 = eastl::enable_if_t<!eastl::is_same_v<eastl::remove_const_t<T_>, eastl::remove_const_t<U_>>>>
-	constexpr bool operator()(const U& b, const T& a) const
-	{
-		return b == a;
-	}
-};
-
 ZenCacheDiskLayer::CacheBucket*
 ZenCacheDiskLayer::GetOrCreateBucket(std::string_view InBucket)
 {
@@ -3604,7 +3698,7 @@ ZenCacheDiskLayer::GetOrCreateBucket(std::string_view InBucket)
 
 	{
 		RwLock::SharedLockScope SharedLock(m_Lock);
-		if (auto It = m_Buckets.find_as(InBucket, std::hash<std::string_view>(), equal_to_2<std::string, std::string_view>());
+		if (auto It = m_Buckets.find_as(InBucket, std::hash<std::string_view>(), eastl::equal_to_2<std::string, std::string_view>());
 			It != m_Buckets.end())
 		{
 			return It->second.get();
@@ -3613,10 +3707,18 @@ ZenCacheDiskLayer::GetOrCreateBucket(std::string_view InBucket)
 
 	// We create the bucket without holding a lock since contructor calls GcManager::AddGcReferencer which takes an exclusive lock.
 	// This can cause a deadlock, if GC is running we would block while holding ZenCacheDiskLayer::m_Lock
-	std::unique_ptr<CacheBucket> Bucket(std::make_unique<CacheBucket>(m_Gc, m_TotalMemCachedSize, InBucket, m_Configuration.BucketConfig));
+	BucketConfiguration* BucketConfig = &m_Configuration.BucketConfig;
+	if (auto It = m_Configuration.BucketConfigMap.find_as(InBucket,
+														  std::hash<std::string_view>(),
+														  eastl::equal_to_2<std::string, std::string_view>());
+		It != m_Configuration.BucketConfigMap.end())
+	{
+		BucketConfig = &It->second;
+	}
+	std::unique_ptr<CacheBucket> Bucket(std::make_unique<CacheBucket>(m_Gc, m_TotalMemCachedSize, InBucket, *BucketConfig));
 
 	RwLock::ExclusiveLockScope Lock(m_Lock);
-	if (auto It = m_Buckets.find_as(InBucket, std::hash<std::string_view>(), equal_to_2<std::string, std::string_view>());
+	if (auto It = m_Buckets.find_as(InBucket, std::hash<std::string_view>(), eastl::equal_to_2<std::string, std::string_view>());
 		It != m_Buckets.end())
 	{
 		return It->second.get();
@@ -3896,7 +3998,11 @@ ZenCacheDiskLayer::DiscoverBuckets()
 		if (IsKnownBadBucketName(BucketName))
 		{
 			BadBucketDirectories.push_back(BucketPath);
-
+			continue;
+		}
+		else if (BucketName.starts_with("[dropped]"))
+		{
+			BadBucketDirectories.push_back(BucketPath);
 			continue;
 		}
 
@@ -3929,50 +4035,66 @@ ZenCacheDiskLayer::DiscoverBuckets()
 	RwLock SyncLock;
 
 	WorkerThreadPool& Pool = GetLargeWorkerPool(EWorkloadType::Burst);
-	Latch			  WorkLatch(1);
-	for (auto& BucketPath : FoundBucketDirectories)
+	std::atomic<bool> AbortFlag;
+	std::atomic<bool> PauseFlag;
+	ParallelWork	  Work(AbortFlag, PauseFlag);
+	try
 	{
-		WorkLatch.AddCount(1);
-		Pool.ScheduleWork([this, &WorkLatch, &SyncLock, BucketPath]() {
-			ZEN_MEMSCOPE(GetCacheDiskTag());
-
-			auto			  _			 = MakeGuard([&]() { WorkLatch.CountDown(); });
-			const std::string BucketName = PathToUtf8(BucketPath.stem());
-			try
-			{
-				std::unique_ptr<CacheBucket> NewBucket =
-					std::make_unique<CacheBucket>(m_Gc, m_TotalMemCachedSize, BucketName, m_Configuration.BucketConfig);
+		for (auto& BucketPath : FoundBucketDirectories)
+		{
+			Work.ScheduleWork(Pool, [this, &SyncLock, BucketPath](std::atomic<bool>&) {
+				ZEN_MEMSCOPE(GetCacheDiskTag());
 
-				CacheBucket* Bucket = nullptr;
+				const std::string BucketName = PathToUtf8(BucketPath.stem());
+				try
 				{
-					RwLock::ExclusiveLockScope __(SyncLock);
-					auto					   InsertResult = m_Buckets.emplace(BucketName, std::move(NewBucket));
-					Bucket									= InsertResult.first->second.get();
-				}
-				ZEN_ASSERT(Bucket);
+					BucketConfiguration* BucketConfig = &m_Configuration.BucketConfig;
+					if (auto It = m_Configuration.BucketConfigMap.find_as(std::string_view(BucketName),
+																		  std::hash<std::string_view>(),
+																		  eastl::equal_to_2<std::string, std::string_view>());
+						It != m_Configuration.BucketConfigMap.end())
+					{
+						BucketConfig = &It->second;
+					}
 
-				if (!Bucket->OpenOrCreate(BucketPath, /* AllowCreate */ false))
-				{
-					ZEN_WARN("Found directory '{}' in our base directory '{}' but it is not a valid bucket", BucketName, m_RootDir);
+					std::unique_ptr<CacheBucket> NewBucket =
+						std::make_unique<CacheBucket>(m_Gc, m_TotalMemCachedSize, BucketName, *BucketConfig);
 
+					CacheBucket* Bucket = nullptr;
 					{
 						RwLock::ExclusiveLockScope __(SyncLock);
-						m_Buckets.erase(BucketName);
+						auto					   InsertResult = m_Buckets.emplace(BucketName, std::move(NewBucket));
+						Bucket									= InsertResult.first->second.get();
+					}
+					ZEN_ASSERT(Bucket);
+
+					if (!Bucket->OpenOrCreate(BucketPath, /* AllowCreate */ false))
+					{
+						ZEN_WARN("Found directory '{}' in our base directory '{}' but it is not a valid bucket", BucketName, m_RootDir);
+
+						{
+							RwLock::ExclusiveLockScope __(SyncLock);
+							m_Buckets.erase(BucketName);
+						}
 					}
 				}
-			}
-			catch (const std::exception& Err)
-			{
-				ZEN_ERROR("Opening bucket '{}' in '{}' FAILED, reason: '{}'", BucketName, BucketPath, Err.what());
-				return;
-			}
-		});
+				catch (const std::exception& Err)
+				{
+					ZEN_ERROR("Opening bucket '{}' in '{}' FAILED, reason: '{}'", BucketName, BucketPath, Err.what());
+					return;
+				}
+			});
+		}
+	}
+	catch (const std::exception& Ex)
+	{
+		AbortFlag.store(true);
+		ZEN_WARN("Failed discovering buckets in {}. Reason: '{}'", m_RootDir, Ex.what());
 	}
-	WorkLatch.CountDown();
-	WorkLatch.Wait();
+	Work.Wait();
 }
 
-bool
+std::function<void()>
 ZenCacheDiskLayer::DropBucket(std::string_view InBucket)
 {
 	ZEN_TRACE_CPU("Z$::DropBucket");
@@ -3990,33 +4112,72 @@ ZenCacheDiskLayer::DropBucket(std::string_view InBucket)
 		return Bucket.Drop();
 	}
 
-	// Make sure we remove the folder even if we don't know about the bucket
 	std::filesystem::path BucketPath = m_RootDir;
 	BucketPath /= std::string(InBucket);
-	return cache::impl::MoveAndDeleteDirectory(BucketPath);
+	std::filesystem::path DroppedPath = cache::impl::MoveDroppedDirectory(BucketPath);
+	if (DroppedPath.empty())
+	{
+		return {};
+	}
+	else
+	{
+		return [DroppedPath = std::move(DroppedPath)]() {
+			std::error_code Ec;
+			(void)DeleteDirectories(DroppedPath, Ec);
+			if (Ec)
+			{
+				ZEN_WARN("Failed to clean up dropped bucket directory '{}', reason: '{}'", DroppedPath, Ec.message());
+			}
+		};
+	}
 }
 
-bool
+std::function<void()>
 ZenCacheDiskLayer::Drop()
 {
 	ZEN_TRACE_CPU("Z$::Drop");
 
-	RwLock::ExclusiveLockScope _(m_Lock);
-
-	std::vector<std::unique_ptr<CacheBucket>> Buckets;
-	Buckets.reserve(m_Buckets.size());
-	while (!m_Buckets.empty())
+	std::vector<std::function<void()>> PostDropOps;
 	{
-		const auto&	 It		= m_Buckets.begin();
-		CacheBucket& Bucket = *It->second;
-		m_DroppedBuckets.push_back(std::move(It->second));
-		m_Buckets.erase(It->first);
-		if (!Bucket.Drop())
+		RwLock::ExclusiveLockScope _(m_Lock);
+		PostDropOps.reserve(m_Buckets.size());
+		while (!m_Buckets.empty())
 		{
-			return false;
+			const auto&	 It		= m_Buckets.begin();
+			CacheBucket& Bucket = *It->second;
+			m_DroppedBuckets.push_back(std::move(It->second));
+			m_Buckets.erase(It->first);
+			if (std::function<void()> PostDropOp = Bucket.Drop(); !PostDropOp)
+			{
+				return {};
+			}
+			else
+			{
+				PostDropOps.emplace_back(std::move(PostDropOp));
+			}
 		}
 	}
-	return cache::impl::MoveAndDeleteDirectory(m_RootDir);
+
+	std::filesystem::path DroppedPath = cache::impl::MoveDroppedDirectory(m_RootDir);
+	if (DroppedPath.empty())
+	{
+		return {};
+	}
+	else
+	{
+		return [DroppedPath = std::move(DroppedPath), PostDropOps = std::move(PostDropOps)]() {
+			for (auto& PostDropOp : PostDropOps)
+			{
+				PostDropOp();
+			}
+			std::error_code Ec;
+			(void)DeleteDirectories(DroppedPath, Ec);
+			if (Ec)
+			{
+				ZEN_WARN("Failed to clean up dropped bucket directory '{}', reason: '{}'", DroppedPath, Ec.message());
+			}
+		};
+	}
 }
 
 void
@@ -4046,16 +4207,16 @@ ZenCacheDiskLayer::Flush()
 	}
 	{
 		WorkerThreadPool& Pool = GetMediumWorkerPool(EWorkloadType::Burst);
-		Latch			  WorkLatch(1);
+		std::atomic<bool> AbortFlag;
+		std::atomic<bool> PauseFlag;
+		ParallelWork	  Work(AbortFlag, PauseFlag);
 		try
 		{
 			for (auto& Bucket : Buckets)
 			{
-				WorkLatch.AddCount(1);
-				Pool.ScheduleWork([&WorkLatch, Bucket]() {
+				Work.ScheduleWork(Pool, [Bucket](std::atomic<bool>&) {
 					ZEN_MEMSCOPE(GetCacheDiskTag());
 
-					auto _ = MakeGuard([&]() { WorkLatch.CountDown(); });
 					try
 					{
 						Bucket->Flush();
@@ -4069,13 +4230,14 @@ ZenCacheDiskLayer::Flush()
 		}
 		catch (const std::exception& Ex)
 		{
+			AbortFlag.store(true);
 			ZEN_ERROR("Failed to flush buckets at '{}'. Reason: '{}'", m_RootDir, Ex.what());
 		}
-		WorkLatch.CountDown();
-		while (!WorkLatch.Wait(1000))
-		{
-			ZEN_DEBUG("Waiting for {} buckets at '{}' to flush", WorkLatch.Remaining(), m_RootDir);
-		}
+
+		Work.Wait(1000, [&](bool IsAborted, bool IsPaused, std::ptrdiff_t RemainingWork) {
+			ZEN_UNUSED(IsAborted, IsPaused);
+			ZEN_DEBUG("Waiting for {} buckets at '{}' to flush", RemainingWork, m_RootDir);
+		});
 	}
 }