path: root/zenstore/compactcas.cpp

// Copyright Epic Games, Inc. All Rights Reserved.

#include "compactcas.h"

#include <zenstore/cas.h>

#include <zencore/except.h>
#include <zencore/filesystem.h>
#include <zencore/fmtutils.h>
#include <zencore/logging.h>
#include <zencore/scopeguard.h>
#include <gsl/gsl-lite.hpp>

#include <xxhash.h>

#if ZEN_WITH_TESTS
#	include <zencore/compactbinarybuilder.h>
#	include <zencore/testing.h>
#	include <zencore/testutils.h>
#	include <zencore/workthreadpool.h>
#	include <zenstore/cidstore.h>
#	include <algorithm>
#	include <random>
#endif

//////////////////////////////////////////////////////////////////////////

namespace zen {

struct CasDiskIndexHeader
{
	static constexpr uint32_t ExpectedMagic	 = 0x75696478;	// 'uidx';
	static constexpr uint32_t CurrentVersion = 1;

	uint32_t Magic			  = ExpectedMagic;
	uint32_t Version		  = CurrentVersion;
	uint64_t EntryCount		  = 0;
	uint64_t LogPosition	  = 0;
	uint32_t PayloadAlignment = 0;
	uint32_t Checksum		  = 0;

	static uint32_t ComputeChecksum(const CasDiskIndexHeader& Header)
	{
		return XXH32(&Header.Magic, sizeof(CasDiskIndexHeader) - sizeof(uint32_t), 0xC0C0'BABA);
	}
};

static_assert(sizeof(CasDiskIndexHeader) == 32);

namespace {
	std::vector<CasDiskIndexEntry> MakeCasDiskEntries(const std::unordered_map<IoHash, BlockStoreDiskLocation>& MovedChunks,
													  const std::vector<IoHash>&								DeletedChunks)
	{
		std::vector<CasDiskIndexEntry> result;
		result.reserve(MovedChunks.size());
		for (const auto& MovedEntry : MovedChunks)
		{
			result.push_back({.Key = MovedEntry.first, .Location = MovedEntry.second});
		}
		for (const IoHash& ChunkHash : DeletedChunks)
		{
			result.push_back({.Key = ChunkHash, .Flags = CasDiskIndexEntry::kTombstone});
		}
		return result;
	}

	const char* IndexExtension = ".uidx";
	const char* LogExtension   = ".ulog";
	const char* DataExtension  = ".ucas";

	std::filesystem::path GetBasePath(const std::filesystem::path& RootPath, const std::string& ContainerBaseName)
	{
		return RootPath / ContainerBaseName;
	}

	std::filesystem::path GetIndexPath(const std::filesystem::path& RootPath, const std::string& ContainerBaseName)
	{
		return GetBasePath(RootPath, ContainerBaseName) / (ContainerBaseName + IndexExtension);
	}

	std::filesystem::path GetTempIndexPath(const std::filesystem::path& RootPath, const std::string& ContainerBaseName)
	{
		return GetBasePath(RootPath, ContainerBaseName) / (ContainerBaseName + ".tmp" + LogExtension);
	}

	std::filesystem::path GetLogPath(const std::filesystem::path& RootPath, const std::string& ContainerBaseName)
	{
		return GetBasePath(RootPath, ContainerBaseName) / (ContainerBaseName + LogExtension);
	}

	std::filesystem::path GetBlocksBasePath(const std::filesystem::path& RootPath, const std::string& ContainerBaseName)
	{
		return GetBasePath(RootPath, ContainerBaseName) / "blocks";
	}

	std::filesystem::path GetBlockPath(const std::filesystem::path& BlocksBasePath, const uint32_t BlockIndex)
	{
		ExtendablePathBuilder<256> Path;

		char BlockHexString[9];
		ToHexNumber(BlockIndex, BlockHexString);

		Path.Append(BlocksBasePath);
		Path.AppendSeparator();
		Path.AppendAsciiRange(BlockHexString, BlockHexString + 4);
		Path.AppendSeparator();
		Path.Append(BlockHexString);
		Path.Append(DataExtension);
		return Path.ToPath();
	}

	std::filesystem::path GetLegacyLogPath(const std::filesystem::path& RootPath, const std::string& ContainerBaseName)
	{
		return RootPath / (ContainerBaseName + LogExtension);
	}

	std::filesystem::path GetLegacyDataPath(const std::filesystem::path& RootPath, const std::string& ContainerBaseName)
	{
		return RootPath / (ContainerBaseName + DataExtension);
	}

	std::filesystem::path GetLegacyIndexPath(const std::filesystem::path& RootPath, const std::string& ContainerBaseName)
	{
		return RootPath / (ContainerBaseName + IndexExtension);
	}

	struct LegacyCasDiskLocation
	{
		LegacyCasDiskLocation(uint64_t InOffset, uint64_t InSize)
		{
			ZEN_ASSERT(InOffset <= 0xff'ffff'ffff);
			ZEN_ASSERT(InSize <= 0xff'ffff'ffff);

			memcpy(&m_Offset[0], &InOffset, sizeof m_Offset);
			memcpy(&m_Size[0], &InSize, sizeof m_Size);
		}

		LegacyCasDiskLocation() = default;

		inline uint64_t GetOffset() const
		{
			uint64_t Offset = 0;
			memcpy(&Offset, &m_Offset, sizeof m_Offset);
			return Offset;
		}

		inline uint64_t GetSize() const
		{
			uint64_t Size = 0;
			memcpy(&Size, &m_Size, sizeof m_Size);
			return Size;
		}

	private:
		uint8_t m_Offset[5];
		uint8_t m_Size[5];
	};

	struct LegacyCasDiskIndexEntry
	{
		static const uint8_t kTombstone = 0x01;

		IoHash				  Key;
		LegacyCasDiskLocation Location;
		ZenContentType		  ContentType = ZenContentType::kUnknownContentType;
		uint8_t				  Flags		  = 0;
	};

	bool ValidateLegacyEntry(const LegacyCasDiskIndexEntry& Entry, std::string& OutReason)
	{
		if (Entry.Key == IoHash::Zero)
		{
			OutReason = fmt::format("Invalid hash key {}", Entry.Key.ToHexString());
			return false;
		}
		if ((Entry.Flags & ~LegacyCasDiskIndexEntry::kTombstone) != 0)
		{
			OutReason = fmt::format("Invalid flags {} for entry {}", Entry.Flags, Entry.Key.ToHexString());
			return false;
		}
		if (Entry.Flags & LegacyCasDiskIndexEntry::kTombstone)
		{
			return true;
		}
		if (Entry.ContentType != ZenContentType::kUnknownContentType)
		{
			OutReason =
				fmt::format("Invalid content type {} for entry {}", static_cast<uint8_t>(Entry.ContentType), Entry.Key.ToHexString());
			return false;
		}
		uint64_t Size = Entry.Location.GetSize();
		if (Size == 0)
		{
			OutReason = fmt::format("Invalid size {} for entry {}", Size, Entry.Key.ToHexString());
			return false;
		}
		return true;
	}

	bool ValidateEntry(const CasDiskIndexEntry& Entry, std::string& OutReason)
	{
		if (Entry.Key == IoHash::Zero)
		{
			OutReason = fmt::format("Invalid hash key {}", Entry.Key.ToHexString());
			return false;
		}
		if ((Entry.Flags & ~CasDiskIndexEntry::kTombstone) != 0)
		{
			OutReason = fmt::format("Invalid flags {} for entry {}", Entry.Flags, Entry.Key.ToHexString());
			return false;
		}
		if (Entry.Flags & CasDiskIndexEntry::kTombstone)
		{
			return true;
		}
		if (Entry.ContentType != ZenContentType::kUnknownContentType)
		{
			OutReason =
				fmt::format("Invalid content type {} for entry {}", static_cast<uint8_t>(Entry.ContentType), Entry.Key.ToHexString());
			return false;
		}
		uint64_t Size = Entry.Location.GetSize();
		if (Size == 0)
		{
			OutReason = fmt::format("Invalid size {} for entry {}", Size, Entry.Key.ToHexString());
			return false;
		}
		return true;
	}

}  // namespace

//////////////////////////////////////////////////////////////////////////

CasContainerStrategy::CasContainerStrategy(const CasStoreConfiguration& Config, CasGc& Gc)
: GcStorage(Gc)
, m_Config(Config)
, m_Log(logging::Get("containercas"))
{
}

CasContainerStrategy::~CasContainerStrategy()
{
}

void
CasContainerStrategy::Initialize(const std::string_view ContainerBaseName, uint32_t MaxBlockSize, uint64_t Alignment, bool IsNewStore)
{
	ZEN_ASSERT(IsPow2(Alignment));
	ZEN_ASSERT(!m_IsInitialized);
	ZEN_ASSERT(MaxBlockSize > 0);

	m_ContainerBaseName = ContainerBaseName;
	m_PayloadAlignment	= Alignment;
	m_MaxBlockSize		= MaxBlockSize;
	m_BlocksBasePath	= GetBlocksBasePath(m_Config.RootDirectory, m_ContainerBaseName);

	OpenContainer(IsNewStore);

	m_IsInitialized = true;
}

CasStore::InsertResult
CasContainerStrategy::InsertChunk(const void* ChunkData, size_t ChunkSize, const IoHash& ChunkHash)
{
	uint32_t			WriteBlockIndex;
	Ref<BlockStoreFile> WriteBlock;
	uint64_t			InsertOffset;
	{
		RwLock::ExclusiveLockScope _(m_InsertLock);

		{
			RwLock::SharedLockScope __(m_LocationMapLock);
			if (m_LocationMap.contains(ChunkHash))
			{
				return CasStore::InsertResult{.New = false};
			}
		}

		// New entry

		WriteBlockIndex = m_WriteBlockIndex.load(std::memory_order_acquire);
		bool IsWriting	= m_WriteBlock != nullptr;
		if (!IsWriting || (m_CurrentInsertOffset + ChunkSize) > m_MaxBlockSize)
		{
			if (m_WriteBlock)
			{
				m_WriteBlock = nullptr;
			}
			{
				RwLock::ExclusiveLockScope __(m_LocationMapLock);
				if (m_ChunkBlocks.size() == BlockStoreDiskLocation::MaxBlockIndex)
				{
					throw std::runtime_error(
						fmt::format("unable to allocate a new block in '{}'", m_Config.RootDirectory / m_ContainerBaseName));
				}
				WriteBlockIndex += IsWriting ? 1 : 0;
				while (m_ChunkBlocks.contains(WriteBlockIndex))
				{
					WriteBlockIndex = (WriteBlockIndex + 1) & BlockStoreDiskLocation::MaxBlockIndex;
				}
				std::filesystem::path BlockPath = GetBlockPath(m_BlocksBasePath, WriteBlockIndex);
				m_WriteBlock					= new BlockStoreFile(BlockPath);
				m_ChunkBlocks[WriteBlockIndex]	= m_WriteBlock;
				m_WriteBlockIndex.store(WriteBlockIndex, std::memory_order_release);
			}
			m_CurrentInsertOffset = 0;
			m_WriteBlock->Create(m_MaxBlockSize);
		}
		InsertOffset		  = m_CurrentInsertOffset;
		m_CurrentInsertOffset = RoundUp(InsertOffset + ChunkSize, m_PayloadAlignment);
		WriteBlock			  = m_WriteBlock;
	}

	// We can end up in a situation that InsertChunk writes the same chunk data in
	// different locations.
	// We release the insert lock once we have the correct WriteBlock ready and we know
	// where to write the data. If a new InsertChunk request for the same chunk hash/data
	// comes in before we update m_LocationMap below we will have a race.
	// The outcome of that is that we will write the chunk data in more than one location
	// but the chunk hash will only point to one of the chunks.
	// We will in that case waste space until the next GC operation.
	//
	// This should be a rare occasion and the current flow reduces the time we block for
	// reads, insert and GC.

	BlockStoreDiskLocation	Location({.BlockIndex = WriteBlockIndex, .Offset = InsertOffset, .Size = ChunkSize}, m_PayloadAlignment);
	const CasDiskIndexEntry IndexEntry{.Key = ChunkHash, .Location = Location};

	WriteBlock->Write(ChunkData, ChunkSize, InsertOffset);
	m_CasLog.Append(IndexEntry);

	m_TotalSize.fetch_add(static_cast<uint64_t>(ChunkSize), std::memory_order_seq_cst);
	{
		RwLock::ExclusiveLockScope __(m_LocationMapLock);
		m_LocationMap.emplace(ChunkHash, Location);
	}

	return CasStore::InsertResult{.New = true};
}

CasStore::InsertResult
CasContainerStrategy::InsertChunk(IoBuffer Chunk, const IoHash& ChunkHash)
{
	return InsertChunk(Chunk.Data(), Chunk.Size(), ChunkHash);
}

IoBuffer
CasContainerStrategy::FindChunk(const IoHash& ChunkHash)
{
	Ref<BlockStoreFile> ChunkBlock;
	BlockStoreLocation	Location;
	{
		RwLock::SharedLockScope _(m_LocationMapLock);
		if (auto KeyIt = m_LocationMap.find(ChunkHash); KeyIt != m_LocationMap.end())
		{
			Location   = KeyIt->second.Get(m_PayloadAlignment);
			ChunkBlock = m_ChunkBlocks[Location.BlockIndex];
		}
		else
		{
			return IoBuffer();
		}
	}
	return ChunkBlock->GetChunk(Location.Offset, Location.Size);
}

bool
CasContainerStrategy::HaveChunk(const IoHash& ChunkHash)
{
	RwLock::SharedLockScope _(m_LocationMapLock);
	return m_LocationMap.contains(ChunkHash);
}

void
CasContainerStrategy::FilterChunks(CasChunkSet& InOutChunks)
{
	// This implementation is good enough for relatively small
	// chunk sets (in terms of chunk identifiers), but would
	// benefit from a better implementation which removes
	// items incrementally for large sets, especially when
	// we're likely to already have a large proportion of the
	// chunks in the set

	InOutChunks.RemoveChunksIf([&](const IoHash& Hash) { return HaveChunk(Hash); });
}

void
CasContainerStrategy::Flush()
{
	{
		RwLock::ExclusiveLockScope _(m_InsertLock);
		if (m_CurrentInsertOffset > 0)
		{
			uint32_t WriteBlockIndex = m_WriteBlockIndex.load(std::memory_order_acquire);
			WriteBlockIndex			 = (WriteBlockIndex + 1) & BlockStoreDiskLocation::MaxBlockIndex;
			m_WriteBlock			 = nullptr;
			m_WriteBlockIndex.store(WriteBlockIndex, std::memory_order_release);
			m_CurrentInsertOffset = 0;
		}
	}
	MakeIndexSnapshot();
}

void
CasContainerStrategy::Scrub(ScrubContext& Ctx)
{
	std::vector<CasDiskIndexEntry> BadChunks;

	// We do a read sweep through the payloads file and validate
	// any entries that are contained within each segment, with
	// the assumption that most entries will be checked in this
	// pass. An alternative strategy would be to use memory mapping.

	{
		std::vector<CasDiskIndexEntry> BigChunks;
		const uint64_t				   WindowSize = 4 * 1024 * 1024;
		IoBuffer					   ReadBuffer{WindowSize};
		void*						   BufferBase = ReadBuffer.MutableData();

		RwLock::SharedLockScope _(m_InsertLock);  // TODO: Refactor so we don't have to keep m_InsertLock all the time?
		RwLock::SharedLockScope __(m_LocationMapLock);

		for (const auto& Block : m_ChunkBlocks)
		{
			uint64_t				   WindowStart = 0;
			uint64_t				   WindowEnd   = WindowSize;
			const Ref<BlockStoreFile>& BlockFile   = Block.second;
			BlockFile->Open();
			const uint64_t FileSize = BlockFile->FileSize();

			do
			{
				const uint64_t ChunkSize = Min(WindowSize, FileSize - WindowStart);
				BlockFile->Read(BufferBase, ChunkSize, WindowStart);

				for (auto& Entry : m_LocationMap)
				{
					const BlockStoreLocation Location	 = Entry.second.Get(m_PayloadAlignment);
					const uint64_t			 EntryOffset = Location.Offset;

					if ((EntryOffset >= WindowStart) && (EntryOffset < WindowEnd))
					{
						const uint64_t EntryEnd = EntryOffset + Location.Size;

						if (EntryEnd >= WindowEnd)
						{
							BigChunks.push_back({.Key = Entry.first, .Location = Entry.second});

							continue;
						}

						const IoHash ComputedHash =
							IoHash::HashBuffer(reinterpret_cast<uint8_t*>(BufferBase) + Location.Offset - WindowStart, Location.Size);

						if (Entry.first != ComputedHash)
						{
							// Hash mismatch
							BadChunks.push_back({.Key = Entry.first, .Location = Entry.second, .Flags = CasDiskIndexEntry::kTombstone});
						}
					}
				}

				WindowStart += WindowSize;
				WindowEnd += WindowSize;
			} while (WindowStart < FileSize);
		}

		// Deal with large chunks

		for (const CasDiskIndexEntry& Entry : BigChunks)
		{
			IoHashStream			   Hasher;
			const BlockStoreLocation   Location	 = Entry.Location.Get(m_PayloadAlignment);
			const Ref<BlockStoreFile>& BlockFile = m_ChunkBlocks[Location.BlockIndex];
			BlockFile->StreamByteRange(Location.Offset, Location.Size, [&](const void* Data, uint64_t Size) { Hasher.Append(Data, Size); });
			IoHash ComputedHash = Hasher.GetHash();

			if (Entry.Key != ComputedHash)
			{
				BadChunks.push_back({.Key = Entry.Key, .Location = Entry.Location, .Flags = CasDiskIndexEntry::kTombstone});
			}
		}
	}

	if (BadChunks.empty())
	{
		return;
	}

	ZEN_ERROR("Scrubbing found {} bad chunks in '{}'", BadChunks.size(), m_Config.RootDirectory / m_ContainerBaseName);

	// Deal with bad chunks by removing them from our lookup map

	std::vector<IoHash> BadChunkHashes;
	BadChunkHashes.reserve(BadChunks.size());

	m_CasLog.Append(BadChunks);
	{
		RwLock::ExclusiveLockScope _(m_LocationMapLock);
		for (const CasDiskIndexEntry& Entry : BadChunks)
		{
			BadChunkHashes.push_back(Entry.Key);
			m_LocationMap.erase(Entry.Key);
		}
	}

	// Let whomever it concerns know about the bad chunks. This could
	// be used to invalidate higher level data structures more efficiently
	// than a full validation pass might be able to do

	Ctx.ReportBadCasChunks(BadChunkHashes);
}

void
CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
{
	// It collects all the blocks that we want to delete chunks from. For each such
	// block we keep a list of chunks to retain and a list of chunks to delete.
	//
	// If there is a block that we are currently writing to, that block is omitted
	// from the garbage collection.
	//
	// Next it will iterate over all blocks that we want to remove chunks from.
	// If the block is empty after removal of chunks we mark the block as pending
	// delete - we want to delete it as soon as there are no IoBuffers using the
	// block file.
	// Once complete we update the m_LocationMap by removing the chunks.
	//
	// If the block is non-empty we write out the chunks we want to keep to a new
	// block file (creating new block files as needed).
	//
	// We update the index as we complete each new block file. This makes it possible
	// to break the GC if we want to limit time for execution.
	//
	// GC can fairly parallell to regular operation - it will block while taking
	// a snapshot of the current m_LocationMap state.
	//
	// While moving blocks it will do a blocking operation and update the m_LocationMap
	// after each new block is written and figuring out the path to the next new block.

	ZEN_INFO("collecting garbage from '{}'", m_Config.RootDirectory / m_ContainerBaseName);
	uint64_t WriteBlockTimeUs		 = 0;
	uint64_t WriteBlockLongestTimeUs = 0;
	uint64_t ReadBlockTimeUs		 = 0;
	uint64_t ReadBlockLongestTimeUs	 = 0;
	uint64_t TotalChunkCount		 = 0;
	uint64_t DeletedSize			 = 0;
	uint64_t OldTotalSize			 = m_TotalSize.load(std::memory_order::relaxed);

	std::vector<IoHash> DeletedChunks;
	uint64_t			MovedCount = 0;

	Stopwatch  TotalTimer;
	const auto _ = MakeGuard([this,
							  &TotalTimer,
							  &WriteBlockTimeUs,
							  &WriteBlockLongestTimeUs,
							  &ReadBlockTimeUs,
							  &ReadBlockLongestTimeUs,
							  &TotalChunkCount,
							  &DeletedChunks,
							  &MovedCount,
							  &DeletedSize,
							  OldTotalSize] {
		ZEN_INFO(
			"garbage collect for '{}' DONE after {}, write lock: {} ({}), read lock: {} ({}), collected {} bytes, deleted #{} and moved "
			"#{} "
			"of #{} "
			"chunks ({}).",
			m_Config.RootDirectory / m_ContainerBaseName,
			NiceTimeSpanMs(TotalTimer.GetElapsedTimeMs()),
			NiceLatencyNs(WriteBlockTimeUs),
			NiceLatencyNs(WriteBlockLongestTimeUs),
			NiceLatencyNs(ReadBlockTimeUs),
			NiceLatencyNs(ReadBlockLongestTimeUs),
			NiceBytes(DeletedSize),
			DeletedChunks.size(),
			MovedCount,
			TotalChunkCount,
			NiceBytes(OldTotalSize));
	});

	LocationMap_t LocationMap;
	size_t		  BlockCount;
	uint64_t	  ExcludeBlockIndex = 0x800000000ull;
	{
		RwLock::SharedLockScope __(m_InsertLock);
		RwLock::SharedLockScope ___(m_LocationMapLock);
		{
			Stopwatch  Timer;
			const auto ____ = MakeGuard([&Timer, &WriteBlockTimeUs, &WriteBlockLongestTimeUs] {
				uint64_t ElapsedUs = Timer.GetElapsedTimeUs();
				WriteBlockTimeUs += ElapsedUs;
				WriteBlockLongestTimeUs = std::max(ElapsedUs, WriteBlockLongestTimeUs);
			});
			if (m_WriteBlock)
			{
				ExcludeBlockIndex = m_WriteBlockIndex.load(std::memory_order_acquire);
			}
			__.ReleaseNow();
		}
		LocationMap = m_LocationMap;
		BlockCount	= m_ChunkBlocks.size();
	}

	if (LocationMap.empty())
	{
		ZEN_INFO("garbage collect SKIPPED, for '{}', container is empty", m_Config.RootDirectory / m_ContainerBaseName);
		return;
	}

	TotalChunkCount = LocationMap.size();

	std::unordered_map<uint32_t, size_t> BlockIndexToChunkMapIndex;
	std::vector<std::vector<IoHash>>	 KeepChunks;
	std::vector<std::vector<IoHash>>	 DeleteChunks;

	BlockIndexToChunkMapIndex.reserve(BlockCount);
	KeepChunks.reserve(BlockCount);
	DeleteChunks.reserve(BlockCount);
	size_t GuesstimateCountPerBlock = TotalChunkCount / BlockCount / 2;

	std::vector<IoHash> TotalChunkHashes;
	TotalChunkHashes.reserve(TotalChunkCount);
	for (const auto& Entry : LocationMap)
	{
		TotalChunkHashes.push_back(Entry.first);
	}

	uint64_t DeleteCount = 0;

	uint64_t NewTotalSize = 0;
	GcCtx.FilterCas(TotalChunkHashes, [&](const IoHash& ChunkHash, bool Keep) {
		auto						  KeyIt		 = LocationMap.find(ChunkHash);
		const BlockStoreDiskLocation& Location	 = KeyIt->second;
		uint32_t					  BlockIndex = Location.GetBlockIndex();

		if (static_cast<uint64_t>(BlockIndex) == ExcludeBlockIndex)
		{
			return;
		}

		auto   BlockIndexPtr = BlockIndexToChunkMapIndex.find(BlockIndex);
		size_t ChunkMapIndex = 0;
		if (BlockIndexPtr == BlockIndexToChunkMapIndex.end())
		{
			ChunkMapIndex						  = KeepChunks.size();
			BlockIndexToChunkMapIndex[BlockIndex] = ChunkMapIndex;
			KeepChunks.resize(ChunkMapIndex + 1);
			KeepChunks.back().reserve(GuesstimateCountPerBlock);
			DeleteChunks.resize(ChunkMapIndex + 1);
			DeleteChunks.back().reserve(GuesstimateCountPerBlock);
		}
		else
		{
			ChunkMapIndex = BlockIndexPtr->second;
		}
		if (Keep)
		{
			std::vector<IoHash>& ChunkMap = KeepChunks[ChunkMapIndex];
			ChunkMap.push_back(ChunkHash);
			NewTotalSize += Location.GetSize();
		}
		else
		{
			std::vector<IoHash>& ChunkMap = DeleteChunks[ChunkMapIndex];
			ChunkMap.push_back(ChunkHash);
			DeleteCount++;
		}
	});

	std::unordered_set<uint32_t> BlocksToReWrite;
	BlocksToReWrite.reserve(BlockIndexToChunkMapIndex.size());
	for (const auto& Entry : BlockIndexToChunkMapIndex)
	{
		uint32_t				   BlockIndex	 = Entry.first;
		size_t					   ChunkMapIndex = Entry.second;
		const std::vector<IoHash>& ChunkMap		 = DeleteChunks[ChunkMapIndex];
		if (ChunkMap.empty())
		{
			continue;
		}
		BlocksToReWrite.insert(BlockIndex);
	}

	const bool PerformDelete = GcCtx.IsDeletionMode() && GcCtx.CollectSmallObjects();
	if (!PerformDelete)
	{
		uint64_t TotalSize = m_TotalSize.load(std::memory_order_relaxed);
		ZEN_INFO("garbage collect for '{}' DISABLED, found #{} {} chunks of total #{} {}",
				 m_Config.RootDirectory / m_ContainerBaseName,
				 DeleteCount,
				 NiceBytes(TotalSize - NewTotalSize),
				 TotalChunkCount,
				 NiceBytes(TotalSize));
		return;
	}

	// Move all chunks in blocks that have chunks removed to new blocks

	Ref<BlockStoreFile> NewBlockFile;
	uint64_t			WriteOffset	  = 0;
	uint32_t			NewBlockIndex = 0;
	DeletedChunks.reserve(DeleteCount);

	auto UpdateLocations = [this](const std::span<CasDiskIndexEntry>& Entries) {
		for (const CasDiskIndexEntry& Entry : Entries)
		{
			if (Entry.Flags & CasDiskIndexEntry::kTombstone)
			{
				auto	 KeyIt	   = m_LocationMap.find(Entry.Key);
				uint64_t ChunkSize = KeyIt->second.GetSize();
				m_TotalSize.fetch_sub(ChunkSize);
				m_LocationMap.erase(KeyIt);
				continue;
			}
			m_LocationMap[Entry.Key] = Entry.Location;
		}
	};

	std::unordered_map<IoHash, BlockStoreDiskLocation> MovedBlockChunks;
	for (uint32_t BlockIndex : BlocksToReWrite)
	{
		const size_t ChunkMapIndex = BlockIndexToChunkMapIndex[BlockIndex];

		Ref<BlockStoreFile> OldBlockFile;
		{
			RwLock::SharedLockScope _i(m_LocationMapLock);
			OldBlockFile = m_ChunkBlocks[BlockIndex];
		}

		const std::vector<IoHash>& KeepMap = KeepChunks[ChunkMapIndex];
		if (KeepMap.empty())
		{
			const std::vector<IoHash>&	   DeleteMap  = DeleteChunks[ChunkMapIndex];
			std::vector<CasDiskIndexEntry> LogEntries = MakeCasDiskEntries({}, DeleteMap);
			m_CasLog.Append(LogEntries);
			m_CasLog.Flush();
			{
				RwLock::ExclusiveLockScope _i(m_LocationMapLock);
				Stopwatch				   Timer;
				const auto				   __ = MakeGuard([&Timer, &ReadBlockTimeUs, &ReadBlockLongestTimeUs] {
					uint64_t ElapsedUs = Timer.GetElapsedTimeUs();
					ReadBlockTimeUs += ElapsedUs;
					ReadBlockLongestTimeUs = std::max(ElapsedUs, ReadBlockLongestTimeUs);
				});
				UpdateLocations(LogEntries);
				m_ChunkBlocks[BlockIndex] = nullptr;
			}
			DeletedChunks.insert(DeletedChunks.end(), DeleteMap.begin(), DeleteMap.end());
			ZEN_DEBUG("marking cas store file in '{}' for delete , block #{}, '{}'",
					  m_ContainerBaseName,
					  BlockIndex,
					  OldBlockFile->GetPath());
			std::error_code Ec;
			OldBlockFile->MarkAsDeleteOnClose(Ec);
			if (Ec)
			{
				ZEN_WARN("Failed to flag file '{}' for deletion: '{}'", OldBlockFile->GetPath(), Ec.message());
			}
			continue;
		}

		std::vector<uint8_t> Chunk;
		for (const IoHash& ChunkHash : KeepMap)
		{
			auto					 KeyIt		   = LocationMap.find(ChunkHash);
			const BlockStoreLocation ChunkLocation = KeyIt->second.Get(m_PayloadAlignment);
			Chunk.resize(ChunkLocation.Size);
			OldBlockFile->Read(Chunk.data(), Chunk.size(), ChunkLocation.Offset);

			if (!NewBlockFile || (WriteOffset + Chunk.size() > m_MaxBlockSize))
			{
				uint32_t					   NextBlockIndex = m_WriteBlockIndex.load(std::memory_order_relaxed);
				std::vector<CasDiskIndexEntry> LogEntries	  = MakeCasDiskEntries(MovedBlockChunks, {});
				m_CasLog.Append(LogEntries);
				m_CasLog.Flush();

				if (NewBlockFile)
				{
					NewBlockFile->Truncate(WriteOffset);
					NewBlockFile->Flush();
				}
				{
					RwLock::ExclusiveLockScope __(m_LocationMapLock);
					Stopwatch				   Timer;
					const auto				   ___ = MakeGuard([&Timer, &ReadBlockTimeUs, &ReadBlockLongestTimeUs] {
						uint64_t ElapsedUs = Timer.GetElapsedTimeUs();
						ReadBlockTimeUs += ElapsedUs;
						ReadBlockLongestTimeUs = std::max(ElapsedUs, ReadBlockLongestTimeUs);
					});
					UpdateLocations(LogEntries);
					if (m_ChunkBlocks.size() == BlockStoreDiskLocation::MaxBlockIndex)
					{
						ZEN_ERROR("unable to allocate a new block in '{}', count limit {} exeeded",
								  m_Config.RootDirectory / m_ContainerBaseName,
								  static_cast<uint64_t>(std::numeric_limits<uint32_t>::max()) + 1);
						return;
					}
					while (m_ChunkBlocks.contains(NextBlockIndex))
					{
						NextBlockIndex = (NextBlockIndex + 1) & BlockStoreDiskLocation::MaxBlockIndex;
					}
					std::filesystem::path NewBlockPath = GetBlockPath(m_BlocksBasePath, NextBlockIndex);
					NewBlockFile					   = new BlockStoreFile(NewBlockPath);
					m_ChunkBlocks[NextBlockIndex]	   = NewBlockFile;
				}

				MovedCount += MovedBlockChunks.size();
				MovedBlockChunks.clear();

				std::error_code Error;
				DiskSpace		Space = DiskSpaceInfo(m_Config.RootDirectory, Error);
				if (Error)
				{
					ZEN_ERROR("get disk space in '{}' FAILED, reason: '{}'", m_Config.RootDirectory, Error.message());
					return;
				}
				if (Space.Free < m_MaxBlockSize)
				{
					uint64_t ReclaimedSpace = GcCtx.ClaimGCReserve();
					if (Space.Free + ReclaimedSpace < m_MaxBlockSize)
					{
						ZEN_WARN("garbage collect for '{}' FAILED, required disk space {}, free {}",
								 m_Config.RootDirectory / m_ContainerBaseName,
								 m_MaxBlockSize,
								 NiceBytes(Space.Free + ReclaimedSpace));
						RwLock::ExclusiveLockScope _l(m_LocationMapLock);
						Stopwatch				   Timer;
						const auto				   __ = MakeGuard([&Timer, &ReadBlockTimeUs, &ReadBlockLongestTimeUs] {
							uint64_t ElapsedUs = Timer.GetElapsedTimeUs();
							ReadBlockTimeUs += ElapsedUs;
							ReadBlockLongestTimeUs = std::max(ElapsedUs, ReadBlockLongestTimeUs);
						});
						m_ChunkBlocks.erase(NextBlockIndex);
						return;
					}

					ZEN_INFO("using gc reserve for '{}', reclaimed {}, disk free {}",
							 m_Config.RootDirectory / m_ContainerBaseName,
							 ReclaimedSpace,
							 NiceBytes(Space.Free + ReclaimedSpace));
				}
				NewBlockFile->Create(m_MaxBlockSize);
				NewBlockIndex = NextBlockIndex;
				WriteOffset	  = 0;
			}

			NewBlockFile->Write(Chunk.data(), Chunk.size(), WriteOffset);
			MovedBlockChunks.emplace(
				ChunkHash,
				BlockStoreDiskLocation({.BlockIndex = NewBlockIndex, .Offset = WriteOffset, .Size = Chunk.size()}, m_PayloadAlignment));
			WriteOffset = RoundUp(WriteOffset + Chunk.size(), m_PayloadAlignment);
		}
		Chunk.clear();
		if (NewBlockFile)
		{
			NewBlockFile->Truncate(WriteOffset);
			NewBlockFile->Flush();
			NewBlockFile = {};
		}

		const std::vector<IoHash>&	   DeleteMap  = DeleteChunks[ChunkMapIndex];
		std::vector<CasDiskIndexEntry> LogEntries = MakeCasDiskEntries(MovedBlockChunks, DeleteMap);
		m_CasLog.Append(LogEntries);
		m_CasLog.Flush();
		{
			RwLock::ExclusiveLockScope __(m_LocationMapLock);
			Stopwatch				   Timer;
			const auto				   ___ = MakeGuard([&Timer, &ReadBlockTimeUs, &ReadBlockLongestTimeUs] {
				uint64_t ElapsedUs = Timer.GetElapsedTimeUs();
				ReadBlockTimeUs += ElapsedUs;
				ReadBlockLongestTimeUs = std::max(ElapsedUs, ReadBlockLongestTimeUs);
			});
			UpdateLocations(LogEntries);
			m_ChunkBlocks[BlockIndex] = nullptr;
		}
		MovedCount += MovedBlockChunks.size();
		DeletedChunks.insert(DeletedChunks.end(), DeleteMap.begin(), DeleteMap.end());
		MovedBlockChunks.clear();

		ZEN_DEBUG("marking cas store file in '{}' for delete , block #{}, '{}'", m_ContainerBaseName, BlockIndex, OldBlockFile->GetPath());
		std::error_code Ec;
		OldBlockFile->MarkAsDeleteOnClose(Ec);
		if (Ec)
		{
			ZEN_WARN("Failed to flag file '{}' for deletion: '{}'", OldBlockFile->GetPath(), Ec.message());
		}
		OldBlockFile = nullptr;
	}

	for (const IoHash& ChunkHash : DeletedChunks)
	{
		DeletedSize += LocationMap[ChunkHash].GetSize();
	}

	GcCtx.DeletedCas(DeletedChunks);
}

void
CasContainerStrategy::MakeIndexSnapshot()
{
	ZEN_INFO("write store snapshot for '{}'", m_Config.RootDirectory / m_ContainerBaseName);
	uint64_t   EntryCount = 0;
	Stopwatch  Timer;
	const auto _ = MakeGuard([this, &EntryCount, &Timer] {
		ZEN_INFO("wrote store snapshot for '{}' containing #{} entries in {}",
				 m_Config.RootDirectory / m_ContainerBaseName,
				 EntryCount,
				 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
	});

	namespace fs = std::filesystem;

	fs::path IndexPath	   = GetIndexPath(m_Config.RootDirectory, m_ContainerBaseName);
	fs::path TempIndexPath = GetTempIndexPath(m_Config.RootDirectory, m_ContainerBaseName);

	// Move index away, we keep it if something goes wrong
	if (fs::is_regular_file(TempIndexPath))
	{
		fs::remove(TempIndexPath);
	}
	if (fs::is_regular_file(IndexPath))
	{
		fs::rename(IndexPath, TempIndexPath);
	}

	try
	{
		m_CasLog.Flush();

		// Write the current state of the location map to a new index state
		uint64_t					   LogCount = 0;
		std::vector<CasDiskIndexEntry> Entries;

		{
			RwLock::SharedLockScope __(m_InsertLock);
			RwLock::SharedLockScope ___(m_LocationMapLock);
			Entries.resize(m_LocationMap.size());

			uint64_t EntryIndex = 0;
			for (auto& Entry : m_LocationMap)
			{
				CasDiskIndexEntry& IndexEntry = Entries[EntryIndex++];
				IndexEntry.Key				  = Entry.first;
				IndexEntry.Location			  = Entry.second;
			}

			LogCount = m_CasLog.GetLogCount();
		}

		BasicFile ObjectIndexFile;
		ObjectIndexFile.Open(IndexPath, BasicFile::Mode::kTruncate);
		CasDiskIndexHeader Header = {.EntryCount	   = Entries.size(),
									 .LogPosition	   = LogCount,
									 .PayloadAlignment = gsl::narrow<uint32_t>(m_PayloadAlignment)};

		Header.Checksum = CasDiskIndexHeader::ComputeChecksum(Header);

		ObjectIndexFile.Write(&Header, sizeof(CasDiskIndexEntry), 0);
		ObjectIndexFile.Write(Entries.data(), Entries.size() * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexEntry));
		ObjectIndexFile.Flush();
		ObjectIndexFile.Close();
		EntryCount = Entries.size();
	}
	catch (std::exception& Err)
	{
		ZEN_ERROR("snapshot FAILED, reason: '{}'", Err.what());

		// Restore any previous snapshot

		if (fs::is_regular_file(TempIndexPath))
		{
			fs::remove(IndexPath);
			fs::rename(TempIndexPath, IndexPath);
		}
	}
	if (fs::is_regular_file(TempIndexPath))
	{
		fs::remove(TempIndexPath);
	}
}

uint64_t
CasContainerStrategy::ReadIndexFile()
{
	std::vector<CasDiskIndexEntry> Entries;
	std::filesystem::path		   IndexPath = GetIndexPath(m_Config.RootDirectory, m_ContainerBaseName);
	if (std::filesystem::is_regular_file(IndexPath))
	{
		Stopwatch  Timer;
		const auto _ = MakeGuard([this, &Entries, &Timer] {
			ZEN_INFO("read store '{}' index containing #{} entries in {}",
					 m_Config.RootDirectory / m_ContainerBaseName,
					 Entries.size(),
					 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
		});

		BasicFile ObjectIndexFile;
		ObjectIndexFile.Open(IndexPath, BasicFile::Mode::kRead);
		uint64_t Size = ObjectIndexFile.FileSize();
		if (Size >= sizeof(CasDiskIndexHeader))
		{
			uint64_t		   ExpectedEntryCount = (Size - sizeof(sizeof(CasDiskIndexHeader))) / sizeof(CasDiskIndexEntry);
			CasDiskIndexHeader Header;
			ObjectIndexFile.Read(&Header, sizeof(Header), 0);
			if ((Header.Magic == CasDiskIndexHeader::ExpectedMagic) && (Header.Version == CasDiskIndexHeader::CurrentVersion) &&
				(Header.Checksum == CasDiskIndexHeader::ComputeChecksum(Header)) && (Header.PayloadAlignment > 0) &&
				(Header.EntryCount <= ExpectedEntryCount))
			{
				Entries.resize(Header.EntryCount);
				ObjectIndexFile.Read(Entries.data(), Header.EntryCount * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexHeader));
				m_PayloadAlignment = Header.PayloadAlignment;

				std::string InvalidEntryReason;
				for (const CasDiskIndexEntry& Entry : Entries)
				{
					if (!ValidateEntry(Entry, InvalidEntryReason))
					{
						ZEN_WARN("skipping invalid entry in '{}', reason: '{}'", IndexPath, InvalidEntryReason);
						continue;
					}
					m_LocationMap[Entry.Key] = Entry.Location;
				}

				return Header.LogPosition;
			}
			else
			{
				ZEN_WARN("skipping invalid index file '{}'", IndexPath);
			}
		}
	}
	return 0;
}

uint64_t
CasContainerStrategy::ReadLog(uint64_t SkipEntryCount)
{
	std::vector<CasDiskIndexEntry> Entries;
	std::filesystem::path		   LogPath = GetLogPath(m_Config.RootDirectory, m_ContainerBaseName);
	if (std::filesystem::is_regular_file(LogPath))
	{
		Stopwatch  Timer;
		const auto _ = MakeGuard([this, &Entries, &Timer] {
			ZEN_INFO("read store '{}' log containing #{} entries in {}",
					 m_Config.RootDirectory / m_ContainerBaseName,
					 Entries.size(),
					 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
		});

		TCasLogFile<CasDiskIndexEntry> CasLog;
		CasLog.Open(LogPath, CasLogFile::Mode::kRead);
		if (CasLog.Initialize())
		{
			uint64_t EntryCount = CasLog.GetLogCount();
			if (EntryCount < SkipEntryCount)
			{
				ZEN_WARN("reading full log at '{}', reason: Log position from index snapshot is out of range", LogPath);
				SkipEntryCount = 0;
			}
			uint64_t ReadCount = EntryCount - SkipEntryCount;
			Entries.reserve(ReadCount);
			CasLog.Replay(
				[&](const CasDiskIndexEntry& Record) {
					std::string InvalidEntryReason;
					if (Record.Flags & CasDiskIndexEntry::kTombstone)
					{
						m_LocationMap.erase(Record.Key);
						return;
					}
					if (!ValidateEntry(Record, InvalidEntryReason))
					{
						ZEN_WARN("skipping invalid entry in '{}', reason: '{}'", LogPath, InvalidEntryReason);
						return;
					}
					m_LocationMap[Record.Key] = Record.Location;
				},
				SkipEntryCount);
			return ReadCount;
		}
	}
	return 0;
}

uint64_t
CasContainerStrategy::MigrateLegacyData(bool CleanSource)
{
	std::filesystem::path LegacyLogPath = GetLegacyLogPath(m_Config.RootDirectory, m_ContainerBaseName);

	if (!std::filesystem::is_regular_file(LegacyLogPath) || std::filesystem::file_size(LegacyLogPath) == 0)
	{
		return 0;
	}

	ZEN_INFO("migrating store '{}'", m_Config.RootDirectory / m_ContainerBaseName);

	std::filesystem::path LegacyDataPath  = GetLegacyDataPath(m_Config.RootDirectory, m_ContainerBaseName);
	std::filesystem::path LegacyIndexPath = GetLegacyIndexPath(m_Config.RootDirectory, m_ContainerBaseName);

	uint64_t   MigratedChunkCount = 0;
	uint32_t   MigratedBlockCount = 0;
	Stopwatch  MigrationTimer;
	uint64_t   TotalSize = 0;
	const auto _		 = MakeGuard([this, &MigrationTimer, &MigratedChunkCount, &MigratedBlockCount, &TotalSize] {
		ZEN_INFO("migrated store '{}' to #{} chunks in #{} blocks in {} ({})",
				 m_Config.RootDirectory / m_ContainerBaseName,
				 MigratedChunkCount,
				 MigratedBlockCount,
				 NiceTimeSpanMs(MigrationTimer.GetElapsedTimeMs()),
				 NiceBytes(TotalSize));
	});

	uint32_t WriteBlockIndex = 0;
	while (std::filesystem::exists(GetBlockPath(m_BlocksBasePath, WriteBlockIndex)))
	{
		++WriteBlockIndex;
	}

	std::error_code Error;
	DiskSpace		Space = DiskSpaceInfo(m_Config.RootDirectory, Error);
	if (Error)
	{
		ZEN_ERROR("get disk space in {} FAILED, reason: '{}'", m_Config.RootDirectory, Error.message());
		return 0;
	}

	if (Space.Free < m_MaxBlockSize)
	{
		ZEN_ERROR("legacy store migration from '{}' FAILED, required disk space {}, free {}",
				  m_Config.RootDirectory / m_ContainerBaseName,
				  m_MaxBlockSize,
				  NiceBytes(Space.Free));
		return 0;
	}

	BasicFile BlockFile;
	BlockFile.Open(LegacyDataPath, CleanSource ? BasicFile::Mode::kWrite : BasicFile::Mode::kRead);

	std::unordered_map<IoHash, LegacyCasDiskIndexEntry, IoHash::Hasher> LegacyDiskIndex;
	uint64_t															InvalidEntryCount = 0;

	TCasLogFile<LegacyCasDiskIndexEntry> LegacyCasLog;
	LegacyCasLog.Open(LegacyLogPath, CleanSource ? CasLogFile::Mode::kWrite : CasLogFile::Mode::kRead);
	{
		Stopwatch  Timer;
		const auto __ = MakeGuard([this, &LegacyDiskIndex, &Timer] {
			ZEN_INFO("read store '{}' legacy log containing #{} entries in {}",
					 m_Config.RootDirectory / m_ContainerBaseName,
					 LegacyDiskIndex.size(),
					 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
		});
		if (LegacyCasLog.Initialize())
		{
			LegacyDiskIndex.reserve(LegacyCasLog.GetLogCount());
			LegacyCasLog.Replay(
				[&](const LegacyCasDiskIndexEntry& Record) {
					std::string InvalidEntryReason;
					if (Record.Flags & LegacyCasDiskIndexEntry::kTombstone)
					{
						LegacyDiskIndex.erase(Record.Key);
						return;
					}
					if (!ValidateLegacyEntry(Record, InvalidEntryReason))
					{
						ZEN_WARN("skipping invalid entry in '{}', reason: '{}'", LegacyLogPath, InvalidEntryReason);
						InvalidEntryCount++;
						return;
					}
					LegacyDiskIndex.insert_or_assign(Record.Key, Record);
				},
				0);

			std::vector<IoHash> BadEntries;
			uint64_t			BlockFileSize = BlockFile.FileSize();
			for (const auto& Entry : LegacyDiskIndex)
			{
				const LegacyCasDiskIndexEntry& Record(Entry.second);
				if (Record.Location.GetOffset() + Record.Location.GetSize() <= BlockFileSize)
				{
					continue;
				}
				ZEN_WARN("skipping invalid entry in '{}', reason: location is outside of file", LegacyLogPath);
				BadEntries.push_back(Entry.first);
			}
			for (const IoHash& BadHash : BadEntries)
			{
				LegacyDiskIndex.erase(BadHash);
			}
			InvalidEntryCount += BadEntries.size();
		}
	}

	if (InvalidEntryCount)
	{
		ZEN_WARN("found #{} invalid entries in '{}'", InvalidEntryCount, m_Config.RootDirectory / m_ContainerBaseName);
	}

	if (LegacyDiskIndex.empty())
	{
		BlockFile.Close();
		LegacyCasLog.Close();
		if (CleanSource)
		{
			// Older versions of CasContainerStrategy expects the legacy files to exist if it can find
			// a CAS manifest and crashes on startup if they don't.
			// In order to not break startup when switching back an older version, lets just reset
			// the legacy data files to zero length.

			BasicFile LegacyLog;
			LegacyLog.Open(LegacyLogPath, BasicFile::Mode::kTruncate);
			BasicFile LegacySobs;
			LegacySobs.Open(LegacyDataPath, BasicFile::Mode::kTruncate);
			BasicFile LegacySidx;
			LegacySidx.Open(LegacyIndexPath, BasicFile::Mode::kTruncate);
		}
		return 0;
	}

	for (const auto& Entry : LegacyDiskIndex)
	{
		const LegacyCasDiskIndexEntry& Record(Entry.second);
		TotalSize += Record.Location.GetSize();
	}

	uint64_t RequiredDiskSpace	   = TotalSize + ((m_PayloadAlignment - 1) * LegacyDiskIndex.size());
	uint64_t MaxRequiredBlockCount = RoundUp(RequiredDiskSpace, m_MaxBlockSize) / m_MaxBlockSize;
	if (MaxRequiredBlockCount > BlockStoreDiskLocation::MaxBlockIndex)
	{
		ZEN_ERROR("legacy store migration from '{}' FAILED, required block count {}, possible {}",
				  m_Config.RootDirectory / m_ContainerBaseName,
				  MaxRequiredBlockCount,
				  BlockStoreDiskLocation::MaxBlockIndex);
		return 0;
	}

	constexpr const uint64_t DiskReserve = 1ul << 28;

	if (CleanSource)
	{
		if (Space.Free < (m_MaxBlockSize + DiskReserve))
		{
			ZEN_INFO("legacy store migration from '{}' aborted, not enough disk space available {} ({})",
					 m_Config.RootDirectory / m_ContainerBaseName,
					 NiceBytes(m_MaxBlockSize + DiskReserve),
					 NiceBytes(Space.Free));
			return 0;
		}
	}
	else
	{
		if (Space.Free < (RequiredDiskSpace + DiskReserve))
		{
			ZEN_INFO("legacy store migration from '{}' aborted, not enough disk space available {} ({})",
					 m_Config.RootDirectory / m_ContainerBaseName,
					 NiceBytes(RequiredDiskSpace + DiskReserve),
					 NiceBytes(Space.Free));
			return 0;
		}
	}

	std::filesystem::path LogPath = GetLogPath(m_Config.RootDirectory, m_ContainerBaseName);
	CreateDirectories(LogPath.parent_path());
	TCasLogFile<CasDiskIndexEntry> CasLog;
	CasLog.Open(LogPath, CasLogFile::Mode::kWrite);

	if (CleanSource && (MaxRequiredBlockCount < 2))
	{
		std::vector<CasDiskIndexEntry> LogEntries;
		LogEntries.reserve(LegacyDiskIndex.size());

		// We can use the block as is, just move it and add the blocks to our new log
		for (auto& Entry : LegacyDiskIndex)
		{
			const LegacyCasDiskIndexEntry& Record(Entry.second);

			BlockStoreLocation	   NewChunkLocation{WriteBlockIndex, Record.Location.GetOffset(), Record.Location.GetSize()};
			BlockStoreDiskLocation NewLocation(NewChunkLocation, m_PayloadAlignment);
			LogEntries.push_back(
				{.Key = Entry.second.Key, .Location = NewLocation, .ContentType = Record.ContentType, .Flags = Record.Flags});
		}
		std::filesystem::path BlockPath = GetBlockPath(m_BlocksBasePath, WriteBlockIndex);
		CreateDirectories(BlockPath.parent_path());
		BlockFile.Close();
		std::filesystem::rename(LegacyDataPath, BlockPath);
		CasLog.Append(LogEntries);
		for (const CasDiskIndexEntry& Entry : LogEntries)
		{
			m_LocationMap.insert_or_assign(Entry.Key, Entry.Location);
		}

		MigratedChunkCount += LogEntries.size();
		MigratedBlockCount++;
	}
	else
	{
		std::vector<IoHash> ChunkHashes;
		ChunkHashes.reserve(LegacyDiskIndex.size());
		for (const auto& Entry : LegacyDiskIndex)
		{
			ChunkHashes.push_back(Entry.first);
		}

		std::sort(begin(ChunkHashes), end(ChunkHashes), [&](IoHash Lhs, IoHash Rhs) {
			auto LhsKeyIt = LegacyDiskIndex.find(Lhs);
			auto RhsKeyIt = LegacyDiskIndex.find(Rhs);
			return LhsKeyIt->second.Location.GetOffset() < RhsKeyIt->second.Location.GetOffset();
		});

		uint64_t						BlockSize	= 0;
		uint64_t						BlockOffset = 0;
		std::vector<BlockStoreLocation> NewLocations;
		struct BlockData
		{
			std::vector<std::pair<IoHash, BlockStoreLocation>> Chunks;
			uint64_t										   BlockOffset;
			uint64_t										   BlockSize;
			uint32_t										   BlockIndex;
		};

		std::vector<BlockData>							   BlockRanges;
		std::vector<std::pair<IoHash, BlockStoreLocation>> Chunks;
		BlockRanges.reserve(MaxRequiredBlockCount);
		for (const IoHash& ChunkHash : ChunkHashes)
		{
			const LegacyCasDiskIndexEntry& LegacyEntry		   = LegacyDiskIndex[ChunkHash];
			const LegacyCasDiskLocation&   LegacyChunkLocation = LegacyEntry.Location;

			uint64_t ChunkOffset = LegacyChunkLocation.GetOffset();
			uint64_t ChunkSize	 = LegacyChunkLocation.GetSize();
			uint64_t ChunkEnd	 = ChunkOffset + ChunkSize;

			if (BlockSize == 0)
			{
				BlockOffset = ChunkOffset;
			}
			if ((ChunkEnd - BlockOffset) > m_MaxBlockSize)
			{
				BlockData BlockRange{.BlockOffset = BlockOffset, .BlockSize = BlockSize, .BlockIndex = WriteBlockIndex};
				BlockRange.Chunks.swap(Chunks);
				BlockRanges.push_back(BlockRange);

				WriteBlockIndex++;
				while (std::filesystem::exists(GetBlockPath(m_BlocksBasePath, WriteBlockIndex)))
				{
					++WriteBlockIndex;
				}
				BlockOffset = ChunkOffset;
				BlockSize	= 0;
			}
			BlockSize						 = RoundUp(BlockSize, m_PayloadAlignment);
			BlockStoreLocation ChunkLocation = {.BlockIndex = WriteBlockIndex, .Offset = ChunkOffset - BlockOffset, .Size = ChunkSize};
			Chunks.push_back({ChunkHash, ChunkLocation});
			BlockSize = ChunkEnd - BlockOffset;
		}
		if (BlockSize > 0)
		{
			BlockRanges.push_back(
				{.Chunks = std::move(Chunks), .BlockOffset = BlockOffset, .BlockSize = BlockSize, .BlockIndex = WriteBlockIndex});
		}
		Stopwatch WriteBlockTimer;

		std::reverse(BlockRanges.begin(), BlockRanges.end());
		std::vector<std::uint8_t> Buffer(1 << 28);
		for (size_t Idx = 0; Idx < BlockRanges.size(); ++Idx)
		{
			const BlockData& BlockRange = BlockRanges[Idx];
			if (Idx > 0)
			{
				uint64_t Remaining = BlockRange.BlockOffset + BlockRange.BlockSize;
				uint64_t Completed = BlockOffset + BlockSize - Remaining;
				uint64_t ETA	   = (WriteBlockTimer.GetElapsedTimeMs() * Remaining) / Completed;

				ZEN_INFO("migrating store '{}' {}/{} blocks, remaining {} ({}) ETA: {}",
						 m_Config.RootDirectory / m_ContainerBaseName,
						 Idx,
						 BlockRanges.size(),
						 NiceBytes(BlockRange.BlockOffset + BlockRange.BlockSize),
						 NiceBytes(BlockOffset + BlockSize),
						 NiceTimeSpanMs(ETA));
			}

			std::filesystem::path BlockPath = GetBlockPath(m_BlocksBasePath, BlockRange.BlockIndex);
			BlockStoreFile		  ChunkBlock(BlockPath);
			ChunkBlock.Create(BlockRange.BlockSize);
			uint64_t Offset = 0;
			while (Offset < BlockRange.BlockSize)
			{
				uint64_t Size = BlockRange.BlockSize - Offset;
				if (Size > Buffer.size())
				{
					Size = Buffer.size();
				}
				BlockFile.Read(Buffer.data(), Size, BlockRange.BlockOffset + Offset);
				ChunkBlock.Write(Buffer.data(), Size, Offset);
				Offset += Size;
			}
			ChunkBlock.Truncate(Offset);
			ChunkBlock.Flush();

			std::vector<CasDiskIndexEntry> LogEntries;
			LogEntries.reserve(BlockRange.Chunks.size());
			for (const auto& Entry : BlockRange.Chunks)
			{
				const LegacyCasDiskIndexEntry& LegacyEntry = LegacyDiskIndex[Entry.first];
				BlockStoreDiskLocation		   Location(Entry.second, m_PayloadAlignment);
				LogEntries.push_back(
					{.Key = Entry.first, .Location = Location, .ContentType = LegacyEntry.ContentType, .Flags = LegacyEntry.Flags});
			}
			CasLog.Append(LogEntries);
			for (const CasDiskIndexEntry& Entry : LogEntries)
			{
				m_LocationMap.insert_or_assign(Entry.Key, Entry.Location);
			}
			MigratedChunkCount += LogEntries.size();
			MigratedBlockCount++;

			if (CleanSource)
			{
				std::vector<LegacyCasDiskIndexEntry> LegacyLogEntries;
				LegacyLogEntries.reserve(BlockRange.Chunks.size());
				for (const auto& Entry : BlockRange.Chunks)
				{
					LegacyLogEntries.push_back({.Key = Entry.first, .Flags = LegacyCasDiskIndexEntry::kTombstone});
				}
				LegacyCasLog.Append(LegacyLogEntries);
				BlockFile.SetFileSize(BlockRange.BlockOffset);
			}
		}
	}

	BlockFile.Close();
	LegacyCasLog.Close();
	CasLog.Close();

	if (CleanSource)
	{
		// Older versions of CasContainerStrategy expects the legacy files to exist if it can find
		// a CAS manifest and crashes on startup if they don't.
		// In order to not break startup when switching back an older version, lets just reset
		// the legacy data files to zero length.

		BasicFile LegacyLog;
		LegacyLog.Open(LegacyLogPath, BasicFile::Mode::kTruncate);
		BasicFile LegacySobs;
		LegacySobs.Open(LegacyDataPath, BasicFile::Mode::kTruncate);
		BasicFile LegacySidx;
		LegacySidx.Open(LegacyIndexPath, BasicFile::Mode::kTruncate);
	}
	return MigratedChunkCount;
}

void
CasContainerStrategy::OpenContainer(bool IsNewStore)
{
	// Add .running file and delete on clean on close to detect bad termination
	m_TotalSize = 0;

	m_LocationMap.clear();

	std::filesystem::path BasePath = GetBasePath(m_Config.RootDirectory, m_ContainerBaseName);

	if (IsNewStore)
	{
		std::filesystem::path LegacyDataPath = GetLegacyDataPath(m_Config.RootDirectory, m_ContainerBaseName);
		std::filesystem::path LegacyLogPath	 = GetLegacyLogPath(m_Config.RootDirectory, m_ContainerBaseName);

		std::filesystem::remove(LegacyLogPath);
		std::filesystem::remove(LegacyDataPath);
		std::filesystem::remove_all(BasePath);
	}

	uint64_t LogPosition		 = ReadIndexFile();
	uint64_t LogEntryCount		 = ReadLog(LogPosition);
	uint64_t LegacyLogEntryCount = MigrateLegacyData(true);

	CreateDirectories(BasePath);

	std::filesystem::path LogPath = GetLogPath(m_Config.RootDirectory, m_ContainerBaseName);
	m_CasLog.Open(LogPath, CasLogFile::Mode::kWrite);

	std::unordered_set<uint32_t> KnownBlocks;
	for (const auto& Entry : m_LocationMap)
	{
		const BlockStoreDiskLocation& Location = Entry.second;
		m_TotalSize.fetch_add(Location.GetSize(), std::memory_order_seq_cst);
		KnownBlocks.insert(Location.GetBlockIndex());
	}

	if (std::filesystem::is_directory(m_BlocksBasePath))
	{
		std::vector<std::filesystem::path> FoldersToScan;
		FoldersToScan.push_back(m_BlocksBasePath);
		size_t FolderOffset = 0;
		while (FolderOffset < FoldersToScan.size())
		{
			for (const std::filesystem::directory_entry& Entry : std::filesystem::directory_iterator(FoldersToScan[FolderOffset]))
			{
				if (Entry.is_directory())
				{
					FoldersToScan.push_back(Entry.path());
					continue;
				}
				if (Entry.is_regular_file())
				{
					const std::filesystem::path Path = Entry.path();
					if (Path.extension() != DataExtension)
					{
						continue;
					}
					std::string FileName = Path.stem().string();
					uint32_t	BlockIndex;
					bool		OK = ParseHexNumber(FileName, BlockIndex);
					if (!OK)
					{
						continue;
					}
					if (!KnownBlocks.contains(BlockIndex))
					{
						// Log removing unreferenced block
						// Clear out unused blocks
						ZEN_INFO("removing unused block for '{}' at '{}'", m_ContainerBaseName, Path);
						std::error_code Ec;
						std::filesystem::remove(Path, Ec);
						if (Ec)
						{
							ZEN_WARN("Failed to delete file '{}' reason: '{}'", Path, Ec.message());
						}
						continue;
					}
					Ref<BlockStoreFile> BlockFile = new BlockStoreFile(Path);
					BlockFile->Open();
					m_ChunkBlocks[BlockIndex] = BlockFile;
				}
			}
			++FolderOffset;
		}
	}
	else
	{
		CreateDirectories(m_BlocksBasePath);
	}

	if (IsNewStore || ((LogEntryCount + LegacyLogEntryCount) > 0))
	{
		MakeIndexSnapshot();
	}

	// TODO: should validate integrity of container files here
}

//////////////////////////////////////////////////////////////////////////

#if ZEN_WITH_TESTS

namespace {
	static IoBuffer CreateChunk(uint64_t Size)
	{
		static std::random_device rd;
		static std::mt19937		  g(rd());

		std::vector<uint8_t> Values;
		Values.resize(Size);
		for (size_t Idx = 0; Idx < Size; ++Idx)
		{
			Values[Idx] = static_cast<uint8_t>(Idx);
		}
		std::shuffle(Values.begin(), Values.end(), g);

		return IoBufferBuilder::MakeCloneFromMemory(Values.data(), Values.size());
	}
}  // namespace

TEST_CASE("compactcas.hex")
{
	uint32_t	Value;
	std::string HexString;
	CHECK(!ParseHexNumber("", Value));
	char Hex[9];

	ToHexNumber(0u, Hex);
	HexString = std::string(Hex);
	CHECK(ParseHexNumber(HexString, Value));
	CHECK(Value == 0u);

	ToHexNumber(std::numeric_limits<std::uint32_t>::max(), Hex);
	HexString = std::string(Hex);
	CHECK(HexString == "ffffffff");
	CHECK(ParseHexNumber(HexString, Value));
	CHECK(Value == std::numeric_limits<std::uint32_t>::max());

	ToHexNumber(0xadf14711u, Hex);
	HexString = std::string(Hex);
	CHECK(HexString == "adf14711");
	CHECK(ParseHexNumber(HexString, Value));
	CHECK(Value == 0xadf14711u);

	ToHexNumber(0x80000000u, Hex);
	HexString = std::string(Hex);
	CHECK(HexString == "80000000");
	CHECK(ParseHexNumber(HexString, Value));
	CHECK(Value == 0x80000000u);

	ToHexNumber(0x718293a4u, Hex);
	HexString = std::string(Hex);
	CHECK(HexString == "718293a4");
	CHECK(ParseHexNumber(HexString, Value));
	CHECK(Value == 0x718293a4u);
}

TEST_CASE("compactcas.compact.gc")
{
	ScopedTemporaryDirectory TempDir;

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	const int kIterationCount = 1000;

	std::vector<IoHash> Keys(kIterationCount);

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 65536, 16, true);

		for (int i = 0; i < kIterationCount; ++i)
		{
			CbObjectWriter Cbo;
			Cbo << "id" << i;
			CbObject Obj = Cbo.Save();

			IoBuffer	 ObjBuffer = Obj.GetBuffer().AsIoBuffer();
			const IoHash Hash	   = HashBuffer(ObjBuffer);

			Cas.InsertChunk(ObjBuffer, Hash);

			Keys[i] = Hash;
		}

		for (int i = 0; i < kIterationCount; ++i)
		{
			IoBuffer Chunk = Cas.FindChunk(Keys[i]);

			CHECK(!!Chunk);

			CbObject Value = LoadCompactBinaryObject(Chunk);

			CHECK_EQ(Value["id"].AsInt32(), i);
		}
	}

	// Validate that we can still read the inserted data after closing
	// the original cas store

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 65536, 16, false);

		for (int i = 0; i < kIterationCount; ++i)
		{
			IoBuffer Chunk = Cas.FindChunk(Keys[i]);

			CHECK(!!Chunk);

			CbObject Value = LoadCompactBinaryObject(Chunk);

			CHECK_EQ(Value["id"].AsInt32(), i);
		}
	}
}

TEST_CASE("compactcas.compact.totalsize")
{
	std::random_device rd;
	std::mt19937	   g(rd());

	//	for (uint32_t i = 0; i < 100; ++i)
	{
		ScopedTemporaryDirectory TempDir;

		CasStoreConfiguration CasConfig;
		CasConfig.RootDirectory = TempDir.Path();

		CreateDirectories(CasConfig.RootDirectory);

		const uint64_t kChunkSize  = 1024;
		const int32_t  kChunkCount = 16;

		{
			CasGc				 Gc;
			CasContainerStrategy Cas(CasConfig, Gc);
			Cas.Initialize("test", 65536, 16, true);

			for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
			{
				IoBuffer			   Chunk		= CreateChunk(kChunkSize);
				const IoHash		   Hash			= HashBuffer(Chunk);
				CasStore::InsertResult InsertResult = Cas.InsertChunk(Chunk, Hash);
				ZEN_ASSERT(InsertResult.New);
			}

			const uint64_t TotalSize = Cas.StorageSize().DiskSize;
			CHECK_EQ(kChunkSize * kChunkCount, TotalSize);
		}

		{
			CasGc				 Gc;
			CasContainerStrategy Cas(CasConfig, Gc);
			Cas.Initialize("test", 65536, 16, false);

			const uint64_t TotalSize = Cas.StorageSize().DiskSize;
			CHECK_EQ(kChunkSize * kChunkCount, TotalSize);
		}

		// Re-open again, this time we should have a snapshot
		{
			CasGc				 Gc;
			CasContainerStrategy Cas(CasConfig, Gc);
			Cas.Initialize("test", 65536, 16, false);

			const uint64_t TotalSize = Cas.StorageSize().DiskSize;
			CHECK_EQ(kChunkSize * kChunkCount, TotalSize);
		}
	}
}

TEST_CASE("compactcas.gc.basic")
{
	ScopedTemporaryDirectory TempDir;

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	CasGc				 Gc;
	CasContainerStrategy Cas(CasConfig, Gc);
	Cas.Initialize("cb", 65536, 1 << 4, true);

	IoBuffer Chunk	   = CreateChunk(128);
	IoHash	 ChunkHash = IoHash::HashBuffer(Chunk);

	const CasStore::InsertResult InsertResult = Cas.InsertChunk(Chunk, ChunkHash);
	CHECK(InsertResult.New);
	Cas.Flush();

	GcContext GcCtx;
	GcCtx.CollectSmallObjects(true);

	Cas.CollectGarbage(GcCtx);

	CHECK(!Cas.HaveChunk(ChunkHash));
}

TEST_CASE("compactcas.gc.removefile")
{
	ScopedTemporaryDirectory TempDir;

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	IoBuffer Chunk	   = CreateChunk(128);
	IoHash	 ChunkHash = IoHash::HashBuffer(Chunk);
	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("cb", 65536, 1 << 4, true);

		const CasStore::InsertResult InsertResult = Cas.InsertChunk(Chunk, ChunkHash);
		CHECK(InsertResult.New);
		const CasStore::InsertResult InsertResultDup = Cas.InsertChunk(Chunk, ChunkHash);
		CHECK(!InsertResultDup.New);
		Cas.Flush();
	}

	CasGc				 Gc;
	CasContainerStrategy Cas(CasConfig, Gc);
	Cas.Initialize("cb", 65536, 1 << 4, false);

	GcContext GcCtx;
	GcCtx.CollectSmallObjects(true);

	Cas.CollectGarbage(GcCtx);

	CHECK(!Cas.HaveChunk(ChunkHash));
}

TEST_CASE("compactcas.gc.compact")
{
	//	for (uint32_t i = 0; i < 100; ++i)
	{
		ScopedTemporaryDirectory TempDir;

		CasStoreConfiguration CasConfig;
		CasConfig.RootDirectory = TempDir.Path();
		CreateDirectories(CasConfig.RootDirectory);

		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("cb", 2048, 1 << 4, true);

		uint64_t			  ChunkSizes[9] = {128, 541, 1023, 781, 218, 37, 4, 997, 5};
		std::vector<IoBuffer> Chunks;
		Chunks.reserve(9);
		for (uint64_t Size : ChunkSizes)
		{
			Chunks.push_back(CreateChunk(Size));
		}

		std::vector<IoHash> ChunkHashes;
		ChunkHashes.reserve(9);
		for (const IoBuffer& Chunk : Chunks)
		{
			ChunkHashes.push_back(IoHash::HashBuffer(Chunk.Data(), Chunk.Size()));
		}

		CHECK(Cas.InsertChunk(Chunks[0], ChunkHashes[0]).New);
		CHECK(Cas.InsertChunk(Chunks[1], ChunkHashes[1]).New);
		CHECK(Cas.InsertChunk(Chunks[2], ChunkHashes[2]).New);
		CHECK(Cas.InsertChunk(Chunks[3], ChunkHashes[3]).New);
		CHECK(Cas.InsertChunk(Chunks[4], ChunkHashes[4]).New);
		CHECK(Cas.InsertChunk(Chunks[5], ChunkHashes[5]).New);
		CHECK(Cas.InsertChunk(Chunks[6], ChunkHashes[6]).New);
		CHECK(Cas.InsertChunk(Chunks[7], ChunkHashes[7]).New);
		CHECK(Cas.InsertChunk(Chunks[8], ChunkHashes[8]).New);

		CHECK(Cas.HaveChunk(ChunkHashes[0]));
		CHECK(Cas.HaveChunk(ChunkHashes[1]));
		CHECK(Cas.HaveChunk(ChunkHashes[2]));
		CHECK(Cas.HaveChunk(ChunkHashes[3]));
		CHECK(Cas.HaveChunk(ChunkHashes[4]));
		CHECK(Cas.HaveChunk(ChunkHashes[5]));
		CHECK(Cas.HaveChunk(ChunkHashes[6]));
		CHECK(Cas.HaveChunk(ChunkHashes[7]));
		CHECK(Cas.HaveChunk(ChunkHashes[8]));

		uint64_t InitialSize = Cas.StorageSize().DiskSize;

		// Keep first and last
		{
			GcContext GcCtx;
			GcCtx.CollectSmallObjects(true);

			std::vector<IoHash> KeepChunks;
			KeepChunks.push_back(ChunkHashes[0]);
			KeepChunks.push_back(ChunkHashes[8]);
			GcCtx.ContributeCas(KeepChunks);

			Cas.Flush();
			Cas.CollectGarbage(GcCtx);

			CHECK(Cas.HaveChunk(ChunkHashes[0]));
			CHECK(!Cas.HaveChunk(ChunkHashes[1]));
			CHECK(!Cas.HaveChunk(ChunkHashes[2]));
			CHECK(!Cas.HaveChunk(ChunkHashes[3]));
			CHECK(!Cas.HaveChunk(ChunkHashes[4]));
			CHECK(!Cas.HaveChunk(ChunkHashes[5]));
			CHECK(!Cas.HaveChunk(ChunkHashes[6]));
			CHECK(!Cas.HaveChunk(ChunkHashes[7]));
			CHECK(Cas.HaveChunk(ChunkHashes[8]));

			CHECK(ChunkHashes[0] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[0])));
			CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));
		}

		Cas.InsertChunk(Chunks[1], ChunkHashes[1]);
		Cas.InsertChunk(Chunks[2], ChunkHashes[2]);
		Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
		Cas.InsertChunk(Chunks[4], ChunkHashes[4]);
		Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
		Cas.InsertChunk(Chunks[6], ChunkHashes[6]);
		Cas.InsertChunk(Chunks[7], ChunkHashes[7]);

		// Keep last
		{
			GcContext GcCtx;
			GcCtx.CollectSmallObjects(true);
			std::vector<IoHash> KeepChunks;
			KeepChunks.push_back(ChunkHashes[8]);
			GcCtx.ContributeCas(KeepChunks);

			Cas.Flush();
			Cas.CollectGarbage(GcCtx);

			CHECK(!Cas.HaveChunk(ChunkHashes[0]));
			CHECK(!Cas.HaveChunk(ChunkHashes[1]));
			CHECK(!Cas.HaveChunk(ChunkHashes[2]));
			CHECK(!Cas.HaveChunk(ChunkHashes[3]));
			CHECK(!Cas.HaveChunk(ChunkHashes[4]));
			CHECK(!Cas.HaveChunk(ChunkHashes[5]));
			CHECK(!Cas.HaveChunk(ChunkHashes[6]));
			CHECK(!Cas.HaveChunk(ChunkHashes[7]));
			CHECK(Cas.HaveChunk(ChunkHashes[8]));

			CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));

			Cas.InsertChunk(Chunks[1], ChunkHashes[1]);
			Cas.InsertChunk(Chunks[2], ChunkHashes[2]);
			Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
			Cas.InsertChunk(Chunks[4], ChunkHashes[4]);
			Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
			Cas.InsertChunk(Chunks[6], ChunkHashes[6]);
			Cas.InsertChunk(Chunks[7], ChunkHashes[7]);
		}

		// Keep mixed
		{
			GcContext GcCtx;
			GcCtx.CollectSmallObjects(true);
			std::vector<IoHash> KeepChunks;
			KeepChunks.push_back(ChunkHashes[1]);
			KeepChunks.push_back(ChunkHashes[4]);
			KeepChunks.push_back(ChunkHashes[7]);
			GcCtx.ContributeCas(KeepChunks);

			Cas.Flush();
			Cas.CollectGarbage(GcCtx);

			CHECK(!Cas.HaveChunk(ChunkHashes[0]));
			CHECK(Cas.HaveChunk(ChunkHashes[1]));
			CHECK(!Cas.HaveChunk(ChunkHashes[2]));
			CHECK(!Cas.HaveChunk(ChunkHashes[3]));
			CHECK(Cas.HaveChunk(ChunkHashes[4]));
			CHECK(!Cas.HaveChunk(ChunkHashes[5]));
			CHECK(!Cas.HaveChunk(ChunkHashes[6]));
			CHECK(Cas.HaveChunk(ChunkHashes[7]));
			CHECK(!Cas.HaveChunk(ChunkHashes[8]));

			CHECK(ChunkHashes[1] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[1])));
			CHECK(ChunkHashes[4] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[4])));
			CHECK(ChunkHashes[7] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[7])));

			Cas.InsertChunk(Chunks[0], ChunkHashes[0]);
			Cas.InsertChunk(Chunks[2], ChunkHashes[2]);
			Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
			Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
			Cas.InsertChunk(Chunks[6], ChunkHashes[6]);
			Cas.InsertChunk(Chunks[8], ChunkHashes[8]);
		}

		// Keep multiple at end
		{
			GcContext GcCtx;
			GcCtx.CollectSmallObjects(true);
			std::vector<IoHash> KeepChunks;
			KeepChunks.push_back(ChunkHashes[6]);
			KeepChunks.push_back(ChunkHashes[7]);
			KeepChunks.push_back(ChunkHashes[8]);
			GcCtx.ContributeCas(KeepChunks);

			Cas.Flush();
			Cas.CollectGarbage(GcCtx);

			CHECK(!Cas.HaveChunk(ChunkHashes[0]));
			CHECK(!Cas.HaveChunk(ChunkHashes[1]));
			CHECK(!Cas.HaveChunk(ChunkHashes[2]));
			CHECK(!Cas.HaveChunk(ChunkHashes[3]));
			CHECK(!Cas.HaveChunk(ChunkHashes[4]));
			CHECK(!Cas.HaveChunk(ChunkHashes[5]));
			CHECK(Cas.HaveChunk(ChunkHashes[6]));
			CHECK(Cas.HaveChunk(ChunkHashes[7]));
			CHECK(Cas.HaveChunk(ChunkHashes[8]));

			CHECK(ChunkHashes[6] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[6])));
			CHECK(ChunkHashes[7] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[7])));
			CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));

			Cas.InsertChunk(Chunks[0], ChunkHashes[0]);
			Cas.InsertChunk(Chunks[1], ChunkHashes[1]);
			Cas.InsertChunk(Chunks[2], ChunkHashes[2]);
			Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
			Cas.InsertChunk(Chunks[4], ChunkHashes[4]);
			Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
		}

		// Keep every other
		{
			GcContext GcCtx;
			GcCtx.CollectSmallObjects(true);
			std::vector<IoHash> KeepChunks;
			KeepChunks.push_back(ChunkHashes[0]);
			KeepChunks.push_back(ChunkHashes[2]);
			KeepChunks.push_back(ChunkHashes[4]);
			KeepChunks.push_back(ChunkHashes[6]);
			KeepChunks.push_back(ChunkHashes[8]);
			GcCtx.ContributeCas(KeepChunks);

			Cas.Flush();
			Cas.CollectGarbage(GcCtx);

			CHECK(Cas.HaveChunk(ChunkHashes[0]));
			CHECK(!Cas.HaveChunk(ChunkHashes[1]));
			CHECK(Cas.HaveChunk(ChunkHashes[2]));
			CHECK(!Cas.HaveChunk(ChunkHashes[3]));
			CHECK(Cas.HaveChunk(ChunkHashes[4]));
			CHECK(!Cas.HaveChunk(ChunkHashes[5]));
			CHECK(Cas.HaveChunk(ChunkHashes[6]));
			CHECK(!Cas.HaveChunk(ChunkHashes[7]));
			CHECK(Cas.HaveChunk(ChunkHashes[8]));

			CHECK(ChunkHashes[0] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[0])));
			CHECK(ChunkHashes[2] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[2])));
			CHECK(ChunkHashes[4] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[4])));
			CHECK(ChunkHashes[6] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[6])));
			CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));

			Cas.InsertChunk(Chunks[1], ChunkHashes[1]);
			Cas.InsertChunk(Chunks[3], ChunkHashes[3]);
			Cas.InsertChunk(Chunks[5], ChunkHashes[5]);
			Cas.InsertChunk(Chunks[7], ChunkHashes[7]);
		}

		// Verify that we nicely appended blocks even after all GC operations
		CHECK(ChunkHashes[0] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[0])));
		CHECK(ChunkHashes[1] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[1])));
		CHECK(ChunkHashes[2] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[2])));
		CHECK(ChunkHashes[3] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[3])));
		CHECK(ChunkHashes[4] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[4])));
		CHECK(ChunkHashes[5] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[5])));
		CHECK(ChunkHashes[6] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[6])));
		CHECK(ChunkHashes[7] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[7])));
		CHECK(ChunkHashes[8] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[8])));

		uint64_t FinalSize = Cas.StorageSize().DiskSize;
		CHECK(InitialSize == FinalSize);
	}
}

TEST_CASE("compactcas.gc.deleteblockonopen")
{
	ScopedTemporaryDirectory TempDir;

	uint64_t			  ChunkSizes[20] = {128, 541, 311, 181, 218, 37, 4, 397, 5, 92, 551, 721, 31, 92, 16, 99, 131, 41, 541, 84};
	std::vector<IoBuffer> Chunks;
	Chunks.reserve(20);
	for (uint64_t Size : ChunkSizes)
	{
		Chunks.push_back(CreateChunk(Size));
	}

	std::vector<IoHash> ChunkHashes;
	ChunkHashes.reserve(20);
	for (const IoBuffer& Chunk : Chunks)
	{
		ChunkHashes.push_back(IoHash::HashBuffer(Chunk.Data(), Chunk.Size()));
	}

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);
	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 1024, 16, true);

		for (size_t i = 0; i < 20; i++)
		{
			CHECK(Cas.InsertChunk(Chunks[i], ChunkHashes[i]).New);
		}

		// GC every other block
		{
			GcContext GcCtx;
			GcCtx.CollectSmallObjects(true);
			std::vector<IoHash> KeepChunks;
			for (size_t i = 0; i < 20; i += 2)
			{
				KeepChunks.push_back(ChunkHashes[i]);
			}
			GcCtx.ContributeCas(KeepChunks);

			Cas.Flush();
			Cas.CollectGarbage(GcCtx);

			for (size_t i = 0; i < 20; i += 2)
			{
				CHECK(Cas.HaveChunk(ChunkHashes[i]));
				CHECK(!Cas.HaveChunk(ChunkHashes[i + 1]));
				CHECK(ChunkHashes[i] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[i])));
			}
		}
	}
	{
		// Re-open
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 1024, 16, false);

		for (size_t i = 0; i < 20; i += 2)
		{
			CHECK(Cas.HaveChunk(ChunkHashes[i]));
			CHECK(!Cas.HaveChunk(ChunkHashes[i + 1]));
			CHECK(ChunkHashes[i] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[i])));
		}
	}
}

TEST_CASE("compactcas.gc.handleopeniobuffer")
{
	ScopedTemporaryDirectory TempDir;

	uint64_t			  ChunkSizes[20] = {128, 541, 311, 181, 218, 37, 4, 397, 5, 92, 551, 721, 31, 92, 16, 99, 131, 41, 541, 84};
	std::vector<IoBuffer> Chunks;
	Chunks.reserve(20);
	for (const uint64_t& Size : ChunkSizes)
	{
		Chunks.push_back(CreateChunk(Size));
	}

	std::vector<IoHash> ChunkHashes;
	ChunkHashes.reserve(20);
	for (const IoBuffer& Chunk : Chunks)
	{
		ChunkHashes.push_back(IoHash::HashBuffer(Chunk.Data(), Chunk.Size()));
	}

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	CasGc				 Gc;
	CasContainerStrategy Cas(CasConfig, Gc);
	Cas.Initialize("test", 1024, 16, true);

	for (size_t i = 0; i < 20; i++)
	{
		CHECK(Cas.InsertChunk(Chunks[i], ChunkHashes[i]).New);
	}

	IoBuffer RetainChunk = Cas.FindChunk(ChunkHashes[5]);
	Cas.Flush();

	// GC everything
	GcContext GcCtx;
	GcCtx.CollectSmallObjects(true);
	Cas.CollectGarbage(GcCtx);

	for (size_t i = 0; i < 20; i++)
	{
		CHECK(!Cas.HaveChunk(ChunkHashes[i]));
	}

	CHECK(ChunkHashes[5] == IoHash::HashBuffer(RetainChunk));
}

TEST_CASE("compactcas.legacyconversion")
{
	ScopedTemporaryDirectory TempDir;

	uint64_t ChunkSizes[]	 = {2041, 1123, 1223, 1239, 341, 1412, 912, 774, 341, 431, 554, 1098, 2048, 339, 561, 16, 16, 2048, 2048};
	size_t	 ChunkCount		 = sizeof(ChunkSizes) / sizeof(uint64_t);
	size_t	 SingleBlockSize = 0;
	std::vector<IoBuffer> Chunks;
	Chunks.reserve(ChunkCount);
	for (uint64_t Size : ChunkSizes)
	{
		Chunks.push_back(CreateChunk(Size));
		SingleBlockSize += Size;
	}

	std::vector<IoHash> ChunkHashes;
	ChunkHashes.reserve(ChunkCount);
	for (const IoBuffer& Chunk : Chunks)
	{
		ChunkHashes.push_back(IoHash::HashBuffer(Chunk.Data(), Chunk.Size()));
	}

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = TempDir.Path();
	CreateDirectories(CasConfig.RootDirectory);

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", gsl::narrow<uint32_t>(SingleBlockSize * 2), 16, true);

		for (size_t i = 0; i < ChunkCount; i++)
		{
			CHECK(Cas.InsertChunk(Chunks[i], ChunkHashes[i]).New);
		}

		std::vector<IoHash> KeepChunks;
		for (size_t i = 0; i < ChunkCount; i += 2)
		{
			KeepChunks.push_back(ChunkHashes[i]);
		}
		GcContext GcCtx;
		GcCtx.CollectSmallObjects(true);
		GcCtx.ContributeCas(KeepChunks);
		Cas.Flush();
		Gc.CollectGarbage(GcCtx);
	}

	std::filesystem::path BlockPath		 = GetBlockPath(GetBlocksBasePath(CasConfig.RootDirectory, "test"), 1);
	std::filesystem::path LegacyDataPath = GetLegacyDataPath(CasConfig.RootDirectory, "test");
	std::filesystem::rename(BlockPath, LegacyDataPath);

	std::vector<CasDiskIndexEntry> LogEntries;
	std::filesystem::path		   IndexPath = GetIndexPath(CasConfig.RootDirectory, "test");
	if (std::filesystem::is_regular_file(IndexPath))
	{
		BasicFile ObjectIndexFile;
		ObjectIndexFile.Open(IndexPath, BasicFile::Mode::kRead);
		uint64_t Size = ObjectIndexFile.FileSize();
		if (Size >= sizeof(CasDiskIndexHeader))
		{
			uint64_t		   ExpectedEntryCount = (Size - sizeof(sizeof(CasDiskIndexHeader))) / sizeof(CasDiskIndexEntry);
			CasDiskIndexHeader Header;
			ObjectIndexFile.Read(&Header, sizeof(Header), 0);
			if (Header.Magic == CasDiskIndexHeader::ExpectedMagic && Header.Version == CasDiskIndexHeader::CurrentVersion &&
				Header.PayloadAlignment > 0 && Header.EntryCount == ExpectedEntryCount)
			{
				LogEntries.resize(Header.EntryCount);
				ObjectIndexFile.Read(LogEntries.data(), Header.EntryCount * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexHeader));
			}
		}
		ObjectIndexFile.Close();
		std::filesystem::remove(IndexPath);
	}

	std::filesystem::path LogPath = GetLogPath(CasConfig.RootDirectory, "test");
	{
		TCasLogFile<CasDiskIndexEntry> CasLog;
		CasLog.Open(LogPath, CasLogFile::Mode::kRead);
		LogEntries.reserve(CasLog.GetLogCount());
		CasLog.Replay([&](const CasDiskIndexEntry& Record) { LogEntries.push_back(Record); }, 0);
	}
	TCasLogFile<LegacyCasDiskIndexEntry> LegacyCasLog;
	std::filesystem::path				 LegacylogPath = GetLegacyLogPath(CasConfig.RootDirectory, "test");
	LegacyCasLog.Open(LegacylogPath, CasLogFile::Mode::kTruncate);

	for (const CasDiskIndexEntry& Entry : LogEntries)
	{
		BlockStoreLocation		Location = Entry.Location.Get(16);
		LegacyCasDiskLocation	LegacyLocation(Location.Offset, Location.Size);
		LegacyCasDiskIndexEntry LegacyEntry = {.Key			= Entry.Key,
											   .Location	= LegacyLocation,
											   .ContentType = Entry.ContentType,
											   .Flags		= Entry.Flags};
		LegacyCasLog.Append(LegacyEntry);
	}
	LegacyCasLog.Close();

	std::filesystem::remove_all(CasConfig.RootDirectory / "test");

	{
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 2048, 16, false);

		for (size_t i = 0; i < ChunkCount; i += 2)
		{
			CHECK(Cas.HaveChunk(ChunkHashes[i]));
			CHECK(!Cas.HaveChunk(ChunkHashes[i + 1]));
			CHECK(ChunkHashes[i] == IoHash::HashBuffer(Cas.FindChunk(ChunkHashes[i])));
		}
	}
}

TEST_CASE("compactcas.threadedinsert")	// * doctest::skip(true))
{
	//	for (uint32_t i = 0; i < 100; ++i)
	{
		ScopedTemporaryDirectory TempDir;

		CasStoreConfiguration CasConfig;
		CasConfig.RootDirectory = TempDir.Path();

		CreateDirectories(CasConfig.RootDirectory);

		const uint64_t kChunkSize	= 1048;
		const int32_t  kChunkCount	= 4096;
		uint64_t	   ExpectedSize = 0;

		std::unordered_map<IoHash, IoBuffer, IoHash::Hasher> Chunks;
		Chunks.reserve(kChunkCount);

		for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
		{
			while (true)
			{
				IoBuffer Chunk = CreateChunk(kChunkSize);
				IoHash	 Hash  = HashBuffer(Chunk);
				if (Chunks.contains(Hash))
				{
					continue;
				}
				Chunks[Hash] = Chunk;
				ExpectedSize += Chunk.Size();
				break;
			}
		}

		std::atomic<size_t>	 WorkCompleted = 0;
		WorkerThreadPool	 ThreadPool(4);
		CasGc				 Gc;
		CasContainerStrategy Cas(CasConfig, Gc);
		Cas.Initialize("test", 32768, 16, true);
		{
			for (const auto& Chunk : Chunks)
			{
				const IoHash&	Hash   = Chunk.first;
				const IoBuffer& Buffer = Chunk.second;
				ThreadPool.ScheduleWork([&Cas, &WorkCompleted, Buffer, Hash]() {
					CasStore::InsertResult InsertResult = Cas.InsertChunk(Buffer, Hash);
					ZEN_ASSERT(InsertResult.New);
					WorkCompleted.fetch_add(1);
				});
			}
			while (WorkCompleted < Chunks.size())
			{
				Sleep(1);
			}
		}

		WorkCompleted			 = 0;
		const uint64_t TotalSize = Cas.StorageSize().DiskSize;
		CHECK_EQ(ExpectedSize, TotalSize);

		{
			for (const auto& Chunk : Chunks)
			{
				ThreadPool.ScheduleWork([&Cas, &WorkCompleted, &Chunk]() {
					IoHash	 ChunkHash = Chunk.first;
					IoBuffer Buffer	   = Cas.FindChunk(ChunkHash);
					IoHash	 Hash	   = IoHash::HashBuffer(Buffer);
					CHECK(ChunkHash == Hash);
					WorkCompleted.fetch_add(1);
				});
			}
			while (WorkCompleted < Chunks.size())
			{
				Sleep(1);
			}
		}

		std::unordered_set<IoHash, IoHash::Hasher> GcChunkHashes;
		GcChunkHashes.reserve(Chunks.size());
		for (const auto& Chunk : Chunks)
		{
			GcChunkHashes.insert(Chunk.first);
		}
		{
			WorkCompleted = 0;
			std::unordered_map<IoHash, IoBuffer, IoHash::Hasher> NewChunks;
			NewChunks.reserve(kChunkCount);

			for (int32_t Idx = 0; Idx < kChunkCount; ++Idx)
			{
				IoBuffer Chunk	= CreateChunk(kChunkSize);
				IoHash	 Hash	= HashBuffer(Chunk);
				NewChunks[Hash] = Chunk;
			}

			std::atomic_uint32_t AddedChunkCount;

			for (const auto& Chunk : NewChunks)
			{
				ThreadPool.ScheduleWork([&Cas, &WorkCompleted, Chunk, &AddedChunkCount]() {
					Cas.InsertChunk(Chunk.second, Chunk.first);
					AddedChunkCount.fetch_add(1);
					WorkCompleted.fetch_add(1);
				});
			}
			for (const auto& Chunk : Chunks)
			{
				ThreadPool.ScheduleWork([&Cas, &WorkCompleted, Chunk]() {
					IoHash	 ChunkHash = Chunk.first;
					IoBuffer Buffer	   = Cas.FindChunk(ChunkHash);
					if (Buffer)
					{
						CHECK(ChunkHash == IoHash::HashBuffer(Buffer));
					}
					WorkCompleted.fetch_add(1);
				});
			}

			while (AddedChunkCount.load() < NewChunks.size())
			{
				// Need to be careful since we might GC blocks we don't know outside of RwLock::ExclusiveLockScope
				for (const auto& Chunk : NewChunks)
				{
					if (Cas.HaveChunk(Chunk.first))
					{
						GcChunkHashes.emplace(Chunk.first);
					}
				}
				std::vector<IoHash> KeepHashes(GcChunkHashes.begin(), GcChunkHashes.end());
				size_t				C = 0;
				while (C < KeepHashes.size())
				{
					if (C % 155 == 0)
					{
						if (C < KeepHashes.size() - 1)
						{
							KeepHashes[C] = KeepHashes[KeepHashes.size() - 1];
							KeepHashes.pop_back();
						}
						if (C + 3 < KeepHashes.size() - 1)
						{
							KeepHashes[C + 3] = KeepHashes[KeepHashes.size() - 1];
							KeepHashes.pop_back();
						}
					}
					C++;
				}

				GcContext GcCtx;
				GcCtx.CollectSmallObjects(true);
				GcCtx.ContributeCas(KeepHashes);
				Cas.CollectGarbage(GcCtx);
				CasChunkSet& Deleted = GcCtx.DeletedCas();
				Deleted.IterateChunks([&GcChunkHashes](const IoHash& ChunkHash) { GcChunkHashes.erase(ChunkHash); });
			}

			while (WorkCompleted < NewChunks.size() + Chunks.size())
			{
				Sleep(1);
			}

			// Need to be careful since we might GC blocks we don't know outside of RwLock::ExclusiveLockScope
			for (const auto& Chunk : NewChunks)
			{
				if (Cas.HaveChunk(Chunk.first))
				{
					GcChunkHashes.emplace(Chunk.first);
				}
			}
			std::vector<IoHash> KeepHashes(GcChunkHashes.begin(), GcChunkHashes.end());
			size_t				C = 0;
			while (C < KeepHashes.size())
			{
				if (C % 155 == 0)
				{
					if (C < KeepHashes.size() - 1)
					{
						KeepHashes[C] = KeepHashes[KeepHashes.size() - 1];
						KeepHashes.pop_back();
					}
					if (C + 3 < KeepHashes.size() - 1)
					{
						KeepHashes[C + 3] = KeepHashes[KeepHashes.size() - 1];
						KeepHashes.pop_back();
					}
				}
				C++;
			}

			GcContext GcCtx;
			GcCtx.CollectSmallObjects(true);
			GcCtx.ContributeCas(KeepHashes);
			Cas.CollectGarbage(GcCtx);
			CasChunkSet& Deleted = GcCtx.DeletedCas();
			Deleted.IterateChunks([&GcChunkHashes](const IoHash& ChunkHash) { GcChunkHashes.erase(ChunkHash); });
		}
		{
			WorkCompleted = 0;
			for (const IoHash& ChunkHash : GcChunkHashes)
			{
				ThreadPool.ScheduleWork([&Cas, &WorkCompleted, ChunkHash]() {
					CHECK(Cas.HaveChunk(ChunkHash));
					CHECK(ChunkHash == IoHash::HashBuffer(Cas.FindChunk(ChunkHash)));
					WorkCompleted.fetch_add(1);
				});
			}
			while (WorkCompleted < GcChunkHashes.size())
			{
				Sleep(1);
			}
		}
	}
}

TEST_CASE("compactcas.migrate.large.data" * doctest::skip(true))
{
	const char*			  BigDataPath  = "D:\\zen-data\\dc4-zen-cache-t\\cas";
	std::filesystem::path TobsBasePath = GetBasePath(BigDataPath, "tobs");
	std::filesystem::path SobsBasePath = GetBasePath(BigDataPath, "sobs");
	std::filesystem::remove_all(TobsBasePath);
	std::filesystem::remove_all(SobsBasePath);

	CasStoreConfiguration CasConfig;
	CasConfig.RootDirectory = BigDataPath;
	uint64_t TObsSize		= 0;
	{
		CasGc				 TobsCasGc;
		CasContainerStrategy TobsCas(CasConfig, TobsCasGc);
		TobsCas.Initialize("tobs", 1u << 28, 16, false);
		TObsSize = TobsCas.StorageSize().DiskSize;
		CHECK(TObsSize > 0);
	}

	uint64_t SObsSize = 0;
	{
		CasGc				 SobsCasGc;
		CasContainerStrategy SobsCas(CasConfig, SobsCasGc);
		SobsCas.Initialize("sobs", 1u << 30, 4096, false);
		SObsSize = SobsCas.StorageSize().DiskSize;
		CHECK(SObsSize > 0);
	}

	CasGc				 TobsCasGc;
	CasContainerStrategy TobsCas(CasConfig, TobsCasGc);
	TobsCas.Initialize("tobs", 1u << 28, 16, false);
	GcContext TobsGcCtx;
	TobsCas.CollectGarbage(TobsGcCtx);
	CHECK(TobsCas.StorageSize().DiskSize == TObsSize);

	CasGc				 SobsCasGc;
	CasContainerStrategy SobsCas(CasConfig, SobsCasGc);
	SobsCas.Initialize("sobs", 1u << 30, 4096, false);
	GcContext SobsGcCtx;
	SobsCas.CollectGarbage(SobsGcCtx);
	CHECK(SobsCas.StorageSize().DiskSize == SObsSize);
}

#endif

void
compactcas_forcelink()
{
}

}  // namespace zen