s3 dehydration touch cas (#977)

* add Touch() function to s3 client
* touch all used cas files in s3 dehydration path

3 files changed, 162 insertions, 25 deletions

diff --git a/src/zenserver/hub/hydration.cpp b/src/zenserver/hub/hydration.cpp
index 72e977bf2..266296fa9 100644
--- a/src/zenserver/hub/hydration.cpp
+++ b/src/zenserver/hub/hydration.cpp
@@ -87,23 +87,24 @@ namespace hydration_impl {
 		virtual void				Configure(std::string_view			   ModuleId,
 											  const std::filesystem::path& TempDir,
 											  std::string_view			   TargetSpecification,
-											  const CbObject&			   Options)						 = 0;
-		virtual void				SaveMetadata(const CbObject& Data)						 = 0;
-		virtual CbObject			LoadMetadata()											 = 0;
-		virtual CbObject			GetSettings()											 = 0;
-		virtual void				ParseSettings(const CbObjectView& Settings)				 = 0;
-		virtual std::vector<IoHash> List()													 = 0;
+											  const CbObject&			   Options)											= 0;
+		virtual void				SaveMetadata(const CbObject& Data)											= 0;
+		virtual CbObject			LoadMetadata()																= 0;
+		virtual CbObject			GetSettings()																= 0;
+		virtual void				ParseSettings(const CbObjectView& Settings)									= 0;
+		virtual std::vector<IoHash> List()																		= 0;
 		virtual void				Put(ParallelWork&				 Work,
 										WorkerThreadPool&			 WorkerPool,
 										const IoHash&				 Hash,
 										uint64_t					 Size,
-										const std::filesystem::path& SourcePath)			 = 0;
+										const std::filesystem::path& SourcePath)								= 0;
 		virtual void				Get(ParallelWork&				 Work,
 										WorkerThreadPool&			 WorkerPool,
 										const IoHash&				 Hash,
 										uint64_t					 Size,
-										const std::filesystem::path& DestinationPath)		 = 0;
-		virtual void				Delete(ParallelWork& Work, WorkerThreadPool& WorkerPool) = 0;
+										const std::filesystem::path& DestinationPath)							= 0;
+		virtual void				Touch(ParallelWork& Work, WorkerThreadPool& WorkerPool, const IoHash& Hash) = 0;
+		virtual void				Delete(ParallelWork& Work, WorkerThreadPool& WorkerPool)					= 0;
 	};
 
 	constexpr std::string_view FileHydratorPrefix = "file://";
@@ -227,6 +228,14 @@ namespace hydration_impl {
 				});
 		}
 
+		virtual void Touch(ParallelWork& Work, WorkerThreadPool& WorkerPool, const IoHash& Hash) override
+		{
+			// Local filesystem backend does not use modification-time-based retention, so no refresh is needed.
+			ZEN_UNUSED(Work);
+			ZEN_UNUSED(WorkerPool);
+			ZEN_UNUSED(Hash);
+		}
+
 		virtual void Delete(ParallelWork& Work, WorkerThreadPool& WorkerPool) override
 		{
 			ZEN_UNUSED(Work);
@@ -521,6 +530,22 @@ namespace hydration_impl {
 			}
 		}
 
+		virtual void Touch(ParallelWork& Work, WorkerThreadPool& WorkerPool, const IoHash& Hash) override
+		{
+			Work.ScheduleWork(WorkerPool, [this, Hash = IoHash(Hash)](std::atomic<bool>& AbortFlag) {
+				if (AbortFlag.load())
+				{
+					return;
+				}
+				std::string Key	   = m_KeyPrefix + "/cas/" + fmt::format("{}", Hash);
+				S3Result	Result = m_Client->Touch(Key);
+				if (!Result.IsSuccess())
+				{
+					throw zen::runtime_error("Failed to touch '{}' in S3: {}", Key, Result.Error);
+				}
+			});
+		}
+
 		virtual void Delete(ParallelWork& Work, WorkerThreadPool& WorkerPool) override
 		{
 			std::string			Prefix	   = m_KeyPrefix + "/";
@@ -781,6 +806,8 @@ IncrementalHydrator::Dehydrate(const CbObject& CachedState)
 
 		uint64_t UploadedFiles = 0;
 		uint64_t UploadedBytes = 0;
+		uint64_t TouchedFiles  = 0;
+		uint64_t TouchedBytes  = 0;
 		{
 			ParallelWork Work(*m_Threading.AbortFlag, *m_Threading.PauseFlag, WorkerThreadPool::EMode::DisableBacklog);
 
@@ -796,6 +823,14 @@ IncrementalHydrator::Dehydrate(const CbObject& CachedState)
 					UploadedFiles++;
 					UploadedBytes += CurrentEntry.Size;
 				}
+				else
+				{
+					// Refresh the backend's modification time so lifecycle-expiration policies
+					// do not evict CAS entries that are still referenced by this module.
+					m_Storage->Touch(Work, *m_Threading.WorkerPool, CurrentEntry.Hash);
+					TouchedFiles++;
+					TouchedBytes += CurrentEntry.Size;
+				}
 			}
 
 			Work.Wait();
@@ -840,16 +875,19 @@ IncrementalHydrator::Dehydrate(const CbObject& CachedState)
 		ZEN_DEBUG("Cleaning server state '{}'", m_Config.ServerStateDir);
 		CleanDirectory(*m_Threading.WorkerPool, *m_Threading.AbortFlag, *m_Threading.PauseFlag, ServerStateDir);
 
-		ZEN_INFO("Dehydration of module '{}' completed from folder '{}'. Hashed {} ({}). Uploaded {} ({}). Total {} ({}) in {}",
-				 m_Config.ModuleId,
-				 m_Config.ServerStateDir,
-				 HashedFiles,
-				 NiceBytes(HashedBytes),
-				 UploadedFiles,
-				 NiceBytes(UploadedBytes),
-				 TotalFiles,
-				 NiceBytes(TotalBytes),
-				 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+		ZEN_INFO(
+			"Dehydration of module '{}' completed from folder '{}'. Hashed {} ({}). Uploaded {} ({}). Touched {} ({}). Total {} ({}) in {}",
+			m_Config.ModuleId,
+			m_Config.ServerStateDir,
+			HashedFiles,
+			NiceBytes(HashedBytes),
+			UploadedFiles,
+			NiceBytes(UploadedBytes),
+			TouchedFiles,
+			NiceBytes(TouchedBytes),
+			TotalFiles,
+			NiceBytes(TotalBytes),
+			NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
 	}
 	catch (const std::exception& Ex)
 	{
diff --git a/src/zenutil/cloud/s3client.cpp b/src/zenutil/cloud/s3client.cpp
index 83238f5cc..3d6dca562 100644
--- a/src/zenutil/cloud/s3client.cpp
+++ b/src/zenutil/cloud/s3client.cpp
@@ -286,7 +286,11 @@ S3Client::GetSigningKey(std::string_view DateStamp)
 }
 
 HttpClient::KeyValueMap
-S3Client::SignRequest(std::string_view Method, std::string_view Path, std::string_view CanonicalQueryString, std::string_view PayloadHash)
+S3Client::SignRequest(std::string_view									   Method,
+					  std::string_view									   Path,
+					  std::string_view									   CanonicalQueryString,
+					  std::string_view									   PayloadHash,
+					  std::span<const std::pair<std::string, std::string>> ExtraSignedHeaders)
 {
 	SigV4Credentials Credentials = GetCurrentCredentials();
 
@@ -294,6 +298,7 @@ S3Client::SignRequest(std::string_view Method, std::string_view Path, std::strin
 
 	// Build sorted headers to sign (must be sorted by lowercase name)
 	std::vector<std::pair<std::string, std::string>> HeadersToSign;
+	HeadersToSign.reserve(4 + ExtraSignedHeaders.size());
 	HeadersToSign.emplace_back("host", m_Host);
 	HeadersToSign.emplace_back("x-amz-content-sha256", std::string(PayloadHash));
 	HeadersToSign.emplace_back("x-amz-date", AmzDate);
@@ -301,6 +306,10 @@ S3Client::SignRequest(std::string_view Method, std::string_view Path, std::strin
 	{
 		HeadersToSign.emplace_back("x-amz-security-token", Credentials.SessionToken);
 	}
+	for (const auto& [K, V] : ExtraSignedHeaders)
+	{
+		HeadersToSign.emplace_back(K, V);
+	}
 	std::sort(HeadersToSign.begin(), HeadersToSign.end());
 
 	std::string_view DateStamp(AmzDate.data(), 8);
@@ -317,6 +326,10 @@ S3Client::SignRequest(std::string_view Method, std::string_view Path, std::strin
 	{
 		Result->emplace("x-amz-security-token", Credentials.SessionToken);
 	}
+	for (const auto& [K, V] : ExtraSignedHeaders)
+	{
+		Result->emplace(K, V);
+	}
 
 	return Result;
 }
@@ -443,6 +456,47 @@ S3Client::DeleteObject(std::string_view Key)
 	return {};
 }
 
+S3Result
+S3Client::Touch(std::string_view Key)
+{
+	std::string Path = KeyToPath(Key);
+
+	// x-amz-copy-source is always "/bucket/key" regardless of addressing style.
+	// Key must be URI-encoded except for '/' separators. When source and destination
+	// are identical, REPLACE is required; COPY is rejected with InvalidRequest.
+	const std::array<std::pair<std::string, std::string>, 2> ExtraSigned{{
+		{"x-amz-copy-source", fmt::format("/{}/{}", m_BucketName, AwsUriEncode(Key, false))},
+		{"x-amz-metadata-directive", "REPLACE"},
+	}};
+
+	HttpClient::KeyValueMap Headers = SignRequest("PUT", Path, "", EmptyPayloadHash, ExtraSigned);
+
+	HttpClient::Response Response = m_HttpClient.Put(Path, IoBuffer{}, Headers);
+	if (!Response.IsSuccess())
+	{
+		std::string Err = Response.ErrorMessage("S3 Touch failed");
+		ZEN_WARN("S3 Touch '{}' failed: {}", Key, Err);
+		return S3Result{std::move(Err)};
+	}
+
+	// Copy operations can return HTTP 200 with an error in the XML body.
+	std::string_view ResponseBody = Response.AsText();
+	if (ResponseBody.find("<Error>") != std::string_view::npos)
+	{
+		std::string_view ErrorCode	  = ExtractXmlValue(ResponseBody, "Code");
+		std::string_view ErrorMessage = ExtractXmlValue(ResponseBody, "Message");
+		std::string		 Err		  = fmt::format("S3 Touch '{}' returned error: {} - {}", Key, ErrorCode, ErrorMessage);
+		ZEN_WARN("{}", Err);
+		return S3Result{std::move(Err)};
+	}
+
+	if (m_Verbose)
+	{
+		ZEN_INFO("S3 Touch '{}' succeeded", Key);
+	}
+	return {};
+}
+
 S3HeadObjectResult
 S3Client::HeadObject(std::string_view Key)
 {
@@ -983,6 +1037,42 @@ TEST_CASE("s3client.minio_integration")
 		CHECK(HeadRes2.Status == HeadObjectResult::NotFound);
 	}
 
+	// -- touch ----------------------------------------------------------------
+	{
+		std::string_view TestData = "touch-me"sv;
+		IoBuffer		 Content  = IoBufferBuilder::MakeFromMemory(MakeMemoryView(TestData));
+		S3Result		 PutRes	  = Client.PutObject("touch/obj.txt", std::move(Content));
+		REQUIRE(PutRes.IsSuccess());
+
+		S3HeadObjectResult Before = Client.HeadObject("touch/obj.txt");
+		REQUIRE(Before.IsSuccess());
+		REQUIRE(Before.Status == HeadObjectResult::Found);
+
+		// S3 LastModified has second precision; sleep past the second boundary so
+		// the touched timestamp is strictly greater.
+		Sleep(1100);
+
+		S3Result TouchRes = Client.Touch("touch/obj.txt");
+		REQUIRE(TouchRes.IsSuccess());
+
+		S3HeadObjectResult After = Client.HeadObject("touch/obj.txt");
+		REQUIRE(After.IsSuccess());
+		REQUIRE(After.Status == HeadObjectResult::Found);
+		CHECK(After.Info.Size == Before.Info.Size);
+		CHECK(After.Info.LastModified != Before.Info.LastModified);
+
+		// Content must be unchanged by a self-copy.
+		S3GetObjectResult GetRes = Client.GetObject("touch/obj.txt");
+		REQUIRE(GetRes.IsSuccess());
+		CHECK(GetRes.AsText() == TestData);
+
+		// Touching a missing key must fail.
+		S3Result MissRes = Client.Touch("touch/does-not-exist.txt");
+		CHECK_FALSE(MissRes.IsSuccess());
+
+		Client.DeleteObject("touch/obj.txt");
+	}
+
 	// -- head_not_found -------------------------------------------------------
 	{
 		S3HeadObjectResult Res = Client.HeadObject("nonexistent/key.dat");
diff --git a/src/zenutil/include/zenutil/cloud/s3client.h b/src/zenutil/include/zenutil/cloud/s3client.h
index b0402d231..4d72dd479 100644
--- a/src/zenutil/include/zenutil/cloud/s3client.h
+++ b/src/zenutil/include/zenutil/cloud/s3client.h
@@ -12,6 +12,7 @@
 #include <zencore/thread.h>
 
 #include <functional>
+#include <span>
 #include <string>
 #include <string_view>
 #include <vector>
@@ -129,6 +130,11 @@ public:
 	/// Delete an object from S3
 	S3Result DeleteObject(std::string_view Key);
 
+	/// Refresh an object's LastModified timestamp via a PUT Object - Copy onto itself
+	/// (x-amz-metadata-directive: REPLACE). Useful to reset lifecycle-expiration timers
+	/// without re-uploading the content.
+	S3Result Touch(std::string_view Key);
+
 	/// Check if an object exists and get its metadata
 	S3HeadObjectResult HeadObject(std::string_view Key);
 
@@ -198,11 +204,14 @@ private:
 	/// Build the bucket root path ("/" for virtual-hosted, "/bucket/" for path-style)
 	std::string BucketRootPath() const;
 
-	/// Sign a request and return headers with Authorization, x-amz-date, x-amz-content-sha256
-	HttpClient::KeyValueMap SignRequest(std::string_view Method,
-										std::string_view Path,
-										std::string_view QueryString,
-										std::string_view PayloadHash);
+	/// Sign a request and return headers with Authorization, x-amz-date, x-amz-content-sha256.
+	/// Additional x-amz-* headers that must participate in the signature are passed via
+	/// ExtraSignedHeaders (lowercase name, value); they are also copied into the returned map.
+	HttpClient::KeyValueMap SignRequest(std::string_view									 Method,
+										std::string_view									 Path,
+										std::string_view									 QueryString,
+										std::string_view									 PayloadHash,
+										std::span<const std::pair<std::string, std::string>> ExtraSignedHeaders = {});
 
 	/// Get or compute the signing key for the given date stamp, caching across requests on the same day
 	Sha256Digest GetSigningKey(std::string_view DateStamp);