Merge pull request #434 from ue-foundation/zs/put-overwrite-policy

Zs/put overwrite policy

1 files changed, 41 insertions, 25 deletions

diff --git a/src/zenstore/cache/structuredcachestore.cpp b/src/zenstore/cache/structuredcachestore.cpp
index d956384ca..973af52b2 100644
--- a/src/zenstore/cache/structuredcachestore.cpp
+++ b/src/zenstore/cache/structuredcachestore.cpp
@@ -154,7 +154,7 @@ struct ZenCacheNamespace::PutBatchHandle
 };
 
 ZenCacheNamespace::PutBatchHandle*
-ZenCacheNamespace::BeginPutBatch(std::vector<bool>& OutResult)
+ZenCacheNamespace::BeginPutBatch(std::vector<PutResult>& OutResult)
 {
 	ZenCacheNamespace::PutBatchHandle* Handle = new ZenCacheNamespace::PutBatchHandle;
 	Handle->DiskLayerHandle					  = m_DiskLayer.BeginPutBatch(OutResult);
@@ -252,11 +252,12 @@ ZenCacheNamespace::Get(std::string_view InBucket, const IoHash& HashKey, GetBatc
 	return;
 }
 
-void
+ZenCacheNamespace::PutResult
 ZenCacheNamespace::Put(std::string_view		InBucket,
 					   const IoHash&		HashKey,
 					   const ZenCacheValue& Value,
 					   std::span<IoHash>	References,
+					   bool					Overwrite,
 					   PutBatchHandle*		OptionalBatchHandle)
 {
 	ZEN_TRACE_CPU(OptionalBatchHandle ? "Z$::Namespace::Put(Batched)" : "Z$::Namespace::Put");
@@ -268,8 +269,12 @@ ZenCacheNamespace::Put(std::string_view		InBucket,
 	ZEN_ASSERT(Value.Value.Size());
 
 	ZenCacheDiskLayer::PutBatchHandle* DiskLayerBatchHandle = OptionalBatchHandle ? OptionalBatchHandle->DiskLayerHandle : nullptr;
-	m_DiskLayer.Put(InBucket, HashKey, Value, References, DiskLayerBatchHandle);
-	m_WriteCount++;
+	PutResult						   RetVal = m_DiskLayer.Put(InBucket, HashKey, Value, References, Overwrite, DiskLayerBatchHandle);
+	if (RetVal.Status == zen::PutStatus::Success)
+	{
+		m_WriteCount++;
+	}
+	return RetVal;
 }
 
 bool
@@ -557,7 +562,7 @@ ZenCacheStore::LogWorker()
 	}
 }
 
-ZenCacheStore::PutBatch::PutBatch(ZenCacheStore& CacheStore, std::string_view InNamespace, std::vector<bool>& OutResult)
+ZenCacheStore::PutBatch::PutBatch(ZenCacheStore& CacheStore, std::string_view InNamespace, std::vector<PutResult>& OutResult)
 : m_CacheStore(CacheStore)
 {
 	ZEN_MEMSCOPE(GetCacheStoreTag());
@@ -720,13 +725,14 @@ ZenCacheStore::Get(const CacheRequestContext& Context,
 	m_MissCount++;
 }
 
-void
+ZenCacheStore::PutResult
 ZenCacheStore::Put(const CacheRequestContext& Context,
 				   std::string_view			  Namespace,
 				   std::string_view			  Bucket,
 				   const IoHash&			  HashKey,
 				   const ZenCacheValue&		  Value,
 				   std::span<IoHash>		  References,
+				   bool						  Overwrite,
 				   PutBatch*				  OptionalBatchHandle)
 {
 	// Ad hoc rejection of known bad usage patterns for DDC bucket names
@@ -734,7 +740,7 @@ ZenCacheStore::Put(const CacheRequestContext& Context,
 	if (IsKnownBadBucketName(Bucket))
 	{
 		m_RejectedWriteCount++;
-		return;
+		return PutResult{zen::PutStatus::Invalid, "Bad bucket name"};
 	}
 
 	ZEN_MEMSCOPE(GetCacheStoreTag());
@@ -764,9 +770,16 @@ ZenCacheStore::Put(const CacheRequestContext& Context,
 	if (ZenCacheNamespace* Store = GetNamespace(Namespace); Store)
 	{
 		ZenCacheNamespace::PutBatchHandle* BatchHandle = OptionalBatchHandle ? OptionalBatchHandle->m_NamespaceBatchHandle : nullptr;
-		Store->Put(Bucket, HashKey, Value, References, BatchHandle);
-		m_WriteCount++;
-		return;
+		PutResult						   RetVal	   = Store->Put(Bucket, HashKey, Value, References, Overwrite, BatchHandle);
+		if (RetVal.Status == zen::PutStatus::Success)
+		{
+			m_WriteCount++;
+		}
+		else
+		{
+			m_RejectedWriteCount++;
+		}
+		return RetVal;
 	}
 
 	ZEN_WARN("request for unknown namespace '{}' in ZenCacheStore::Put [{}] bucket '{}', key '{}'",
@@ -774,6 +787,7 @@ ZenCacheStore::Put(const CacheRequestContext& Context,
 			 Namespace,
 			 Bucket,
 			 HashKey.ToHexString());
+	return PutResult{zen::PutStatus::Fail, fmt::format("Unknown namespace '{}'", Namespace)};
 }
 
 bool
@@ -1378,7 +1392,7 @@ TEST_CASE("cachestore.store")
 		Value.Value = Obj.GetBuffer().AsIoBuffer();
 		Value.Value.SetContentType(ZenContentType::kCbObject);
 
-		Zcs.Put("test_bucket"sv, Key, Value, {});
+		Zcs.Put("test_bucket"sv, Key, Value, {}, false);
 	}
 
 	for (int i = 0; i < kIterationCount; ++i)
@@ -1432,7 +1446,7 @@ TEST_CASE("cachestore.size")
 				const size_t Bucket		= Key % 4;
 				std::string	 BucketName = fmt::format("test_bucket-{}", Bucket);
 				IoHash		 Hash		= IoHash::HashBuffer(&Key, sizeof(uint32_t));
-				Zcs.Put(BucketName, Hash, ZenCacheValue{.Value = Buffer}, {});
+				Zcs.Put(BucketName, Hash, ZenCacheValue{.Value = Buffer}, {}, false);
 				Keys.push_back({BucketName, Hash});
 			}
 			CacheSize = Zcs.StorageSize();
@@ -1486,7 +1500,7 @@ TEST_CASE("cachestore.size")
 			for (size_t Key = 0; Key < Count; ++Key)
 			{
 				const size_t Bucket = Key % 4;
-				Zcs.Put(fmt::format("test_bucket-{}", Bucket), IoHash::HashBuffer(&Key, sizeof(uint32_t)), {.Value = Buffer}, {});
+				Zcs.Put(fmt::format("test_bucket-{}", Bucket), IoHash::HashBuffer(&Key, sizeof(uint32_t)), {.Value = Buffer}, {}, false);
 			}
 
 			CacheSize = Zcs.StorageSize();
@@ -1569,7 +1583,7 @@ TEST_CASE("cachestore.threadedinsert")	// * doctest::skip(true))
 			for (const auto& Chunk : Chunks)
 			{
 				ThreadPool.ScheduleWork([&Zcs, &WorkCompleted, &Chunk]() {
-					Zcs.Put(Chunk.second.Bucket, Chunk.first, {.Value = Chunk.second.Buffer}, {});
+					Zcs.Put(Chunk.second.Bucket, Chunk.first, {.Value = Chunk.second.Buffer}, {}, false);
 					WorkCompleted.fetch_add(1);
 				});
 			}
@@ -1650,7 +1664,7 @@ TEST_CASE("cachestore.threadedinsert")	// * doctest::skip(true))
 			for (const auto& Chunk : NewChunks)
 			{
 				ThreadPool.ScheduleWork([&Zcs, &WorkCompleted, Chunk, &AddedChunkCount]() {
-					Zcs.Put(Chunk.second.Bucket, Chunk.first, {.Value = Chunk.second.Buffer}, {});
+					Zcs.Put(Chunk.second.Bucket, Chunk.first, {.Value = Chunk.second.Buffer}, {}, false);
 					AddedChunkCount.fetch_add(1);
 					WorkCompleted.fetch_add(1);
 				});
@@ -1755,14 +1769,14 @@ TEST_CASE("cachestore.namespaces")
 		Buffer.SetContentType(ZenContentType::kCbObject);
 
 		ZenCacheValue PutValue = {.Value = Buffer};
-		Zcs.Put(Context, ZenCacheStore::DefaultNamespace, Bucket, Key1, PutValue, {});
+		Zcs.Put(Context, ZenCacheStore::DefaultNamespace, Bucket, Key1, PutValue, {}, false);
 
 		ZenCacheValue GetValue;
 		CHECK(Zcs.Get(Context, ZenCacheStore::DefaultNamespace, Bucket, Key1, GetValue));
 		CHECK(!Zcs.Get(Context, CustomNamespace, Bucket, Key1, GetValue));
 
 		// This should just be dropped as we don't allow creating of namespaces on the fly
-		Zcs.Put(Context, CustomNamespace, Bucket, Key1, PutValue, {});
+		Zcs.Put(Context, CustomNamespace, Bucket, Key1, PutValue, {}, false);
 		CHECK(!Zcs.Get(Context, CustomNamespace, Bucket, Key1, GetValue));
 	}
 
@@ -1778,7 +1792,7 @@ TEST_CASE("cachestore.namespaces")
 		IoBuffer Buffer2 = CacheValue2.GetBuffer().AsIoBuffer();
 		Buffer2.SetContentType(ZenContentType::kCbObject);
 		ZenCacheValue PutValue2 = {.Value = Buffer2};
-		Zcs.Put(Context, CustomNamespace, Bucket, Key2, PutValue2, {});
+		Zcs.Put(Context, CustomNamespace, Bucket, Key2, PutValue2, {}, false);
 
 		ZenCacheValue GetValue;
 		CHECK(!Zcs.Get(Context, ZenCacheStore::DefaultNamespace, Bucket, Key2, GetValue));
@@ -1820,7 +1834,7 @@ TEST_CASE("cachestore.drop.bucket")
 		Buffer.SetContentType(ZenContentType::kCbObject);
 
 		ZenCacheValue PutValue = {.Value = Buffer};
-		Zcs.Put(Context, Namespace, Bucket, Key, PutValue, {});
+		Zcs.Put(Context, Namespace, Bucket, Key, PutValue, {}, false);
 		return Key;
 	};
 	auto GetValue = [&Context](ZenCacheStore& Zcs, std::string_view Namespace, std::string_view Bucket, const IoHash& Key) {
@@ -1893,7 +1907,7 @@ TEST_CASE("cachestore.drop.namespace")
 		Buffer.SetContentType(ZenContentType::kCbObject);
 
 		ZenCacheValue PutValue = {.Value = Buffer};
-		Zcs.Put(Context, Namespace, Bucket, Key, PutValue, {});
+		Zcs.Put(Context, Namespace, Bucket, Key, PutValue, {}, false);
 		return Key;
 	};
 	auto GetValue = [&Context](ZenCacheStore& Zcs, std::string_view Namespace, std::string_view Bucket, const IoHash& Key) {
@@ -1979,7 +1993,7 @@ TEST_CASE("cachestore.blocked.disklayer.put")
 
 	size_t Key	   = Buffer.Size();
 	IoHash HashKey = IoHash::HashBuffer(&Key, sizeof(uint32_t));
-	Zcs.Put("test_bucket", HashKey, {.Value = Buffer}, {});
+	Zcs.Put("test_bucket", HashKey, {.Value = Buffer}, {}, false);
 
 	ZenCacheValue BufferGet;
 	CHECK(Zcs.Get("test_bucket", HashKey, BufferGet));
@@ -1989,7 +2003,7 @@ TEST_CASE("cachestore.blocked.disklayer.put")
 	Buffer2.SetContentType(ZenContentType::kCbObject);
 
 	// We should be able to overwrite even if the file is open for read
-	Zcs.Put("test_bucket", HashKey, {.Value = Buffer2}, {});
+	Zcs.Put("test_bucket", HashKey, {.Value = Buffer2}, {}, false);
 
 	MemoryView OldView = BufferGet.Value.GetView();
 
@@ -2080,7 +2094,7 @@ TEST_CASE("cachestore.scrub")
 					AttachmentHashes.push_back(Attachment.DecodeRawHash());
 					CidStore.AddChunk(Attachment.GetCompressed().Flatten().AsIoBuffer(), AttachmentHashes.back());
 				}
-				Zcs.Put("mybucket", Cid, {.Value = Record.Record}, AttachmentHashes);
+				Zcs.Put("mybucket", Cid, {.Value = Record.Record}, AttachmentHashes, false);
 			}
 		};
 
@@ -2129,7 +2143,8 @@ TEST_CASE("cachestore.newgc.basics")
 					{.Value	  = Record.second,
 					 .RawSize = Record.second.GetSize(),
 					 .RawHash = IoHash::HashBuffer(Record.second.GetData(), Record.second.GetSize())},
-					AttachmentKeys);
+					AttachmentKeys,
+					false);
 			for (const auto& Attachment : Attachments)
 			{
 				CidStore.AddChunk(Attachment.second.GetCompressed().Flatten().AsIoBuffer(), Attachment.second.DecodeRawHash());
@@ -2145,7 +2160,8 @@ TEST_CASE("cachestore.newgc.basics")
 				{.Value	  = CacheValue.second,
 				 .RawSize = CacheValue.second.GetSize(),
 				 .RawHash = IoHash::HashBuffer(CacheValue.second.GetData(), CacheValue.second.GetSize())},
-				{});
+				{},
+				false);
 		CacheEntries.insert({Key, CacheEntry{CacheValue.second, {}}});
 		return Key;
 	};