Control overwrite enforcement with a config setting

1 files changed, 140 insertions, 45 deletions

diff --git a/src/zenserver-test/zenserver-test.cpp b/src/zenserver-test/zenserver-test.cpp
index a41eaed75..f469fc60e 100644
--- a/src/zenserver-test/zenserver-test.cpp
+++ b/src/zenserver-test/zenserver-test.cpp
@@ -549,9 +549,13 @@ namespace utils {
 			return ZenConfig{.DataDir = TestEnv.CreateNewTestDir(), .Port = Port, .Args = std::move(Args)};
 		}
 
-		static ZenConfig NewWithUpstream(uint16_t Port, uint16_t UpstreamPort)
+		static ZenConfig NewWithUpstream(uint16_t Port, uint16_t UpstreamPort, std::string Args = "")
 		{
-			return New(Port, fmt::format("--debug --upstream-thread-count=0 --upstream-zen-url=http://localhost:{}", UpstreamPort));
+			return New(Port,
+					   fmt::format("{}{}--debug --upstream-thread-count=0 --upstream-zen-url=http://localhost:{}",
+								   Args,
+								   Args.length() > 0 ? " " : "",
+								   UpstreamPort));
 		}
 
 		static ZenConfig NewWithThreadedUpstreams(uint16_t NewPort, std::span<uint16_t> UpstreamPorts, bool Debug)
@@ -1503,7 +1507,7 @@ TEST_CASE("zcache.rpc")
 		}
 	}
 
-	SUBCASE("policy - 'QueryLocal' on put denies overwrite with differing value")
+	SUBCASE("policy - 'QueryLocal' on put allows overwrite with differing value when not limiting overwrites")
 	{
 		using namespace utils;
 
@@ -1519,7 +1523,7 @@ TEST_CASE("zcache.rpc")
 		std::string_view		   Namespace("ue4.ddc"sv);
 		std::string_view		   Bucket("mastodon"sv);
 		const size_t			   NumRecords = 4;
-		std::vector<zen::CacheKey> Keys		  = PutCacheRecords(UpstreamCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize);
+		std::vector<zen::CacheKey> Keys		  = PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize);
 
 		for (const zen::CacheKey& CacheKey : Keys)
 		{
@@ -1530,7 +1534,7 @@ TEST_CASE("zcache.rpc")
 			CHECK(Request.Format(Package));
 
 			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
-			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", UpstreamCfg.BaseUri)},
+			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
 											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
 											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
 
@@ -1541,28 +1545,103 @@ TEST_CASE("zcache.rpc")
 			CHECK(ParsedResult.Parse(Response));
 			for (bool ResponseSuccess : ParsedResult.Success)
 			{
-				CHECK(!ResponseSuccess);
+				CHECK(ResponseSuccess);
 			}
 		}
 
-		CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
-		GetCacheRecordResult Result = GetCacheRecords(LocalCfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+		auto CheckRecordCorrectness = [&](const ZenConfig& Cfg) {
+			CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+			GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
 
-		CHECK(Result.Result.Results.size() == Keys.size());
+			CHECK(Result.Result.Results.size() == Keys.size());
 
-		for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
+			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
+			{
+				CHECK(Record);
+				const CacheKey& ExpectedKey = Keys[Index++];
+				CHECK(Record->Key == ExpectedKey);
+				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				{
+					CHECK(Value.RawSize == PayloadSize * 2);
+				}
+			}
+		};
+
+		// Check that the records are present and overwritten in the local server
+		CheckRecordCorrectness(LocalCfg);
+
+		// Check that the records are present and overwritten in the upstream server
+		CheckRecordCorrectness(UpstreamCfg);
+	}
+
+	SUBCASE("policy - 'QueryLocal' on put denies overwrite with differing value when limiting overwrites")
+	{
+		using namespace utils;
+
+		ZenConfig		  UpstreamCfg = ZenConfig::New(TestEnv.GetNewPortNumber());
+		ZenServerInstance UpstreamServer(TestEnv);
+		SpawnServer(UpstreamServer, UpstreamCfg);
+
+		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port, "--cache-limit-overwrites");
+		ZenServerInstance LocalServer(TestEnv);
+		SpawnServer(LocalServer, LocalCfg);
+
+		size_t					   PayloadSize = 1024;
+		std::string_view		   Namespace("ue4.ddc"sv);
+		std::string_view		   Bucket("mastodon"sv);
+		const size_t			   NumRecords = 4;
+		std::vector<zen::CacheKey> Keys		  = PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize);
+
+		for (const zen::CacheKey& CacheKey : Keys)
 		{
-			CHECK(Record);
-			const CacheKey& ExpectedKey = Keys[Index++];
-			CHECK(Record->Key == ExpectedKey);
-			for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+			cacherequests::PutCacheRecordsRequest Request = {.AcceptMagic = kCbPkgMagic, .Namespace = std::string(Namespace)};
+			AppendCacheRecord(Request, CacheKey, PayloadSize * 2, CachePolicy::Default);
+
+			CbPackage Package;
+			CHECK(Request.Format(Package));
+
+			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
+			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
+											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
+											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
+
+			CHECK(Result.status_code == 200);
+			cacherequests::PutCacheRecordsResult ParsedResult;
+			CbPackage Response = ParsePackageMessage(zen::IoBuffer(zen::IoBuffer::Wrap, Result.text.data(), Result.text.size()));
+			CHECK(!Response.IsNull());
+			CHECK(ParsedResult.Parse(Response));
+			for (bool ResponseSuccess : ParsedResult.Success)
 			{
-				CHECK(Value.RawSize == PayloadSize);
+				CHECK(!ResponseSuccess);
 			}
 		}
+
+		auto CheckRecordCorrectness = [&](const ZenConfig& Cfg) {
+			CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+			GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+
+			CHECK(Result.Result.Results.size() == Keys.size());
+
+			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
+			{
+				CHECK(Record);
+				const CacheKey& ExpectedKey = Keys[Index++];
+				CHECK(Record->Key == ExpectedKey);
+				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				{
+					CHECK(Value.RawSize == PayloadSize);
+				}
+			}
+		};
+
+		// Check that the records are present and not overwritten in the local server
+		CheckRecordCorrectness(LocalCfg);
+
+		// Check that the records are present and not overwritten in the upstream server
+		CheckRecordCorrectness(UpstreamCfg);
 	}
 
-	SUBCASE("policy - no 'QueryLocal' on put allows overwrite with differing value")
+	SUBCASE("policy - no 'QueryLocal' on put allows overwrite with differing value when limiting overwrites")
 	{
 		using namespace utils;
 
@@ -1570,7 +1649,7 @@ TEST_CASE("zcache.rpc")
 		ZenServerInstance UpstreamServer(TestEnv);
 		SpawnServer(UpstreamServer, UpstreamCfg);
 
-		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port);
+		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port, "--cache-limit-overwrites");
 		ZenServerInstance LocalServer(TestEnv);
 		SpawnServer(LocalServer, LocalCfg);
 
@@ -1578,7 +1657,7 @@ TEST_CASE("zcache.rpc")
 		std::string_view		   Namespace("ue4.ddc"sv);
 		std::string_view		   Bucket("mastodon"sv);
 		const size_t			   NumRecords = 4;
-		std::vector<zen::CacheKey> Keys		  = PutCacheRecords(UpstreamCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize);
+		std::vector<zen::CacheKey> Keys		  = PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize);
 
 		for (const zen::CacheKey& CacheKey : Keys)
 		{
@@ -1589,7 +1668,7 @@ TEST_CASE("zcache.rpc")
 			CHECK(Request.Format(Package));
 
 			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
-			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", UpstreamCfg.BaseUri)},
+			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
 											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
 											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
 
@@ -1604,24 +1683,32 @@ TEST_CASE("zcache.rpc")
 			}
 		}
 
-		CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
-		GetCacheRecordResult Result = GetCacheRecords(LocalCfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+		auto CheckRecordCorrectness = [&](const ZenConfig& Cfg) {
+			CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+			GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
 
-		CHECK(Result.Result.Results.size() == Keys.size());
+			CHECK(Result.Result.Results.size() == Keys.size());
 
-		for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
-		{
-			CHECK(Record);
-			const CacheKey& ExpectedKey = Keys[Index++];
-			CHECK(Record->Key == ExpectedKey);
-			for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
 			{
-				CHECK(Value.RawSize == PayloadSize * 2);
+				CHECK(Record);
+				const CacheKey& ExpectedKey = Keys[Index++];
+				CHECK(Record->Key == ExpectedKey);
+				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				{
+					CHECK(Value.RawSize == PayloadSize * 2);
+				}
 			}
-		}
+		};
+
+		// Check that the records are present and overwritten in the local server
+		CheckRecordCorrectness(LocalCfg);
+
+		// Check that the records are present and overwritten in the upstream server
+		CheckRecordCorrectness(UpstreamCfg);
 	}
 
-	SUBCASE("policy - 'QueryLocal' on put allows overwrite with equivalent value")
+	SUBCASE("policy - 'QueryLocal' on put allows overwrite with equivalent value when limiting overwrites")
 	{
 		using namespace utils;
 
@@ -1629,7 +1716,7 @@ TEST_CASE("zcache.rpc")
 		ZenServerInstance UpstreamServer(TestEnv);
 		SpawnServer(UpstreamServer, UpstreamCfg);
 
-		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port);
+		ZenConfig		  LocalCfg = ZenConfig::NewWithUpstream(TestEnv.GetNewPortNumber(), UpstreamCfg.Port, "--cache-limit-overwrites");
 		ZenServerInstance LocalServer(TestEnv);
 		SpawnServer(LocalServer, LocalCfg);
 
@@ -1638,12 +1725,12 @@ TEST_CASE("zcache.rpc")
 		std::string_view		   Bucket("mastodon"sv);
 		const size_t			   NumRecords = 4;
 		std::vector<CbPackage>	   Packages;
-		std::vector<zen::CacheKey> Keys = PutCacheRecords(UpstreamCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize, 1, &Packages);
+		std::vector<zen::CacheKey> Keys = PutCacheRecords(LocalCfg.BaseUri, Namespace, Bucket, NumRecords, PayloadSize, 1, &Packages);
 
 		for (const CbPackage& Package : Packages)
 		{
 			IoBuffer	  Body	 = FormatPackageMessageBuffer(Package).Flatten().AsIoBuffer();
-			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", UpstreamCfg.BaseUri)},
+			cpr::Response Result = cpr::Post(cpr::Url{fmt::format("{}/$rpc", LocalCfg.BaseUri)},
 											 cpr::Header{{"Content-Type", "application/x-ue-cbpkg"}, {"Accept", "application/x-ue-cbpkg"}},
 											 cpr::Body{(const char*)Body.GetData(), Body.GetSize()});
 
@@ -1658,21 +1745,29 @@ TEST_CASE("zcache.rpc")
 			}
 		}
 
-		CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
-		GetCacheRecordResult Result = GetCacheRecords(LocalCfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
+		auto CheckRecordCorrectness = [&](const ZenConfig& Cfg) {
+			CachePolicy			 Policy = (CachePolicy::QueryLocal | CachePolicy::QueryRemote);
+			GetCacheRecordResult Result = GetCacheRecords(Cfg.BaseUri, "ue4.ddc"sv, Keys, Policy);
 
-		CHECK(Result.Result.Results.size() == Keys.size());
+			CHECK(Result.Result.Results.size() == Keys.size());
 
-		for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
-		{
-			CHECK(Record);
-			const CacheKey& ExpectedKey = Keys[Index++];
-			CHECK(Record->Key == ExpectedKey);
-			for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+			for (size_t Index = 0; const std::optional<cacherequests::GetCacheRecordResult>& Record : Result.Result.Results)
 			{
-				CHECK(Value.RawSize == PayloadSize);
+				CHECK(Record);
+				const CacheKey& ExpectedKey = Keys[Index++];
+				CHECK(Record->Key == ExpectedKey);
+				for (const cacherequests::GetCacheRecordResultValue& Value : Record->Values)
+				{
+					CHECK(Value.RawSize == PayloadSize);
+				}
 			}
-		}
+		};
+
+		// Check that the records are present and unchanged in the local server
+		CheckRecordCorrectness(LocalCfg);
+
+		// Check that the records are present and unchanged in the upstream server
+		CheckRecordCorrectness(UpstreamCfg);
 	}
 
 	SUBCASE("RpcAcceptOptions")