Merge branch 'main' into zcache-batch

1 files changed, 102 insertions, 93 deletions

diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index 168449d05..c0cd858b6 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -175,8 +175,10 @@ namespace detail {
 													  IoBuffer					 RecordValue,
 													  std::span<IoBuffer const>	 Payloads) override
 		{
+			using namespace fmt::literals;
+
 			ZEN_ASSERT(CacheRecord.PayloadIds.size() == Payloads.size());
-			const uint32_t MaxAttempts = 3;
+			const size_t MaxAttempts = 3;
 
 			try
 			{
@@ -204,117 +206,112 @@ namespace detail {
 				}
 				else
 				{
-					bool	Success				= false;
 					int64_t TotalBytes			= 0ull;
 					double	TotalElapsedSeconds = 0.0;
 
-					for (size_t Idx = 0, Count = Payloads.size(); Idx < Count; Idx++)
-					{
-						Success = false;
-						for (int32_t Attempt = 0; Attempt < MaxAttempts; Attempt++)
+					const auto PutBlobs = [&](std::span<IoHash> PayloadIds, std::string& OutReason) -> bool {
+						for (const IoHash& PayloadId : PayloadIds)
 						{
-							if (CloudCacheResult Result = Session.PutCompressedBlob(CacheRecord.PayloadIds[Idx], Payloads[Idx]);
-								Result.Success)
+							const auto It = std::find(std::begin(CacheRecord.PayloadIds), std::end(CacheRecord.PayloadIds), PayloadId);
+
+							if (It == std::end(CacheRecord.PayloadIds))
 							{
-								TotalBytes += Result.Bytes;
-								TotalElapsedSeconds += Result.ElapsedSeconds;
-								Success = true;
-								break;
+								OutReason = "payload '{}' MISSING from local cache"_format(PayloadId);
+								return false;
 							}
-						}
 
-						if (!Success)
-						{
-							return {.Reason			= "Failed to upload payload",
-									.Bytes			= TotalBytes,
-									.ElapsedSeconds = TotalElapsedSeconds,
-									.Success		= false};
+							const size_t Idx = std::distance(std::begin(CacheRecord.PayloadIds), It);
+
+							CloudCacheResult BlobResult;
+							for (size_t Attempt = 0; Attempt < MaxAttempts && !BlobResult.Success; Attempt++)
+							{
+								BlobResult = Session.PutCompressedBlob(CacheRecord.PayloadIds[Idx], Payloads[Idx]);
+							}
+
+							if (!BlobResult.Success)
+							{
+								OutReason = "upload payload '{}' FAILED, reason '{}'"_format(PayloadId, BlobResult.Reason);
+								return false;
+							}
+
+							TotalBytes += BlobResult.Bytes;
+							TotalElapsedSeconds += BlobResult.ElapsedSeconds;
 						}
+
+						return true;
+					};
+
+					PutRefResult RefResult;
+					for (int32_t Attempt = 0; Attempt < MaxAttempts && !RefResult.Success; Attempt++)
+					{
+						RefResult =
+							Session.PutRef(CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, RecordValue, ZenContentType::kCbObject);
 					}
 
-					Success = false;
-					for (int32_t Attempt = 0; Attempt < MaxAttempts; Attempt++)
+					if (!RefResult.Success)
 					{
-						if (PutRefResult Result = Session.PutRef(CacheRecord.CacheKey.Bucket,
-																 CacheRecord.CacheKey.Hash,
-																 RecordValue,
-																 ZenContentType::kCbObject);
-							Result.Success)
-						{
-							TotalBytes += Result.Bytes;
-							TotalElapsedSeconds += Result.ElapsedSeconds;
-							Success = true;
+						return {.Reason	 = "upload cache record '{}/{}' FAILED, reason '{}'"_format(CacheRecord.CacheKey.Bucket,
+																									CacheRecord.CacheKey.Hash,
+																									RefResult.Reason),
+								.Success = false};
+					}
 
-							if (!Result.Needs.empty())
-							{
-								for (const IoHash& NeededHash : Result.Needs)
-								{
-									Success = false;
-
-									if (auto It =
-											std::find(std::begin(CacheRecord.PayloadIds), std::end(CacheRecord.PayloadIds), NeededHash);
-										It != std::end(CacheRecord.PayloadIds))
-									{
-										const size_t Idx = It - std::begin(CacheRecord.PayloadIds);
-
-										if (CloudCacheResult BlobResult =
-												Session.PutCompressedBlob(CacheRecord.PayloadIds[Idx], Payloads[Idx]);
-											BlobResult.Success)
-										{
-											TotalBytes += BlobResult.Bytes;
-											TotalElapsedSeconds += BlobResult.ElapsedSeconds;
-											Success = true;
-										}
-										else
-										{
-											ZEN_WARN("upload missing payload '{}/{}/{}' FAILED",
-													 CacheRecord.CacheKey.Bucket,
-													 CacheRecord.CacheKey.Hash,
-													 NeededHash);
-										}
-									}
-									else
-									{
-										ZEN_WARN("needed payload '{}/{}/{}' MISSING",
-												 CacheRecord.CacheKey.Bucket,
-												 CacheRecord.CacheKey.Hash,
-												 NeededHash);
-									}
-								}
+					TotalBytes += RefResult.Bytes;
+					TotalElapsedSeconds += RefResult.ElapsedSeconds;
 
-								const IoHash RefHash = IoHash::HashBuffer(RecordValue);
+					std::string Reason;
+					if (!PutBlobs(RefResult.Needs, Reason))
+					{
+						return {.Reason = std::move(Reason), .Success = false};
+					}
 
-								if (FinalizeRefResult FinalizeResult =
-										Session.FinalizeRef(CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, RefHash);
-									FinalizeResult.Success)
-								{
-									TotalBytes += FinalizeResult.Bytes;
-									TotalElapsedSeconds += FinalizeResult.ElapsedSeconds;
-									Success = true;
-
-									for (const IoHash& MissingHash : FinalizeResult.Needs)
-									{
-										ZEN_WARN("finalize '{}/{}' FAILED, missing '{}'",
-												 CacheRecord.CacheKey.Bucket,
-												 CacheRecord.CacheKey.Hash,
-												 MissingHash);
-									}
-								}
-								else
-								{
-									ZEN_WARN("finalize '{}/{}' FAILED", CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash);
-									Success = false;
-								}
-							}
+					const IoHash	  RefHash		 = IoHash::HashBuffer(RecordValue);
+					FinalizeRefResult FinalizeResult = Session.FinalizeRef(CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, RefHash);
+
+					if (!FinalizeResult.Success)
+					{
+						return {.Reason	 = "finalize cache record '{}/{}' FAILED, reason '{}'"_format(CacheRecord.CacheKey.Bucket,
+																									  CacheRecord.CacheKey.Hash,
+																									  FinalizeResult.Reason),
+								.Success = false};
+					}
+
+					if (!FinalizeResult.Needs.empty())
+					{
+						if (!PutBlobs(FinalizeResult.Needs, Reason))
+						{
+							return {.Reason = std::move(Reason), .Success = false};
+						}
+
+						FinalizeResult = Session.FinalizeRef(CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, RefHash);
+
+						if (!FinalizeResult.Success)
+						{
+							return {.Reason	 = "finalize '{}/{}' FAILED, reason '{}'"_format(CacheRecord.CacheKey.Bucket,
+																							 CacheRecord.CacheKey.Hash,
+																							 FinalizeResult.Reason),
+									.Success = false};
+						}
 
-							if (Success)
+						if (!FinalizeResult.Needs.empty())
+						{
+							ExtendableStringBuilder<256> Sb;
+							for (const IoHash& MissingHash : FinalizeResult.Needs)
 							{
-								break;
+								Sb << MissingHash.ToHexString() << ",";
 							}
+
+							return {.Reason	 = "finalize '{}/{}' FAILED, still needs payload(s) '{}'"_format(CacheRecord.CacheKey.Bucket,
+																											 CacheRecord.CacheKey.Hash,
+																											 Sb.ToString()),
+									.Success = false};
 						}
 					}
 
-					return {.Bytes = TotalBytes, .ElapsedSeconds = TotalElapsedSeconds, .Success = Success};
+					TotalBytes += FinalizeResult.Bytes;
+					TotalElapsedSeconds += FinalizeResult.ElapsedSeconds;
+
+					return {.Bytes = TotalBytes, .ElapsedSeconds = TotalElapsedSeconds, .Success = true};
 				}
 			}
 			catch (std::exception& Err)
@@ -890,6 +887,15 @@ private:
 			{
 				const PutUpstreamCacheResult Result = Endpoint->PutCacheRecord(CacheRecord, CacheValue.Value, std::span(Payloads));
 				m_Stats.Add(m_Log, *Endpoint, Result, m_Endpoints);
+
+				if (!Result.Success)
+				{
+					ZEN_WARN("upload cache record '{}/{}' FAILED, endpoint '{}', reason '{}'",
+							 CacheRecord.CacheKey.Bucket,
+							 CacheRecord.CacheKey.Hash,
+							 Endpoint->DisplayName(),
+							 Result.Reason);
+				}
 			}
 		}
 	}
@@ -907,7 +913,10 @@ private:
 				}
 				catch (std::exception& e)
 				{
-					ZEN_WARN("process upstream ({}/{}) FAILED '{}'", CacheRecord.CacheKey.Bucket, CacheRecord.CacheKey.Hash, e.what());
+					ZEN_WARN("upload cache record '{}/{}' FAILED, reason '{}'",
+							 CacheRecord.CacheKey.Bucket,
+							 CacheRecord.CacheKey.Hash,
+							 e.what());
 				}
 			}