Enforce Overwrite Prevention According To Cache Policy

Overwrite with differing value should be denied if QueryLocal is not present and StoreLocal is present.  Overwrite with equal value should succeed regardless of policy flags.

1 files changed, 39 insertions, 23 deletions

diff --git a/src/zenstore/cache/structuredcachestore.cpp b/src/zenstore/cache/structuredcachestore.cpp
index 133cb42d7..25fd23b93 100644
--- a/src/zenstore/cache/structuredcachestore.cpp
+++ b/src/zenstore/cache/structuredcachestore.cpp
@@ -252,11 +252,12 @@ ZenCacheNamespace::Get(std::string_view InBucket, const IoHash& HashKey, GetBatc
 	return;
 }
 
-void
+bool
 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++;
+	bool							   RetVal = m_DiskLayer.Put(InBucket, HashKey, Value, References, Overwrite, DiskLayerBatchHandle);
+	if (RetVal)
+	{
+		m_WriteCount++;
+	}
+	return RetVal;
 }
 
 bool
@@ -716,13 +721,14 @@ ZenCacheStore::Get(const CacheRequestContext& Context,
 	m_MissCount++;
 }
 
-void
+bool
 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
@@ -730,7 +736,7 @@ ZenCacheStore::Put(const CacheRequestContext& Context,
 	if (IsKnownBadBucketName(Bucket))
 	{
 		m_RejectedWriteCount++;
-		return;
+		return false;
 	}
 
 	ZEN_MEMSCOPE(GetCacheStoreTag());
@@ -760,9 +766,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;
+		bool							   RetVal	   = Store->Put(Bucket, HashKey, Value, References, Overwrite, BatchHandle);
+		if (RetVal)
+		{
+			m_WriteCount++;
+		}
+		else
+		{
+			m_RejectedWriteCount++;
+		}
+		return RetVal;
 	}
 
 	ZEN_WARN("request for unknown namespace '{}' in ZenCacheStore::Put [{}] bucket '{}', key '{}'",
@@ -770,6 +783,7 @@ ZenCacheStore::Put(const CacheRequestContext& Context,
 			 Namespace,
 			 Bucket,
 			 HashKey.ToHexString());
+	return false;
 }
 
 bool
@@ -1363,7 +1377,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)
@@ -1417,7 +1431,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();
@@ -1471,7 +1485,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();
@@ -1554,7 +1568,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);
 				});
 			}
@@ -1635,7 +1649,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);
 				});
@@ -1740,14 +1754,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));
 	}
 
@@ -1763,7 +1777,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));
@@ -1805,7 +1819,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) {
@@ -1878,7 +1892,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) {
@@ -1964,7 +1978,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));
@@ -1974,7 +1988,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();
 
@@ -2065,7 +2079,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);
 			}
 		};
 
@@ -2114,7 +2128,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());
@@ -2130,7 +2145,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;
 	};