merge from main

8 files changed, 726 insertions, 343 deletions

diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index e11499289..79d15a204 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -398,7 +398,7 @@ HttpStructuredCacheService::HandleRequest(HttpServerRequest& Request)
 }
 
 void
-HttpStructuredCacheService::HandleCacheNamespaceRequest(zen::HttpServerRequest& Request, std::string_view)
+HttpStructuredCacheService::HandleCacheNamespaceRequest(zen::HttpServerRequest& Request, std::string_view Namespace)
 {
 	switch (Request.RequestVerb())
 	{
@@ -412,14 +412,14 @@ HttpStructuredCacheService::HandleCacheNamespaceRequest(zen::HttpServerRequest&
 		case HttpVerb::kDelete:
 			// Drop namespace
 			{
-				//				if (m_CacheStore.DropNamespace(Namespace))
-				//				{
-				//					return Request.WriteResponse(HttpResponseCode::OK);
-				//				}
-				//				else
-				//				{
-				//					return Request.WriteResponse(HttpResponseCode::NotFound);
-				//				}
+				if (m_CacheStore.DropNamespace(Namespace))
+				{
+					return Request.WriteResponse(HttpResponseCode::OK);
+				}
+				else
+				{
+					return Request.WriteResponse(HttpResponseCode::NotFound);
+				}
 			}
 			break;
 
diff --git a/zenserver/cache/structuredcachestore.cpp b/zenserver/cache/structuredcachestore.cpp
index da948fd72..ee0835fd3 100644
--- a/zenserver/cache/structuredcachestore.cpp
+++ b/zenserver/cache/structuredcachestore.cpp
@@ -205,6 +205,36 @@ namespace {
 		return true;
 	}
 
+	bool MoveAndDeleteDirectory(const std::filesystem::path& Dir)
+	{
+		int DropIndex = 0;
+		do
+		{
+			if (!std::filesystem::exists(Dir))
+			{
+				return false;
+			}
+
+			std::string			  DroppedName		= fmt::format("[dropped]{}({})", Dir.filename().string(), DropIndex);
+			std::filesystem::path DroppedBucketPath = Dir.parent_path() / DroppedName;
+			if (std::filesystem::exists(DroppedBucketPath))
+			{
+				DropIndex++;
+				continue;
+			}
+
+			std::error_code Ec;
+			std::filesystem::rename(Dir, DroppedBucketPath, Ec);
+			if (!Ec)
+			{
+				DeleteDirectories(DroppedBucketPath);
+				return true;
+			}
+			// TODO: Do we need to bail at some point?
+			zen::Sleep(100);
+		} while (true);
+	}
+
 }  // namespace
 
 namespace fs = std::filesystem;
@@ -342,6 +372,13 @@ ZenCacheNamespace::DropBucket(std::string_view Bucket)
 	return AnyDropped;
 }
 
+bool
+ZenCacheNamespace::Drop()
+{
+	m_MemLayer.Drop();
+	return m_DiskLayer.Drop();
+}
+
 void
 ZenCacheNamespace::Flush()
 {
@@ -404,14 +441,14 @@ ZenCacheMemoryLayer::Get(std::string_view InBucket, const IoHash& HashKey, ZenCa
 {
 	RwLock::SharedLockScope _(m_Lock);
 
-	auto it = m_Buckets.find(std::string(InBucket));
+	auto It = m_Buckets.find(std::string(InBucket));
 
-	if (it == m_Buckets.end())
+	if (It == m_Buckets.end())
 	{
 		return false;
 	}
 
-	CacheBucket* Bucket = &it->second;
+	CacheBucket* Bucket = It->second.get();
 
 	_.ReleaseNow();
 
@@ -425,14 +462,15 @@ ZenCacheMemoryLayer::Get(std::string_view InBucket, const IoHash& HashKey, ZenCa
 void
 ZenCacheMemoryLayer::Put(std::string_view InBucket, const IoHash& HashKey, const ZenCacheValue& Value)
 {
-	CacheBucket* Bucket = nullptr;
+	const auto	 BucketName = std::string(InBucket);
+	CacheBucket* Bucket		= nullptr;
 
 	{
 		RwLock::SharedLockScope _(m_Lock);
 
 		if (auto It = m_Buckets.find(std::string(InBucket)); It != m_Buckets.end())
 		{
-			Bucket = &It->second;
+			Bucket = It->second.get();
 		}
 	}
 
@@ -444,11 +482,12 @@ ZenCacheMemoryLayer::Put(std::string_view InBucket, const IoHash& HashKey, const
 
 		if (auto It = m_Buckets.find(std::string(InBucket)); It != m_Buckets.end())
 		{
-			Bucket = &It->second;
+			Bucket = It->second.get();
 		}
 		else
 		{
-			Bucket = &m_Buckets[std::string(InBucket)];
+			auto InsertResult = m_Buckets.emplace(BucketName, std::make_unique<CacheBucket>());
+			Bucket			  = InsertResult.first->second.get();
 		}
 	}
 
@@ -458,11 +497,37 @@ ZenCacheMemoryLayer::Put(std::string_view InBucket, const IoHash& HashKey, const
 }
 
 bool
-ZenCacheMemoryLayer::DropBucket(std::string_view Bucket)
+ZenCacheMemoryLayer::DropBucket(std::string_view InBucket)
 {
 	RwLock::ExclusiveLockScope _(m_Lock);
 
-	return !!m_Buckets.erase(std::string(Bucket));
+	auto It = m_Buckets.find(std::string(InBucket));
+
+	if (It != m_Buckets.end())
+	{
+		CacheBucket& Bucket = *It->second;
+		m_DroppedBuckets.push_back(std::move(It->second));
+		m_Buckets.erase(It);
+		Bucket.Drop();
+		return true;
+	}
+	return false;
+}
+
+void
+ZenCacheMemoryLayer::Drop()
+{
+	RwLock::ExclusiveLockScope				  _(m_Lock);
+	std::vector<std::unique_ptr<CacheBucket>> Buckets;
+	Buckets.reserve(m_Buckets.size());
+	while (!m_Buckets.empty())
+	{
+		const auto&	 It		= m_Buckets.begin();
+		CacheBucket& Bucket = *It->second;
+		m_DroppedBuckets.push_back(std::move(It->second));
+		m_Buckets.erase(It->first);
+		Bucket.Drop();
+	}
 }
 
 void
@@ -472,7 +537,7 @@ ZenCacheMemoryLayer::Scrub(ScrubContext& Ctx)
 
 	for (auto& Kv : m_Buckets)
 	{
-		Kv.second.Scrub(Ctx);
+		Kv.second->Scrub(Ctx);
 	}
 }
 
@@ -486,7 +551,7 @@ ZenCacheMemoryLayer::GatherAccessTimes(zen::access_tracking::AccessTimes& Access
 	for (auto& Kv : m_Buckets)
 	{
 		std::vector<KeyAccessTime>& Bucket = AccessTimes.Buckets[Kv.first];
-		Kv.second.GatherAccessTimes(Bucket);
+		Kv.second->GatherAccessTimes(Bucket);
 	}
 }
 
@@ -505,7 +570,7 @@ ZenCacheMemoryLayer::TotalSize() const
 
 	for (auto& Kv : m_Buckets)
 	{
-		TotalSize += Kv.second.TotalSize();
+		TotalSize += Kv.second->TotalSize();
 	}
 
 	return TotalSize;
@@ -570,9 +635,16 @@ ZenCacheMemoryLayer::CacheBucket::Put(const IoHash& HashKey, const ZenCacheValue
 	m_TotalSize.fetch_add(Value.Value.GetSize(), std::memory_order::relaxed);
 }
 
+void
+ZenCacheMemoryLayer::CacheBucket::Drop()
+{
+	RwLock::ExclusiveLockScope _(m_BucketLock);
+	m_CacheMap.clear();
+}
+
 //////////////////////////////////////////////////////////////////////////
 
-ZenCacheDiskLayer::CacheBucket::CacheBucket(std::string BucketName) : m_BucketName(std::move(BucketName))
+ZenCacheDiskLayer::CacheBucket::CacheBucket(std::string BucketName) : m_BucketName(std::move(BucketName)), m_BucketId(Oid::Zero)
 {
 }
 
@@ -581,19 +653,6 @@ ZenCacheDiskLayer::CacheBucket::~CacheBucket()
 }
 
 bool
-ZenCacheDiskLayer::CacheBucket::Delete(std::filesystem::path BucketDir)
-{
-	if (std::filesystem::exists(BucketDir))
-	{
-		DeleteDirectories(BucketDir);
-
-		return true;
-	}
-
-	return false;
-}
-
-void
 ZenCacheDiskLayer::CacheBucket::OpenOrCreate(std::filesystem::path BucketDir, bool AllowCreate)
 {
 	using namespace std::literals;
@@ -611,7 +670,10 @@ ZenCacheDiskLayer::CacheBucket::OpenOrCreate(std::filesystem::path BucketDir, bo
 	if (Manifest)
 	{
 		m_BucketId = Manifest["BucketId"].AsObjectId();
-		m_IsOk	   = m_BucketId != Oid::Zero;
+		if (m_BucketId == Oid::Zero)
+		{
+			return false;
+		}
 	}
 	else if (AllowCreate)
 	{
@@ -625,7 +687,7 @@ ZenCacheDiskLayer::CacheBucket::OpenOrCreate(std::filesystem::path BucketDir, bo
 	}
 	else
 	{
-		return;
+		return false;
 	}
 
 	OpenLog(BucketDir, IsNew);
@@ -641,7 +703,7 @@ ZenCacheDiskLayer::CacheBucket::OpenOrCreate(std::filesystem::path BucketDir, bo
 		}
 	}
 
-	m_IsOk = true;
+	return true;
 }
 
 void
@@ -1144,7 +1206,7 @@ ZenCacheDiskLayer::CacheBucket::GetStandaloneCacheValue(const DiskLocation& Loc,
 
 	RwLock::SharedLockScope ValueLock(LockForHash(HashKey));
 
-	if (IoBuffer Data = IoBufferBuilder::MakeFromFile(DataFilePath.ToPath()))
+	if (IoBuffer Data = IoBufferBuilder::MakeFromFileWithSharedDelete(DataFilePath.ToPath()))
 	{
 		OutValue.Value = Data;
 		OutValue.Value.SetContentType(Loc.GetContentType());
@@ -1158,11 +1220,6 @@ ZenCacheDiskLayer::CacheBucket::GetStandaloneCacheValue(const DiskLocation& Loc,
 bool
 ZenCacheDiskLayer::CacheBucket::Get(const IoHash& HashKey, ZenCacheValue& OutValue)
 {
-	if (!m_IsOk)
-	{
-		return false;
-	}
-
 	RwLock::SharedLockScope _(m_IndexLock);
 	auto					It = m_Index.find(HashKey);
 	if (It == m_Index.end())
@@ -1184,11 +1241,6 @@ ZenCacheDiskLayer::CacheBucket::Get(const IoHash& HashKey, ZenCacheValue& OutVal
 void
 ZenCacheDiskLayer::CacheBucket::Put(const IoHash& HashKey, const ZenCacheValue& Value)
 {
-	if (!m_IsOk)
-	{
-		return;
-	}
-
 	if (Value.Value.Size() >= m_LargeObjectThreshold)
 	{
 		return PutStandaloneCacheValue(HashKey, Value);
@@ -1196,12 +1248,24 @@ ZenCacheDiskLayer::CacheBucket::Put(const IoHash& HashKey, const ZenCacheValue&
 	PutInlineCacheValue(HashKey, Value);
 }
 
-void
+bool
 ZenCacheDiskLayer::CacheBucket::Drop()
 {
+	RwLock::ExclusiveLockScope _(m_IndexLock);
+
+	std::vector<std::unique_ptr<RwLock::ExclusiveLockScope>> ShardLocks;
+	ShardLocks.reserve(256);
+	for (RwLock& Lock : m_ShardedLocks)
+	{
+		ShardLocks.push_back(std::make_unique<RwLock::ExclusiveLockScope>(Lock));
+	}
 	m_BlockStore.Close();
 	m_SlogFile.Close();
-	DeleteDirectories(m_BucketDir);
+
+	bool Deleted = MoveAndDeleteDirectory(m_BucketDir);
+
+	m_Index.clear();
+	return Deleted;
 }
 
 void
@@ -1703,7 +1767,8 @@ ZenCacheDiskLayer::CollectGarbage(GcContext& GcCtx)
 
 	for (auto& Kv : m_Buckets)
 	{
-		Kv.second.CollectGarbage(GcCtx);
+		CacheBucket& Bucket = *Kv.second;
+		Bucket.CollectGarbage(GcCtx);
 	}
 }
 
@@ -1716,7 +1781,7 @@ ZenCacheDiskLayer::UpdateAccessTimes(const zen::access_tracking::AccessTimes& Ac
 	{
 		if (auto It = m_Buckets.find(Kv.first); It != m_Buckets.end())
 		{
-			CacheBucket& Bucket = It->second;
+			CacheBucket& Bucket = *It->second;
 			Bucket.UpdateAccessTimes(Kv.second);
 		}
 	}
@@ -1766,110 +1831,80 @@ ZenCacheDiskLayer::CacheBucket::PutStandaloneCacheValue(const IoHash& HashKey, c
 	BuildPath(DataFilePath, HashKey);
 	std::filesystem::path FsPath{DataFilePath.ToPath()};
 
-	// We retry to move the file since it can be held open for read.
-	// This happens if the server processes a Get request for the file or
-	// if we are busy sending the file upstream
-	int RetryCount = 4;
-	do
-	{
-		Ec.clear();
-		{
-			RwLock::ExclusiveLockScope ValueLock(LockForHash(HashKey));
-
-			DataFile.MoveTemporaryIntoPlace(FsPath, Ec);
+	RwLock::ExclusiveLockScope ValueLock(LockForHash(HashKey));
 
-			// Once we have called MoveTemporaryIntoPlace automatic clean up the temp file
-			// will be disabled as the file handle has already been closed
-			CleanUpTempFile = Ec ? true : false;
-
-			if (Ec)
-			{
-				std::error_code ExistingEc;
-				uint64_t		OldFileSize = std::filesystem::file_size(FsPath, ExistingEc);
-				if (!ExistingEc && (OldFileSize == NewFileSize))
-				{
-					ZEN_INFO(
-						"Failed to move temporary file '{}' to '{}' for '{}'. Target file has same size, assuming concurrent write of same "
-						"value, "
-						"move "
-						"failed with reason '{}'",
-						DataFile.GetPath(),
-						FsPath.string(),
-						m_BucketDir,
-						Ec.message());
-					return;
-				}
-			}
-		}
+	// 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);
+	if (Ec && Ec.value() != ENOENT)
+	{
+		throw std::system_error(Ec, fmt::format("Failed to replace file '{}' for put in '{}'", DataFilePath.ToUtf8(), m_BucketDir));
+	}
 
-		if (!Ec)
+	DataFile.MoveTemporaryIntoPlace(FsPath, Ec);
+	if (Ec)
+	{
+		std::filesystem::path ParentPath = FsPath.parent_path();
+		if (std::filesystem::is_directory(ParentPath))
 		{
-			uint8_t EntryFlags = DiskLocation::kStandaloneFile;
-
-			if (Value.Value.GetContentType() == ZenContentType::kCbObject)
-			{
-				EntryFlags |= DiskLocation::kStructured;
-			}
-			else if (Value.Value.GetContentType() == ZenContentType::kCompressedBinary)
-			{
-				EntryFlags |= DiskLocation::kCompressed;
-			}
-
-			DiskLocation Loc(NewFileSize, EntryFlags);
-			IndexEntry	 Entry = IndexEntry(Loc, GcClock::TickCount());
-
-			uint64_t				   OldFileSize = 0;
-			RwLock::ExclusiveLockScope _(m_IndexLock);
-			if (auto It = m_Index.find(HashKey); It == m_Index.end())
-			{
-				// Previously unknown object
-				m_Index.insert({HashKey, Entry});
-			}
-			else
-			{
-				// TODO: should check if write is idempotent and bail out if it is?
-				OldFileSize = It.value().Location.Size();
-				It.value()	= Entry;
-			}
-
-			m_SlogFile.Append({.Key = HashKey, .Location = Loc});
-			if (OldFileSize <= NewFileSize)
-			{
-				m_TotalSize.fetch_add(NewFileSize - OldFileSize, std::memory_order::relaxed);
-			}
-			else
-			{
-				m_TotalSize.fetch_sub(OldFileSize - NewFileSize, std::memory_order::relaxed);
-			}
-			return;
+			throw std::system_error(Ec, fmt::format("Failed to finalize file '{}' for put in '{}'", DataFilePath.ToUtf8(), m_BucketDir));
 		}
-
-		std::filesystem::path ParentPath = FsPath.parent_path();
-		if (!std::filesystem::is_directory(ParentPath))
+		Ec.clear();
+		std::filesystem::create_directories(ParentPath, Ec);
+		if (Ec)
 		{
-			Ec.clear();
-			std::filesystem::create_directories(ParentPath, Ec);
-			if (!Ec)
-			{
-				// Retry without sleep
-				continue;
-			}
 			throw std::system_error(
 				Ec,
 				fmt::format("Failed to create parent directory '{}' for file '{}' for put in '{}'", ParentPath, FsPath, m_BucketDir));
 		}
 
-		ZEN_INFO("Failed renaming temporary file '{}' to '{}' for put in '{}', pausing and retrying, reason '{}'",
-				 DataFile.GetPath().string(),
-				 FsPath.string(),
-				 m_BucketDir,
-				 Ec.message());
+		DataFile.MoveTemporaryIntoPlace(FsPath, Ec);
+		if (Ec)
+		{
+			throw std::system_error(Ec, fmt::format("Failed to finalize file '{}' for put in '{}'", DataFilePath.ToUtf8(), m_BucketDir));
+		}
+	}
+
+	// Once we have called MoveTemporaryIntoPlace automatic clean up the temp file
+	// will be disabled as the file handle has already been closed
+	CleanUpTempFile = false;
 
-		// Semi arbitrary back-off
-		zen::Sleep(200 * (5 - RetryCount));	 // Sleep at most for a total of 3 seconds
-	} while (RetryCount-- > 0);
+	uint8_t EntryFlags = DiskLocation::kStandaloneFile;
 
-	throw std::system_error(Ec, fmt::format("Failed to finalize file '{}' for put in '{}'", DataFilePath.ToUtf8(), m_BucketDir));
+	if (Value.Value.GetContentType() == ZenContentType::kCbObject)
+	{
+		EntryFlags |= DiskLocation::kStructured;
+	}
+	else if (Value.Value.GetContentType() == ZenContentType::kCompressedBinary)
+	{
+		EntryFlags |= DiskLocation::kCompressed;
+	}
+
+	DiskLocation Loc(NewFileSize, EntryFlags);
+	IndexEntry	 Entry = IndexEntry(Loc, GcClock::TickCount());
+
+	uint64_t				   OldFileSize = 0;
+	RwLock::ExclusiveLockScope _(m_IndexLock);
+	if (auto It = m_Index.find(HashKey); It == m_Index.end())
+	{
+		// Previously unknown object
+		m_Index.insert({HashKey, Entry});
+	}
+	else
+	{
+		// TODO: should check if write is idempotent and bail out if it is?
+		OldFileSize = It.value().Location.Size();
+		It.value()	= Entry;
+	}
+
+	m_SlogFile.Append({.Key = HashKey, .Location = Loc});
+	if (OldFileSize <= NewFileSize)
+	{
+		m_TotalSize.fetch_add(NewFileSize - OldFileSize, std::memory_order::relaxed);
+	}
+	else
+	{
+		m_TotalSize.fetch_sub(OldFileSize - NewFileSize, std::memory_order::relaxed);
+	}
 }
 
 void
@@ -1925,11 +1960,11 @@ ZenCacheDiskLayer::Get(std::string_view InBucket, const IoHash& HashKey, ZenCach
 	{
 		RwLock::SharedLockScope _(m_Lock);
 
-		auto it = m_Buckets.find(BucketName);
+		auto It = m_Buckets.find(BucketName);
 
-		if (it != m_Buckets.end())
+		if (It != m_Buckets.end())
 		{
-			Bucket = &it->second;
+			Bucket = It->second.get();
 		}
 	}
 
@@ -1939,24 +1974,27 @@ ZenCacheDiskLayer::Get(std::string_view InBucket, const IoHash& HashKey, ZenCach
 
 		RwLock::ExclusiveLockScope _(m_Lock);
 
-		if (auto it = m_Buckets.find(BucketName); it != m_Buckets.end())
+		if (auto It = m_Buckets.find(BucketName); It != m_Buckets.end())
 		{
-			Bucket = &it->second;
+			Bucket = It->second.get();
 		}
 		else
 		{
-			auto It = m_Buckets.try_emplace(BucketName, BucketName);
-			Bucket	= &It.first->second;
+			auto InsertResult = m_Buckets.emplace(BucketName, std::make_unique<CacheBucket>(BucketName));
+			Bucket			  = InsertResult.first->second.get();
 
 			std::filesystem::path BucketPath = m_RootDir;
 			BucketPath /= BucketName;
 
-			Bucket->OpenOrCreate(BucketPath);
+			if (!Bucket->OpenOrCreate(BucketPath))
+			{
+				m_Buckets.erase(BucketName);
+				return false;
+			}
 		}
 	}
 
 	ZEN_ASSERT(Bucket != nullptr);
-
 	return Bucket->Get(HashKey, OutValue);
 }
 
@@ -1969,11 +2007,11 @@ ZenCacheDiskLayer::Put(std::string_view InBucket, const IoHash& HashKey, const Z
 	{
 		RwLock::SharedLockScope _(m_Lock);
 
-		auto it = m_Buckets.find(BucketName);
+		auto It = m_Buckets.find(BucketName);
 
-		if (it != m_Buckets.end())
+		if (It != m_Buckets.end())
 		{
-			Bucket = &it->second;
+			Bucket = It->second.get();
 		}
 	}
 
@@ -1983,28 +2021,29 @@ ZenCacheDiskLayer::Put(std::string_view InBucket, const IoHash& HashKey, const Z
 
 		RwLock::ExclusiveLockScope _(m_Lock);
 
-		if (auto it = m_Buckets.find(BucketName); it != m_Buckets.end())
+		if (auto It = m_Buckets.find(BucketName); It != m_Buckets.end())
 		{
-			Bucket = &it->second;
+			Bucket = It->second.get();
 		}
 		else
 		{
-			auto It = m_Buckets.try_emplace(BucketName, BucketName);
-			Bucket	= &It.first->second;
+			auto InsertResult = m_Buckets.emplace(BucketName, std::make_unique<CacheBucket>(BucketName));
+			Bucket			  = InsertResult.first->second.get();
 
 			std::filesystem::path BucketPath = m_RootDir;
 			BucketPath /= BucketName;
 
-			Bucket->OpenOrCreate(BucketPath);
+			if (!Bucket->OpenOrCreate(BucketPath))
+			{
+				m_Buckets.erase(BucketName);
+				return;
+			}
 		}
 	}
 
 	ZEN_ASSERT(Bucket != nullptr);
 
-	if (Bucket->IsOk())
-	{
-		Bucket->Put(HashKey, Value);
-	}
+	Bucket->Put(HashKey, Value);
 }
 
 void
@@ -2023,26 +2062,20 @@ ZenCacheDiskLayer::DiscoverBuckets()
 		// New bucket needs to be created
 		if (auto It = m_Buckets.find(BucketName); It != m_Buckets.end())
 		{
+			continue;
 		}
-		else
-		{
-			auto InsertResult = m_Buckets.try_emplace(BucketName, BucketName);
-
-			CacheBucket& Bucket = InsertResult.first->second;
 
-			Bucket.OpenOrCreate(BucketPath, /* AllowCreate */ false);
+		auto		 InsertResult = m_Buckets.emplace(BucketName, std::make_unique<CacheBucket>(BucketName));
+		CacheBucket& Bucket		  = *InsertResult.first->second;
 
-			if (Bucket.IsOk())
-			{
-				ZEN_INFO("Discovered bucket '{}'", BucketName);
-			}
-			else
-			{
-				ZEN_WARN("Found directory '{}' in our base directory '{}' but it is not a valid bucket", BucketName, m_RootDir);
+		if (!Bucket.OpenOrCreate(BucketPath, /* AllowCreate */ false))
+		{
+			ZEN_WARN("Found directory '{}' in our base directory '{}' but it is not a valid bucket", BucketName, m_RootDir);
 
-				m_Buckets.erase(InsertResult.first);
-			}
+			m_Buckets.erase(InsertResult.first);
+			continue;
 		}
+		ZEN_INFO("Discovered bucket '{}'", BucketName);
 	}
 }
 
@@ -2051,23 +2084,42 @@ ZenCacheDiskLayer::DropBucket(std::string_view InBucket)
 {
 	RwLock::ExclusiveLockScope _(m_Lock);
 
-	auto it = m_Buckets.find(std::string(InBucket));
+	auto It = m_Buckets.find(std::string(InBucket));
 
-	if (it != m_Buckets.end())
+	if (It != m_Buckets.end())
 	{
-		CacheBucket* Bucket = &it->second;
-
-		Bucket->Drop();
-
-		m_Buckets.erase(it);
+		CacheBucket& Bucket = *It->second;
+		m_DroppedBuckets.push_back(std::move(It->second));
+		m_Buckets.erase(It);
 
-		return true;
+		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 MoveAndDeleteDirectory(BucketPath);
+}
 
-	return CacheBucket::Delete(BucketPath);
+bool
+ZenCacheDiskLayer::Drop()
+{
+	RwLock::ExclusiveLockScope _(m_Lock);
+
+	std::vector<std::unique_ptr<CacheBucket>> Buckets;
+	Buckets.reserve(m_Buckets.size());
+	while (!m_Buckets.empty())
+	{
+		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())
+		{
+			return false;
+		}
+	}
+	return MoveAndDeleteDirectory(m_RootDir);
 }
 
 void
@@ -2080,7 +2132,8 @@ ZenCacheDiskLayer::Flush()
 		Buckets.reserve(m_Buckets.size());
 		for (auto& Kv : m_Buckets)
 		{
-			Buckets.push_back(&Kv.second);
+			CacheBucket* Bucket = Kv.second.get();
+			Buckets.push_back(Bucket);
 		}
 	}
 
@@ -2097,7 +2150,8 @@ ZenCacheDiskLayer::Scrub(ScrubContext& Ctx)
 
 	for (auto& Kv : m_Buckets)
 	{
-		Kv.second.Scrub(Ctx);
+		CacheBucket& Bucket = *Kv.second;
+		Bucket.Scrub(Ctx);
 	}
 }
 
@@ -2108,7 +2162,8 @@ ZenCacheDiskLayer::GatherReferences(GcContext& GcCtx)
 
 	for (auto& Kv : m_Buckets)
 	{
-		Kv.second.GatherReferences(GcCtx);
+		CacheBucket& Bucket = *Kv.second;
+		Bucket.GatherReferences(GcCtx);
 	}
 }
 
@@ -2120,7 +2175,7 @@ ZenCacheDiskLayer::TotalSize() const
 
 	for (auto& Kv : m_Buckets)
 	{
-		TotalSize += Kv.second.TotalSize();
+		TotalSize += Kv.second->TotalSize();
 	}
 
 	return TotalSize;
@@ -2130,18 +2185,22 @@ ZenCacheDiskLayer::TotalSize() const
 
 static constexpr std::string_view UE4DDCNamespaceName = "ue4.ddc";
 
-ZenCacheStore::ZenCacheStore(CasGc& Gc, std::filesystem::path BasePath) : GcStorage(Gc), GcContributor(Gc)
+ZenCacheStore::ZenCacheStore(CasGc& Gc, const Configuration& Configuration)
+: GcStorage(Gc)
+, GcContributor(Gc)
+, m_Gc(Gc)
+, m_Configuration(Configuration)
 {
-	CreateDirectories(BasePath);
+	CreateDirectories(m_Configuration.BasePath);
 
 	DirectoryContent DirContent;
-	GetDirectoryContent(BasePath, DirectoryContent::IncludeDirsFlag, DirContent);
+	GetDirectoryContent(m_Configuration.BasePath, DirectoryContent::IncludeDirsFlag, DirContent);
 
 	std::vector<std::string> LegacyBuckets;
 	std::vector<std::string> Namespaces;
 	for (const std::filesystem::path& DirPath : DirContent.Directories)
 	{
-		std::string DirName = PathToUtf8(DirPath.stem());
+		std::string DirName = PathToUtf8(DirPath.filename());
 		if (DirName.starts_with(NamespaceDiskPrefix))
 		{
 			Namespaces.push_back(DirName.substr(NamespaceDiskPrefix.length()));
@@ -2150,7 +2209,7 @@ ZenCacheStore::ZenCacheStore(CasGc& Gc, std::filesystem::path BasePath) : GcStor
 		LegacyBuckets.push_back(DirName);
 	}
 
-	ZEN_INFO("Found #{} namespaces in '{}' and #{} legacy buckets", Namespaces.size(), BasePath, LegacyBuckets.size());
+	ZEN_INFO("Found #{} namespaces in '{}' and #{} legacy buckets", Namespaces.size(), m_Configuration.BasePath, LegacyBuckets.size());
 
 	if (std::find(Namespaces.begin(), Namespaces.end(), UE4DDCNamespaceName) == Namespaces.end())
 	{
@@ -2158,13 +2217,14 @@ ZenCacheStore::ZenCacheStore(CasGc& Gc, std::filesystem::path BasePath) : GcStor
 
 		ZEN_INFO("Moving #{} legacy buckets to '{}' namespace", LegacyBuckets.size(), UE4DDCNamespaceName);
 
-		std::filesystem::path DefaultNamespaceFolder = BasePath / fmt::format("{}{}", NamespaceDiskPrefix, UE4DDCNamespaceName);
+		std::filesystem::path DefaultNamespaceFolder =
+			m_Configuration.BasePath / fmt::format("{}{}", NamespaceDiskPrefix, UE4DDCNamespaceName);
 		CreateDirectories(DefaultNamespaceFolder);
 
 		// Move any non-namespace folders into the default namespace folder
 		for (const std::string& DirName : LegacyBuckets)
 		{
-			std::filesystem::path LegacyFolder = BasePath / DirName;
+			std::filesystem::path LegacyFolder = m_Configuration.BasePath / DirName;
 			std::filesystem::path NewPath	   = DefaultNamespaceFolder / DirName;
 			std::error_code		  Ec;
 			std::filesystem::rename(LegacyFolder, NewPath, Ec);
@@ -2179,7 +2239,7 @@ ZenCacheStore::ZenCacheStore(CasGc& Gc, std::filesystem::path BasePath) : GcStor
 	for (const std::string& NamespaceName : Namespaces)
 	{
 		m_Namespaces[NamespaceName] =
-			std::make_unique<ZenCacheNamespace>(Gc, BasePath / fmt::format("{}{}", NamespaceDiskPrefix, NamespaceName));
+			std::make_unique<ZenCacheNamespace>(Gc, m_Configuration.BasePath / fmt::format("{}{}", NamespaceDiskPrefix, NamespaceName));
 	}
 }
 
@@ -2216,7 +2276,22 @@ ZenCacheStore::DropBucket(std::string_view Namespace, std::string_view Bucket)
 	{
 		return Store->DropBucket(Bucket);
 	}
-	ZEN_WARN("request for unknown namespace '{}' in ZenCacheStore::Put, bucket '{}'", Namespace, Bucket);
+	ZEN_WARN("request for unknown namespace '{}' in ZenCacheStore::DropBucket, bucket '{}'", Namespace, Bucket);
+	return false;
+}
+
+bool
+ZenCacheStore::DropNamespace(std::string_view InNamespace)
+{
+	RwLock::SharedLockScope _(m_NamespacesLock);
+	if (auto It = m_Namespaces.find(std::string(InNamespace)); It != m_Namespaces.end())
+	{
+		ZenCacheNamespace& Namespace = *It->second;
+		m_DroppedNamespaces.push_back(std::move(It->second));
+		m_Namespaces.erase(It);
+		return Namespace.Drop();
+	}
+	ZEN_WARN("request for unknown namespace '{}' in ZenCacheStore::DropNamespace", InNamespace);
 	return false;
 }
 
@@ -2247,13 +2322,29 @@ ZenCacheStore::GetNamespace(std::string_view Namespace)
 			return It->second.get();
 		}
 	}
-	return nullptr;
+	_.ReleaseNow();
+
+	if (!m_Configuration.AllowAutomaticCreationOfNamespaces)
+	{
+		return nullptr;
+	}
+
+	RwLock::ExclusiveLockScope __(m_NamespacesLock);
+	if (auto It = m_Namespaces.find(std::string(Namespace)); It != m_Namespaces.end())
+	{
+		return It->second.get();
+	}
+
+	auto NewNamespace = m_Namespaces.insert_or_assign(
+		std::string(Namespace),
+		std::make_unique<ZenCacheNamespace>(m_Gc, m_Configuration.BasePath / fmt::format("{}{}", NamespaceDiskPrefix, Namespace)));
+	return NewNamespace.first->second.get();
 }
 
 void
 ZenCacheStore::IterateNamespaces(const std::function<void(std::string_view Namespace, ZenCacheNamespace& Store)>& Callback) const
 {
-	std::vector<std::pair<std::string, ZenCacheNamespace&> > Namespaces;
+	std::vector<std::pair<std::string, ZenCacheNamespace&>> Namespaces;
 	{
 		RwLock::SharedLockScope _(m_NamespacesLock);
 		Namespaces.reserve(m_Namespaces.size());
@@ -3048,40 +3139,210 @@ TEST_CASE("z$.namespaces")
 	ScopedTemporaryDirectory TempDir;
 	CreateDirectories(TempDir.Path());
 
+	IoHash Key1;
+	IoHash Key2;
 	{
 		CasGc		  Gc;
-		ZenCacheStore Zcs(Gc, TempDir.Path() / "cache");
+		ZenCacheStore Zcs(Gc, {.BasePath = TempDir.Path() / "cache", .AllowAutomaticCreationOfNamespaces = false});
 		const auto	  Bucket		  = "teardrinker"sv;
 		const auto	  CustomNamespace = "mynamespace"sv;
 
 		// Create a cache record
-		const IoHash Key		= CreateKey(42);
-		CbObject	 CacheValue = CreateCacheValue(4096);
+		Key1				= CreateKey(42);
+		CbObject CacheValue = CreateCacheValue(4096);
 
 		IoBuffer Buffer = CacheValue.GetBuffer().AsIoBuffer();
 		Buffer.SetContentType(ZenContentType::kCbObject);
 
 		ZenCacheValue PutValue = {.Value = Buffer};
-		Zcs.Put(ZenCacheStore::DefaultNamespace, Bucket, Key, PutValue);
+		Zcs.Put(ZenCacheStore::DefaultNamespace, Bucket, Key1, PutValue);
 
 		ZenCacheValue GetValue;
-		CHECK(Zcs.Get(ZenCacheStore::DefaultNamespace, Bucket, Key, GetValue));
+		CHECK(Zcs.Get(ZenCacheStore::DefaultNamespace, Bucket, Key1, GetValue));
+		CHECK(!Zcs.Get(CustomNamespace, Bucket, Key1, GetValue));
 
-		CHECK(!Zcs.Get(CustomNamespace, Bucket, Key, GetValue));
+		// This should just be dropped as we don't allow creating of namespaces on the fly
+		Zcs.Put(CustomNamespace, Bucket, Key1, PutValue);
+		CHECK(!Zcs.Get(CustomNamespace, Bucket, Key1, GetValue));
+	}
 
-		// This should just be dropped for now until we decide how we add namespaces
-		Zcs.Put(CustomNamespace, Bucket, Key, PutValue);
-		CHECK(!Zcs.Get(CustomNamespace, Bucket, Key, GetValue));
+	{
+		CasGc		  Gc;
+		ZenCacheStore Zcs(Gc, {.BasePath = TempDir.Path() / "cache", .AllowAutomaticCreationOfNamespaces = true});
+		const auto	  Bucket		  = "teardrinker"sv;
+		const auto	  CustomNamespace = "mynamespace"sv;
 
-		const IoHash Key2		 = CreateKey(43);
-		CbObject	 CacheValue2 = CreateCacheValue(4096);
+		Key2				 = CreateKey(43);
+		CbObject CacheValue2 = CreateCacheValue(4096);
 
 		IoBuffer Buffer2 = CacheValue2.GetBuffer().AsIoBuffer();
 		Buffer2.SetContentType(ZenContentType::kCbObject);
 		ZenCacheValue PutValue2 = {.Value = Buffer2};
 		Zcs.Put(CustomNamespace, Bucket, Key2, PutValue2);
 
+		ZenCacheValue GetValue;
 		CHECK(!Zcs.Get(ZenCacheStore::DefaultNamespace, Bucket, Key2, GetValue));
+		CHECK(Zcs.Get(ZenCacheStore::DefaultNamespace, Bucket, Key1, GetValue));
+		CHECK(!Zcs.Get(CustomNamespace, Bucket, Key1, GetValue));
+		CHECK(Zcs.Get(CustomNamespace, Bucket, Key2, GetValue));
+	}
+}
+
+TEST_CASE("z$.drop.bucket")
+{
+	using namespace testutils;
+
+	const auto CreateCacheValue = [](size_t Size) -> CbObject {
+		std::vector<uint8_t> Buf;
+		Buf.resize(Size);
+
+		CbObjectWriter Writer;
+		Writer.AddBinary("Binary"sv, Buf.data(), Buf.size());
+		return Writer.Save();
+	};
+
+	ScopedTemporaryDirectory TempDir;
+	CreateDirectories(TempDir.Path());
+
+	IoHash Key1;
+	IoHash Key2;
+
+	auto PutValue =
+		[&CreateCacheValue](ZenCacheStore& Zcs, std::string_view Namespace, std::string_view Bucket, size_t KeyIndex, size_t Size) {
+			// Create a cache record
+			IoHash	 Key		= CreateKey(KeyIndex);
+			CbObject CacheValue = CreateCacheValue(Size);
+
+			IoBuffer Buffer = CacheValue.GetBuffer().AsIoBuffer();
+			Buffer.SetContentType(ZenContentType::kCbObject);
+
+			ZenCacheValue PutValue = {.Value = Buffer};
+			Zcs.Put(Namespace, Bucket, Key, PutValue);
+			return Key;
+		};
+	auto GetValue = [](ZenCacheStore& Zcs, std::string_view Namespace, std::string_view Bucket, const IoHash& Key) {
+		ZenCacheValue GetValue;
+		Zcs.Get(Namespace, Bucket, Key, GetValue);
+		return GetValue;
+	};
+	WorkerThreadPool Workers(1);
+	{
+		CasGc		  Gc;
+		ZenCacheStore Zcs(Gc, {.BasePath = TempDir.Path() / "cache", .AllowAutomaticCreationOfNamespaces = true});
+		const auto	  Bucket	= "teardrinker"sv;
+		const auto	  Namespace = "mynamespace"sv;
+
+		Key1 = PutValue(Zcs, Namespace, Bucket, 42, 4096);
+		Key2 = PutValue(Zcs, Namespace, Bucket, 43, 2048);
+
+		ZenCacheValue Value1 = GetValue(Zcs, Namespace, Bucket, Key1);
+		CHECK(Value1.Value);
+
+		std::atomic_bool WorkComplete = false;
+		Workers.ScheduleWork([&]() {
+			zen::Sleep(100);
+			Value1.Value = IoBuffer{};
+			WorkComplete = true;
+		});
+		// On Windows, DropBucket() will be blocked as long as we hold a reference to a buffer in the bucket
+		// Our DropBucket execution blocks any incoming request from completing until we are done with the drop
+		CHECK(Zcs.DropBucket(Namespace, Bucket));
+		while (!WorkComplete)
+		{
+			zen::Sleep(1);
+		}
+
+		// Entire bucket should be dropped, but doing a request should will re-create the namespace but it must still be empty
+		Value1 = GetValue(Zcs, Namespace, Bucket, Key1);
+		CHECK(!Value1.Value);
+		ZenCacheValue Value2 = GetValue(Zcs, Namespace, Bucket, Key2);
+		CHECK(!Value2.Value);
+	}
+}
+
+TEST_CASE("z$.drop.namespace")
+{
+	using namespace testutils;
+
+	const auto CreateCacheValue = [](size_t Size) -> CbObject {
+		std::vector<uint8_t> Buf;
+		Buf.resize(Size);
+
+		CbObjectWriter Writer;
+		Writer.AddBinary("Binary"sv, Buf.data(), Buf.size());
+		return Writer.Save();
+	};
+
+	ScopedTemporaryDirectory TempDir;
+	CreateDirectories(TempDir.Path());
+
+	auto PutValue =
+		[&CreateCacheValue](ZenCacheStore& Zcs, std::string_view Namespace, std::string_view Bucket, size_t KeyIndex, size_t Size) {
+			// Create a cache record
+			IoHash	 Key		= CreateKey(KeyIndex);
+			CbObject CacheValue = CreateCacheValue(Size);
+
+			IoBuffer Buffer = CacheValue.GetBuffer().AsIoBuffer();
+			Buffer.SetContentType(ZenContentType::kCbObject);
+
+			ZenCacheValue PutValue = {.Value = Buffer};
+			Zcs.Put(Namespace, Bucket, Key, PutValue);
+			return Key;
+		};
+	auto GetValue = [](ZenCacheStore& Zcs, std::string_view Namespace, std::string_view Bucket, const IoHash& Key) {
+		ZenCacheValue GetValue;
+		Zcs.Get(Namespace, Bucket, Key, GetValue);
+		return GetValue;
+	};
+	WorkerThreadPool Workers(1);
+	{
+		CasGc		  Gc;
+		ZenCacheStore Zcs(Gc, {.BasePath = TempDir.Path() / "cache", .AllowAutomaticCreationOfNamespaces = true});
+		const auto	  Bucket1	 = "teardrinker1"sv;
+		const auto	  Bucket2	 = "teardrinker2"sv;
+		const auto	  Namespace1 = "mynamespace1"sv;
+		const auto	  Namespace2 = "mynamespace2"sv;
+
+		IoHash Key1 = PutValue(Zcs, Namespace1, Bucket1, 42, 4096);
+		IoHash Key2 = PutValue(Zcs, Namespace1, Bucket2, 43, 2048);
+		IoHash Key3 = PutValue(Zcs, Namespace2, Bucket1, 44, 4096);
+		IoHash Key4 = PutValue(Zcs, Namespace2, Bucket2, 45, 2048);
+
+		ZenCacheValue Value1 = GetValue(Zcs, Namespace1, Bucket1, Key1);
+		CHECK(Value1.Value);
+		ZenCacheValue Value2 = GetValue(Zcs, Namespace1, Bucket2, Key2);
+		CHECK(Value2.Value);
+		ZenCacheValue Value3 = GetValue(Zcs, Namespace2, Bucket1, Key3);
+		CHECK(Value3.Value);
+		ZenCacheValue Value4 = GetValue(Zcs, Namespace2, Bucket2, Key4);
+		CHECK(Value4.Value);
+
+		std::atomic_bool WorkComplete = false;
+		Workers.ScheduleWork([&]() {
+			zen::Sleep(100);
+			Value1.Value = IoBuffer{};
+			Value2.Value = IoBuffer{};
+			Value3.Value = IoBuffer{};
+			Value4.Value = IoBuffer{};
+			WorkComplete = true;
+		});
+		// On Windows, DropBucket() will be blocked as long as we hold a reference to a buffer in the bucket
+		// Our DropBucket execution blocks any incoming request from completing until we are done with the drop
+		CHECK(Zcs.DropNamespace(Namespace1));
+		while (!WorkComplete)
+		{
+			zen::Sleep(1);
+		}
+
+		// Entire namespace should be dropped, but doing a request should will re-create the namespace but it must still be empty
+		Value1 = GetValue(Zcs, Namespace1, Bucket1, Key1);
+		CHECK(!Value1.Value);
+		Value2 = GetValue(Zcs, Namespace1, Bucket2, Key2);
+		CHECK(!Value2.Value);
+		Value3 = GetValue(Zcs, Namespace2, Bucket1, Key3);
+		CHECK(Value3.Value);
+		Value4 = GetValue(Zcs, Namespace2, Bucket2, Key4);
+		CHECK(Value4.Value);
 	}
 }
 
@@ -3093,7 +3354,7 @@ TEST_CASE("z$.blocked.disklayer.put")
 
 	const auto CreateCacheValue = [](size_t Size) -> CbObject {
 		std::vector<uint8_t> Buf;
-		Buf.resize(Size);
+		Buf.resize(Size, Size & 0xff);
 
 		CbObjectWriter Writer;
 		Writer.AddBinary("Binary"sv, Buf.data(), Buf.size());
@@ -3115,25 +3376,26 @@ TEST_CASE("z$.blocked.disklayer.put")
 	ZenCacheValue BufferGet;
 	CHECK(Zcs.Get("test_bucket", HashKey, BufferGet));
 
-	// Overwriting with a value of same size should go fine
-	Zcs.Put("test_bucket", HashKey, {.Value = Buffer});
-
 	CbObject CacheValue2 = CreateCacheValue(64 * 1024 + 64 + 1);
 	IoBuffer Buffer2	 = CacheValue2.GetBuffer().AsIoBuffer();
 	Buffer2.SetContentType(ZenContentType::kCbObject);
-#	if ZEN_PLATFORM_WINDOWS
-	// On Windows platform, overwriting with different size while we have
-	// it open for read should throw exception if file is held open
-	CHECK_THROWS(Zcs.Put("test_bucket", HashKey, {.Value = Buffer2}));
-#	else
-	// Other platforms should handle overwrite just fine
+
+	// We should be able to overwrite even if the file is open for read
 	Zcs.Put("test_bucket", HashKey, {.Value = Buffer2});
-#	endif
 
-	BufferGet = ZenCacheValue{};
+	MemoryView OldView = BufferGet.Value.GetView();
 
-	// Read access has been removed, we should now be able to overwrite it
-	Zcs.Put("test_bucket", HashKey, {.Value = Buffer2});
+	ZenCacheValue BufferGet2;
+	CHECK(Zcs.Get("test_bucket", HashKey, BufferGet2));
+	MemoryView NewView = BufferGet2.Value.GetView();
+
+	// Make sure file openend for read before we wrote it still have old data
+	CHECK(OldView.GetSize() == Buffer.GetSize());
+	CHECK(memcmp(OldView.GetData(), Buffer.GetData(), OldView.GetSize()) == 0);
+
+	// Make sure we get the new data when reading after we write new data
+	CHECK(NewView.GetSize() == Buffer2.GetSize());
+	CHECK(memcmp(NewView.GetData(), Buffer2.GetData(), NewView.GetSize()) == 0);
 }
 
 #endif
diff --git a/zenserver/cache/structuredcachestore.h b/zenserver/cache/structuredcachestore.h
index 34b8d18f4..c226074fe 100644
--- a/zenserver/cache/structuredcachestore.h
+++ b/zenserver/cache/structuredcachestore.h
@@ -150,6 +150,7 @@ public:
 
 	bool	 Get(std::string_view Bucket, const IoHash& HashKey, ZenCacheValue& OutValue);
 	void	 Put(std::string_view Bucket, const IoHash& HashKey, const ZenCacheValue& Value);
+	void	 Drop();
 	bool	 DropBucket(std::string_view Bucket);
 	void	 Scrub(ScrubContext& Ctx);
 	void	 GatherAccessTimes(zen::access_tracking::AccessTimes& AccessTimes);
@@ -193,14 +194,16 @@ private:
 
 		bool			Get(const IoHash& HashKey, ZenCacheValue& OutValue);
 		void			Put(const IoHash& HashKey, const ZenCacheValue& Value);
+		void			Drop();
 		void			Scrub(ScrubContext& Ctx);
 		void			GatherAccessTimes(std::vector<zen::access_tracking::KeyAccessTime>& AccessTimes);
 		inline uint64_t TotalSize() const { return m_TotalSize; }
 	};
 
-	mutable RwLock								 m_Lock;
-	std::unordered_map<std::string, CacheBucket> m_Buckets;
-	Configuration								 m_Configuration;
+	mutable RwLock												  m_Lock;
+	std::unordered_map<std::string, std::unique_ptr<CacheBucket>> m_Buckets;
+	std::vector<std::unique_ptr<CacheBucket>>					  m_DroppedBuckets;
+	Configuration												  m_Configuration;
 
 	ZenCacheMemoryLayer(const ZenCacheMemoryLayer&) = delete;
 	ZenCacheMemoryLayer& operator=(const ZenCacheMemoryLayer&) = delete;
@@ -214,6 +217,7 @@ public:
 
 	bool Get(std::string_view Bucket, const IoHash& HashKey, ZenCacheValue& OutValue);
 	void Put(std::string_view Bucket, const IoHash& HashKey, const ZenCacheValue& Value);
+	bool Drop();
 	bool DropBucket(std::string_view Bucket);
 	void Flush();
 	void Scrub(ScrubContext& Ctx);
@@ -233,19 +237,16 @@ private:
 		CacheBucket(std::string BucketName);
 		~CacheBucket();
 
-		void		OpenOrCreate(std::filesystem::path BucketDir, bool AllowCreate = true);
-		static bool Delete(std::filesystem::path BucketDir);
-		bool		Get(const IoHash& HashKey, ZenCacheValue& OutValue);
-		void		Put(const IoHash& HashKey, const ZenCacheValue& Value);
-		void		Drop();
-		void		Flush();
-		void		SaveManifest();
-		void		Scrub(ScrubContext& Ctx);
-		void		GatherReferences(GcContext& GcCtx);
-		void		CollectGarbage(GcContext& GcCtx);
-		void		UpdateAccessTimes(const std::vector<zen::access_tracking::KeyAccessTime>& AccessTimes);
+		bool OpenOrCreate(std::filesystem::path BucketDir, bool AllowCreate = true);
+		bool Get(const IoHash& HashKey, ZenCacheValue& OutValue);
+		void Put(const IoHash& HashKey, const ZenCacheValue& Value);
+		bool Drop();
+		void Flush();
+		void Scrub(ScrubContext& Ctx);
+		void GatherReferences(GcContext& GcCtx);
+		void CollectGarbage(GcContext& GcCtx);
+		void UpdateAccessTimes(const std::vector<zen::access_tracking::KeyAccessTime>& AccessTimes);
 
-		inline bool		IsOk() const { return m_IsOk; }
 		inline uint64_t TotalSize() const { return m_TotalSize.load(std::memory_order::relaxed); }
 
 	private:
@@ -257,7 +258,6 @@ private:
 		std::filesystem::path m_BlocksBasePath;
 		BlockStore			  m_BlockStore;
 		Oid					  m_BucketId;
-		bool				  m_IsOk				 = false;
 		uint64_t			  m_LargeObjectThreshold = 64 * 1024;
 
 		// These files are used to manage storage of small objects for this bucket
@@ -292,16 +292,18 @@ private:
 
 		std::atomic_uint64_t m_TotalSize{};
 
-		void	 BuildPath(PathBuilderBase& Path, const IoHash& HashKey);
-		void	 PutStandaloneCacheValue(const IoHash& HashKey, const ZenCacheValue& Value);
-		bool	 GetStandaloneCacheValue(const DiskLocation& Loc, const IoHash& HashKey, ZenCacheValue& OutValue);
-		void	 PutInlineCacheValue(const IoHash& HashKey, const ZenCacheValue& Value);
-		bool	 GetInlineCacheValue(const DiskLocation& Loc, ZenCacheValue& OutValue);
-		void	 MakeIndexSnapshot();
-		uint64_t ReadIndexFile();
-		uint64_t ReadLog(uint64_t LogPosition);
-		uint64_t MigrateLegacyData(bool CleanSource);
-		void	 OpenLog(const std::filesystem::path& BucketDir, const bool IsNew);
+		void		BuildPath(PathBuilderBase& Path, const IoHash& HashKey);
+		void		PutStandaloneCacheValue(const IoHash& HashKey, const ZenCacheValue& Value);
+		bool		GetStandaloneCacheValue(const DiskLocation& Loc, const IoHash& HashKey, ZenCacheValue& OutValue);
+		void		PutInlineCacheValue(const IoHash& HashKey, const ZenCacheValue& Value);
+		bool		GetInlineCacheValue(const DiskLocation& Loc, ZenCacheValue& OutValue);
+		void		MakeIndexSnapshot();
+		uint64_t	ReadIndexFile();
+		uint64_t	ReadLog(uint64_t LogPosition);
+		uint64_t	MigrateLegacyData(bool CleanSource);
+		void		OpenLog(const std::filesystem::path& BucketDir, const bool IsNew);
+		static bool Delete(std::filesystem::path BucketDir);
+		void		SaveManifest();
 
 		// These locks are here to avoid contention on file creation, therefore it's sufficient
 		// that we take the same lock for the same hash
@@ -314,9 +316,10 @@ private:
 		inline RwLock& LockForHash(const IoHash& Hash) { return m_ShardedLocks[Hash.Hash[19]]; }
 	};
 
-	std::filesystem::path						 m_RootDir;
-	mutable RwLock								 m_Lock;
-	std::unordered_map<std::string, CacheBucket> m_Buckets;	 // TODO: make this case insensitive
+	std::filesystem::path										  m_RootDir;
+	mutable RwLock												  m_Lock;
+	std::unordered_map<std::string, std::unique_ptr<CacheBucket>> m_Buckets;  // TODO: make this case insensitive
+	std::vector<std::unique_ptr<CacheBucket>>					  m_DroppedBuckets;
 
 	ZenCacheDiskLayer(const ZenCacheDiskLayer&) = delete;
 	ZenCacheDiskLayer& operator=(const ZenCacheDiskLayer&) = delete;
@@ -330,6 +333,7 @@ public:
 
 	bool				  Get(std::string_view Bucket, const IoHash& HashKey, ZenCacheValue& OutValue);
 	void				  Put(std::string_view Bucket, const IoHash& HashKey, const ZenCacheValue& Value);
+	bool				  Drop();
 	bool				  DropBucket(std::string_view Bucket);
 	void				  Flush();
 	void				  Scrub(ScrubContext& Ctx);
@@ -360,12 +364,19 @@ public:
 		"!default!";  // This is intentionally not a valid namespace name and will only be used for mapping when no namespace is given
 	static constexpr std::string_view NamespaceDiskPrefix = "ns_";
 
-	ZenCacheStore(CasGc& Gc, std::filesystem::path BasePath);
+	struct Configuration
+	{
+		std::filesystem::path BasePath;
+		bool				  AllowAutomaticCreationOfNamespaces = true;
+	};
+
+	ZenCacheStore(CasGc& Gc, const Configuration& Configuration);
 	~ZenCacheStore();
 
 	bool Get(std::string_view Namespace, std::string_view Bucket, const IoHash& HashKey, ZenCacheValue& OutValue);
 	void Put(std::string_view Namespace, std::string_view Bucket, const IoHash& HashKey, const ZenCacheValue& Value);
 	bool DropBucket(std::string_view Namespace, std::string_view Bucket);
+	bool DropNamespace(std::string_view Namespace);
 	void Flush();
 	void Scrub(ScrubContext& Ctx);
 
@@ -377,10 +388,14 @@ private:
 	ZenCacheNamespace* GetNamespace(std::string_view Namespace);
 	void			   IterateNamespaces(const std::function<void(std::string_view Namespace, ZenCacheNamespace& Store)>& Callback) const;
 
-	typedef std::unordered_map<std::string, std::unique_ptr<ZenCacheNamespace>> NameSpaceMap;
+	typedef std::unordered_map<std::string, std::unique_ptr<ZenCacheNamespace>> NamespaceMap;
+
+	mutable RwLock									m_NamespacesLock;
+	NamespaceMap									m_Namespaces;
+	std::vector<std::unique_ptr<ZenCacheNamespace>> m_DroppedNamespaces;
 
-	mutable RwLock m_NamespacesLock;
-	NameSpaceMap   m_Namespaces;
+	CasGc&		  m_Gc;
+	Configuration m_Configuration;
 };
 
 void z$_forcelink();
diff --git a/zenserver/config.cpp b/zenserver/config.cpp
index be91ae4f8..c534865dc 100644
--- a/zenserver/config.cpp
+++ b/zenserver/config.cpp
@@ -478,6 +478,13 @@ ParseCliOptions(int argc, char* argv[], ZenServerOptions& ServerOptions)
 					   "Size of gc disk reserve in bytes.",
 					   cxxopts::value<uint64_t>(ServerOptions.GcConfig.DiskReserveSize)->default_value("268435456"),
 					   "");
+
+	options.add_option("gc",
+					   "",
+					   "gc-monitor-interval-seconds",
+					   "Garbage collection monitoring interval in seconds.",
+					   cxxopts::value<int32_t>(ServerOptions.GcConfig.MonitorIntervalSeconds)->default_value("30"),
+					   "");
 	try
 	{
 		auto result			  = options.parse(argc, argv);
@@ -770,8 +777,9 @@ ParseConfigFile(const std::filesystem::path& Path, ZenServerOptions& ServerOptio
 
 		if (sol::optional<sol::table> GcConfig = lua["gc"])
 		{
-			ServerOptions.GcConfig.IntervalSeconds = GcConfig.value().get_or("intervalseconds", 0);
-			ServerOptions.GcConfig.DiskReserveSize = GcConfig.value().get_or("diskreservesize", uint64_t(1u << 28));
+			ServerOptions.GcConfig.MonitorIntervalSeconds = GcConfig.value().get_or("monitorintervalseconds", 30);
+			ServerOptions.GcConfig.IntervalSeconds		  = GcConfig.value().get_or("intervalseconds", 0);
+			ServerOptions.GcConfig.DiskReserveSize		  = GcConfig.value().get_or("diskreservesize", uint64_t(1u << 28));
 
 			if (sol::optional<sol::table> CacheGcConfig = GcConfig.value()["cache"])
 			{
diff --git a/zenserver/config.h b/zenserver/config.h
index 49f039d8d..a07bba9a4 100644
--- a/zenserver/config.h
+++ b/zenserver/config.h
@@ -96,10 +96,11 @@ struct ZenGcConfig
 {
 	ZenCasEvictionPolicy   Cas;
 	ZenCacheEvictionPolicy Cache;
-	int32_t				   IntervalSeconds	   = 0;
-	bool				   CollectSmallObjects = true;
-	bool				   Enabled			   = true;
-	uint64_t			   DiskReserveSize	   = 1ul << 28;
+	int32_t				   MonitorIntervalSeconds = 30;
+	int32_t				   IntervalSeconds		  = 0;
+	bool				   CollectSmallObjects	  = true;
+	bool				   Enabled				  = true;
+	uint64_t			   DiskReserveSize		  = 1ul << 28;
 };
 
 struct ZenServerOptions
diff --git a/zenserver/upstream/hordecompute.cpp b/zenserver/upstream/hordecompute.cpp
index 2ec24b303..38c798569 100644
--- a/zenserver/upstream/hordecompute.cpp
+++ b/zenserver/upstream/hordecompute.cpp
@@ -157,8 +157,13 @@ namespace detail {
 
 				{
 					ApplyResult.Timepoints["zen-storage-build-ref"] = DateTime::NowTicks();
-					std::scoped_lock Lock(m_TaskMutex);
-					if (m_PendingTasks.contains(UpstreamData.TaskId))
+
+					bool AlreadyQueued;
+					{
+						std::scoped_lock Lock(m_TaskMutex);
+						AlreadyQueued = m_PendingTasks.contains(UpstreamData.TaskId);
+					}
+					if (AlreadyQueued)
 					{
 						// Pending task is already queued, return success
 						ApplyResult.Success = true;
@@ -171,7 +176,7 @@ namespace detail {
 				CloudCacheSession StorageSession(m_StorageClient);
 
 				{
-					CloudCacheResult Result = BatchPutBlobsIfMissing(StorageSession, UpstreamData.Blobs);
+					CloudCacheResult Result = BatchPutBlobsIfMissing(StorageSession, UpstreamData.Blobs, UpstreamData.CasIds);
 					ApplyResult.Bytes += Result.Bytes;
 					ApplyResult.ElapsedSeconds += Result.ElapsedSeconds;
 					ApplyResult.Timepoints["zen-storage-upload-blobs"] = DateTime::NowTicks();
@@ -182,6 +187,22 @@ namespace detail {
 						return ApplyResult;
 					}
 					UpstreamData.Blobs.clear();
+					UpstreamData.CasIds.clear();
+				}
+
+				{
+					CloudCacheResult Result = BatchPutCompressedBlobsIfMissing(StorageSession, UpstreamData.Cids);
+					ApplyResult.Bytes += Result.Bytes;
+					ApplyResult.ElapsedSeconds += Result.ElapsedSeconds;
+					ApplyResult.Timepoints["zen-storage-upload-compressed-blobs"] = DateTime::NowTicks();
+					if (!Result.Success)
+					{
+						ApplyResult.Error = {
+							.ErrorCode = Result.ErrorCode,
+							.Reason	   = !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to upload compressed blobs"};
+						return ApplyResult;
+					}
+					UpstreamData.Cids.clear();
 				}
 
 				{
@@ -279,9 +300,11 @@ namespace detail {
 			}
 		}
 
-		[[nodiscard]] CloudCacheResult BatchPutBlobsIfMissing(CloudCacheSession& Session, const std::map<IoHash, IoBuffer>& Blobs)
+		[[nodiscard]] CloudCacheResult BatchPutBlobsIfMissing(CloudCacheSession&				Session,
+															  const std::map<IoHash, IoBuffer>& Blobs,
+															  const std::set<IoHash>&			CasIds)
 		{
-			if (Blobs.size() == 0)
+			if (Blobs.size() == 0 && CasIds.size() == 0)
 			{
 				return {.Success = true};
 			}
@@ -292,6 +315,7 @@ namespace detail {
 			// Batch check for missing blobs
 			std::set<IoHash> Keys;
 			std::transform(Blobs.begin(), Blobs.end(), std::inserter(Keys, Keys.end()), [](const auto& It) { return It.first; });
+			Keys.insert(CasIds.begin(), CasIds.end());
 
 			CloudCacheExistsResult ExistsResult = Session.BlobExists(Session.Client().DefaultBlobStoreNamespace(), Keys);
 			Log().debug("Queried {} missing blobs Need={} Duration={}s Result={}",
@@ -310,7 +334,22 @@ namespace detail {
 
 			for (const auto& Hash : ExistsResult.Needs)
 			{
-				CloudCacheResult Result = Session.PutBlob(Session.Client().DefaultBlobStoreNamespace(), Hash, Blobs.at(Hash));
+				IoBuffer DataBuffer;
+				if (Blobs.contains(Hash))
+				{
+					DataBuffer = Blobs.at(Hash);
+				}
+				else
+				{
+					DataBuffer = m_CasStore.FindChunk(Hash);
+					if (!DataBuffer)
+					{
+						Log().warn("Put blob FAILED, input chunk '{}' missing", Hash);
+						return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .ErrorCode = -1, .Reason = "Failed to put blobs"};
+					}
+				}
+
+				CloudCacheResult Result = Session.PutBlob(Session.Client().DefaultBlobStoreNamespace(), Hash, DataBuffer);
 				Log().debug("Put blob {} Bytes={} Duration={}s Result={}", Hash, Result.Bytes, Result.ElapsedSeconds, Result.Success);
 				Bytes += Result.Bytes;
 				ElapsedSeconds += Result.ElapsedSeconds;
@@ -326,6 +365,62 @@ namespace detail {
 			return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
 		}
 
+		[[nodiscard]] CloudCacheResult BatchPutCompressedBlobsIfMissing(CloudCacheSession& Session, const std::set<IoHash>& Cids)
+		{
+			if (Cids.size() == 0)
+			{
+				return {.Success = true};
+			}
+
+			int64_t Bytes{};
+			double	ElapsedSeconds{};
+
+			// Batch check for missing compressed blobs
+			CloudCacheExistsResult ExistsResult = Session.CompressedBlobExists(Session.Client().DefaultBlobStoreNamespace(), Cids);
+			Log().debug("Queried {} missing compressed blobs Need={} Duration={}s Result={}",
+						Cids.size(),
+						ExistsResult.Needs.size(),
+						ExistsResult.ElapsedSeconds,
+						ExistsResult.Success);
+			ElapsedSeconds += ExistsResult.ElapsedSeconds;
+			if (!ExistsResult.Success)
+			{
+				return {
+					.Bytes			= Bytes,
+					.ElapsedSeconds = ElapsedSeconds,
+					.ErrorCode		= ExistsResult.ErrorCode ? ExistsResult.ErrorCode : -1,
+					.Reason = !ExistsResult.Reason.empty() ? std::move(ExistsResult.Reason) : "Failed to check if compressed blobs exist"};
+			}
+
+			for (const auto& Hash : ExistsResult.Needs)
+			{
+				IoBuffer DataBuffer = m_CidStore.FindChunkByCid(Hash);
+				if (!DataBuffer)
+				{
+					Log().warn("Put compressed blob FAILED, input CID chunk '{}' missing", Hash);
+					return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .ErrorCode = -1, .Reason = "Failed to put compressed blobs"};
+				}
+
+				CloudCacheResult Result = Session.PutCompressedBlob(Session.Client().DefaultBlobStoreNamespace(), Hash, DataBuffer);
+				Log().debug("Put compressed blob {} Bytes={} Duration={}s Result={}",
+							Hash,
+							Result.Bytes,
+							Result.ElapsedSeconds,
+							Result.Success);
+				Bytes += Result.Bytes;
+				ElapsedSeconds += Result.ElapsedSeconds;
+				if (!Result.Success)
+				{
+					return {.Bytes			= Bytes,
+							.ElapsedSeconds = ElapsedSeconds,
+							.ErrorCode		= Result.ErrorCode ? Result.ErrorCode : -1,
+							.Reason			= !Result.Reason.empty() ? std::move(Result.Reason) : "Failed to put compressed blobs"};
+				}
+			}
+
+			return {.Bytes = Bytes, .ElapsedSeconds = ElapsedSeconds, .Success = true};
+		}
+
 		[[nodiscard]] CloudCacheResult BatchPutObjectsIfMissing(CloudCacheSession& Session, const std::map<IoHash, CbObject>& Objects)
 		{
 			if (Objects.size() == 0)
@@ -599,6 +694,8 @@ namespace detail {
 		{
 			std::map<IoHash, IoBuffer> Blobs;
 			std::map<IoHash, CbObject> Objects;
+			std::set<IoHash>		   CasIds;
+			std::set<IoHash>		   Cids;
 			IoHash					   TaskId;
 			IoHash					   RequirementsId;
 		};
@@ -957,7 +1054,7 @@ namespace detail {
 			for (auto& It : ApplyRecord.WorkerDescriptor["executables"sv])
 			{
 				CbObjectView FileEntry = It.AsObjectView();
-				if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.Blobs))
+				if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.CasIds))
 				{
 					return false;
 				}
@@ -966,7 +1063,7 @@ namespace detail {
 			for (auto& It : ApplyRecord.WorkerDescriptor["files"sv])
 			{
 				CbObjectView FileEntry = It.AsObjectView();
-				if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.Blobs))
+				if (!ProcessFileEntry(FileEntry, InputFiles, InputFileHashes, Data.CasIds))
 				{
 					return false;
 				}
@@ -1034,11 +1131,10 @@ namespace detail {
 
 						bool AnyErrors = false;
 						ApplyRecord.Action.IterateAttachments([&](CbFieldView Field) {
-							const IoHash				Cid		   = Field.AsHash();
-							const std::filesystem::path FilePath   = {InputPath / Cid.ToHexString()};
-							IoBuffer					DataBuffer = m_CidStore.FindChunkByCid(Cid);
+							const IoHash				Cid		 = Field.AsHash();
+							const std::filesystem::path FilePath = {InputPath / Cid.ToHexString()};
 
-							if (!DataBuffer)
+							if (!m_CidStore.ContainsChunk(Cid))
 							{
 								Log().warn("process apply upstream FAILED, input CID chunk '{}' missing", Cid);
 								AnyErrors = true;
@@ -1050,11 +1146,9 @@ namespace detail {
 								return;
 							}
 
-							const IoHash CompressedId = IoHash::HashBuffer(DataBuffer.GetData(), DataBuffer.GetSize());
-
 							InputFiles.insert(FilePath);
-							InputFileHashes[FilePath] = CompressedId;
-							Data.Blobs[CompressedId]  = std::move(DataBuffer);
+							InputFileHashes[FilePath] = Cid;
+							Data.Cids.insert(Cid);
 						});
 
 						if (AnyErrors)
@@ -1067,7 +1161,7 @@ namespace detail {
 
 			const UpstreamDirectory RootDirectory = BuildDirectoryTree(InputFiles);
 
-			CbObject	 Sandbox	  = BuildMerkleTreeDirectory(RootDirectory, InputFileHashes, Data.Blobs, Data.Objects);
+			CbObject	 Sandbox	  = BuildMerkleTreeDirectory(RootDirectory, InputFileHashes, Data.Cids, Data.Objects);
 			const IoHash SandboxHash  = Sandbox.GetHash();
 			Data.Objects[SandboxHash] = std::move(Sandbox);
 
@@ -1118,28 +1212,18 @@ namespace detail {
 		[[nodiscard]] bool ProcessFileEntry(const CbObjectView&						 FileEntry,
 											std::set<std::filesystem::path>&		 InputFiles,
 											std::map<std::filesystem::path, IoHash>& InputFileHashes,
-											std::map<IoHash, IoBuffer>&				 Blobs)
+											std::set<IoHash>&						 CasIds)
 		{
-			const std::filesystem::path FilePath   = FileEntry["name"sv].AsString();
-			const IoHash				ChunkId	   = FileEntry["hash"sv].AsHash();
-			const uint64_t				Size	   = FileEntry["size"sv].AsUInt64();
-			IoBuffer					DataBuffer = m_CasStore.FindChunk(ChunkId);
+			const std::filesystem::path FilePath = FileEntry["name"sv].AsString();
+			const IoHash				ChunkId	 = FileEntry["hash"sv].AsHash();
+			const uint64_t				Size	 = FileEntry["size"sv].AsUInt64();
 
-			if (!DataBuffer)
+			if (!m_CasStore.ContainsChunk(ChunkId))
 			{
 				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' missing", ChunkId);
 				return false;
 			}
 
-			if (DataBuffer.Size() != Size)
-			{
-				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' size: {}, action spec expected {}",
-						   ChunkId,
-						   DataBuffer.Size(),
-						   Size);
-				return false;
-			}
-
 			if (InputFiles.contains(FilePath))
 			{
 				Log().warn("process apply upstream FAILED, worker CAS chunk '{}' size: {} duplicate filename {}", ChunkId, Size, FilePath);
@@ -1148,7 +1232,7 @@ namespace detail {
 
 			InputFiles.insert(FilePath);
 			InputFileHashes[FilePath] = ChunkId;
-			Blobs[ChunkId]			  = std::move(DataBuffer);
+			CasIds.insert(ChunkId);
 			return true;
 		}
 
@@ -1204,7 +1288,7 @@ namespace detail {
 
 		[[nodiscard]] CbObject BuildMerkleTreeDirectory(const UpstreamDirectory&					   RootDirectory,
 														const std::map<std::filesystem::path, IoHash>& InputFileHashes,
-														const std::map<IoHash, IoBuffer>&			   Blobs,
+														const std::set<IoHash>&						   Cids,
 														std::map<IoHash, CbObject>&					   Objects)
 		{
 			CbObjectWriter DirectoryTreeWriter;
@@ -1214,14 +1298,13 @@ namespace detail {
 				DirectoryTreeWriter.BeginArray("f"sv);
 				for (const auto& File : RootDirectory.Files)
 				{
-					const std::filesystem::path FilePath = {RootDirectory.Path / File};
-					const IoHash&				FileHash = InputFileHashes.at(FilePath);
-					const uint64_t				FileSize = Blobs.at(FileHash).Size();
+					const std::filesystem::path FilePath   = {RootDirectory.Path / File};
+					const IoHash&				FileHash   = InputFileHashes.at(FilePath);
+					const bool					Compressed = Cids.contains(FileHash);
 					DirectoryTreeWriter.BeginObject();
 					DirectoryTreeWriter.AddString("n"sv, File);
 					DirectoryTreeWriter.AddBinaryAttachment("h"sv, FileHash);
-					DirectoryTreeWriter.AddInteger("s"sv, FileSize);  // Size
-					// DirectoryTreeWriter.AddInteger("a"sv, 0);		  // Attributes Currently unneeded
+					DirectoryTreeWriter.AddBool("c"sv, Compressed);
 					DirectoryTreeWriter.EndObject();
 				}
 				DirectoryTreeWriter.EndArray();
@@ -1232,7 +1315,7 @@ namespace detail {
 				DirectoryTreeWriter.BeginArray("d"sv);
 				for (const auto& Item : RootDirectory.Directories)
 				{
-					CbObject	 Directory	   = BuildMerkleTreeDirectory(Item.second, InputFileHashes, Blobs, Objects);
+					CbObject	 Directory	   = BuildMerkleTreeDirectory(Item.second, InputFileHashes, Cids, Objects);
 					const IoHash DirectoryHash = Directory.GetHash();
 					Objects[DirectoryHash]	   = std::move(Directory);
 
diff --git a/zenserver/upstream/zen.cpp b/zenserver/upstream/zen.cpp
index efc75b5b4..0237ec346 100644
--- a/zenserver/upstream/zen.cpp
+++ b/zenserver/upstream/zen.cpp
@@ -408,14 +408,15 @@ ZenStructuredCacheSession::CheckHealth()
 }
 
 ZenCacheResult
-ZenStructuredCacheSession::GetCacheRecord(std::string_view Namespace, std::string_view BucketId, const IoHash& Key, ZenContentType Type)
+ZenStructuredCacheSession::GetCacheRecord(std::string_view, std::string_view BucketId, const IoHash& Key, ZenContentType Type)
 {
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/z$/";
-	if (Namespace != ZenCacheStore::DefaultNamespace)
-	{
-		Uri << Namespace << "/";
-	}
+	// TODO: DE20220530: Disable adding namespace into URL until we have updated the shared instances with namespace support
+	//	if (Namespace != ZenCacheStore::DefaultNamespace)
+	//	{
+	//		Uri << Namespace << "/";
+	//	}
 	Uri << BucketId << "/" << Key.ToHexString();
 
 	cpr::Session& Session = m_SessionState->GetSession();
@@ -437,17 +438,15 @@ ZenStructuredCacheSession::GetCacheRecord(std::string_view Namespace, std::strin
 }
 
 ZenCacheResult
-ZenStructuredCacheSession::GetCacheValue(std::string_view Namespace,
-										 std::string_view BucketId,
-										 const IoHash&	  Key,
-										 const IoHash&	  ValueContentId)
+ZenStructuredCacheSession::GetCacheValue(std::string_view, std::string_view BucketId, const IoHash& Key, const IoHash& ValueContentId)
 {
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/z$/";
-	if (Namespace != ZenCacheStore::DefaultNamespace)
-	{
-		Uri << Namespace << "/";
-	}
+	// TODO: DE20220530: Disable adding namespace into URL until we have updated the shared instances with namespace support
+	//	if (Namespace != ZenCacheStore::DefaultNamespace)
+	//	{
+	//		Uri << Namespace << "/";
+	//	}
 	Uri << BucketId << "/" << Key.ToHexString() << "/" << ValueContentId.ToHexString();
 
 	cpr::Session& Session = m_SessionState->GetSession();
@@ -466,11 +465,15 @@ ZenStructuredCacheSession::GetCacheValue(std::string_view Namespace,
 	const bool	   Success = Response.status_code == 200;
 	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
 
-	return {.Response = Buffer, .Bytes = Response.downloaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
+	return {.Response		= Buffer,
+			.Bytes			= Response.downloaded_bytes,
+			.ElapsedSeconds = Response.elapsed,
+			.Reason			= Response.reason,
+			.Success		= Success};
 }
 
 ZenCacheResult
-ZenStructuredCacheSession::PutCacheRecord(std::string_view Namespace,
+ZenStructuredCacheSession::PutCacheRecord(std::string_view,
 										  std::string_view BucketId,
 										  const IoHash&	   Key,
 										  IoBuffer		   Value,
@@ -478,10 +481,11 @@ ZenStructuredCacheSession::PutCacheRecord(std::string_view Namespace,
 {
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/z$/";
-	if (Namespace != ZenCacheStore::DefaultNamespace)
-	{
-		Uri << Namespace << "/";
-	}
+	// TODO: DE20220530: Disable adding namespace into URL until we have updated the shared instances with namespace support
+	//	if (Namespace != ZenCacheStore::DefaultNamespace)
+	//	{
+	//		Uri << Namespace << "/";
+	//	}
 	Uri << BucketId << "/" << Key.ToHexString();
 
 	cpr::Session& Session = m_SessionState->GetSession();
@@ -501,13 +505,12 @@ ZenStructuredCacheSession::PutCacheRecord(std::string_view Namespace,
 		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = std::move(Response.error.message)};
 	}
 
-	return {.Bytes			= Response.uploaded_bytes,
-			.ElapsedSeconds = Response.elapsed,
-			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
+	const bool Success = Response.status_code == 200 || Response.status_code == 201;
+	return {.Bytes = Response.uploaded_bytes, .ElapsedSeconds = Response.elapsed, .Reason = Response.reason, .Success = Success};
 }
 
 ZenCacheResult
-ZenStructuredCacheSession::PutCacheValue(std::string_view Namespace,
+ZenStructuredCacheSession::PutCacheValue(std::string_view,
 										 std::string_view BucketId,
 										 const IoHash&	  Key,
 										 const IoHash&	  ValueContentId,
@@ -515,10 +518,11 @@ ZenStructuredCacheSession::PutCacheValue(std::string_view Namespace,
 {
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_Client.ServiceUrl() << "/z$/";
-	if (Namespace != ZenCacheStore::DefaultNamespace)
-	{
-		Uri << Namespace << "/";
-	}
+	// TODO: DE20220530: Disable adding namespace into URL until we have updated the shared instances with namespace support
+	//	if (Namespace != ZenCacheStore::DefaultNamespace)
+	//	{
+	//		Uri << Namespace << "/";
+	//	}
 	Uri << BucketId << "/" << Key.ToHexString() << "/" << ValueContentId.ToHexString();
 
 	cpr::Session& Session = m_SessionState->GetSession();
@@ -535,9 +539,8 @@ ZenStructuredCacheSession::PutCacheValue(std::string_view Namespace,
 		return {.ErrorCode = static_cast<int32_t>(Response.error.code), .Reason = std::move(Response.error.message)};
 	}
 
-	return {.Bytes			= Response.uploaded_bytes,
-			.ElapsedSeconds = Response.elapsed,
-			.Success		= (Response.status_code == 200 || Response.status_code == 201)};
+	const bool Success = Response.status_code == 200 || Response.status_code == 201;
+	return {.Bytes = Response.uploaded_bytes, .ElapsedSeconds = Response.elapsed, .Reason = Response.reason, .Success = Success};
 }
 
 ZenCacheResult
@@ -566,7 +569,11 @@ ZenStructuredCacheSession::InvokeRpc(const CbObjectView& Request)
 	const bool	   Success = Response.status_code == 200;
 	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
 
-	return {.Response = std::move(Buffer), .Bytes = Response.uploaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
+	return {.Response		= std::move(Buffer),
+			.Bytes			= Response.uploaded_bytes,
+			.ElapsedSeconds = Response.elapsed,
+			.Reason			= Response.reason,
+			.Success		= Success};
 }
 
 ZenCacheResult
@@ -594,7 +601,11 @@ ZenStructuredCacheSession::InvokeRpc(const CbPackage& Request)
 	const bool	   Success = Response.status_code == 200;
 	const IoBuffer Buffer  = Success ? IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()) : IoBuffer();
 
-	return {.Response = std::move(Buffer), .Bytes = Response.uploaded_bytes, .ElapsedSeconds = Response.elapsed, .Success = Success};
+	return {.Response		= std::move(Buffer),
+			.Bytes			= Response.uploaded_bytes,
+			.ElapsedSeconds = Response.elapsed,
+			.Reason			= Response.reason,
+			.Success		= Success};
 }
 
 }  // namespace zen
diff --git a/zenserver/zenserver.cpp b/zenserver/zenserver.cpp
index 53a046f80..4db69c265 100644
--- a/zenserver/zenserver.cpp
+++ b/zenserver/zenserver.cpp
@@ -364,6 +364,7 @@ public:
 		ZEN_INFO("initializing GC, enabled '{}', interval {}s", ServerOptions.GcConfig.Enabled, ServerOptions.GcConfig.IntervalSeconds);
 		zen::GcSchedulerConfig GcConfig{
 			.RootDirectory		 = m_DataRoot / "gc",
+			.MonitorInterval	 = std::chrono::seconds(ServerOptions.GcConfig.MonitorIntervalSeconds),
 			.Interval			 = std::chrono::seconds(ServerOptions.GcConfig.IntervalSeconds),
 			.MaxCacheDuration	 = std::chrono::seconds(ServerOptions.GcConfig.Cache.MaxDurationSeconds),
 			.CollectSmallObjects = ServerOptions.GcConfig.CollectSmallObjects,
@@ -754,7 +755,9 @@ ZenServer::InitializeStructuredCache(const ZenServerOptions& ServerOptions)
 	using namespace std::literals;
 
 	ZEN_INFO("instantiating structured cache service");
-	m_CacheStore = std::make_unique<ZenCacheStore>(m_CasGc, m_DataRoot / "cache");
+	m_CacheStore = std::make_unique<ZenCacheStore>(
+		m_CasGc,
+		ZenCacheStore::Configuration{.BasePath = m_DataRoot / "cache", .AllowAutomaticCreationOfNamespaces = true});
 
 	const ZenUpstreamCacheConfig& UpstreamConfig = ServerOptions.UpstreamCacheConfig;
 
@@ -1161,7 +1164,7 @@ test_main(int argc, char** argv)
 
 	zen::MaximizeOpenFileCount();
 
-  	return ZEN_RUN_TESTS(argc, argv);
+	return ZEN_RUN_TESTS(argc, argv);
 }
 #endif