stream oplog attachments from jupiter (#384)

* stream large downloads from jupiter to temporary file
* rework DeleteOnClose - top level marks file for delete and if lower level parts wants to keep it it clears that flag
* changelog
* log number of attachments to download
* add delay on jupiter request failure when retrying
* make sure we upload all attachments even if Needs are empty when ForceUpload is true
release TempAttachment as soon as it is used
* sort attachments so we get predictable blocks for the same oplog

5 files changed, 119 insertions, 73 deletions

diff --git a/src/zenserver/projectstore/httpprojectstore.cpp b/src/zenserver/projectstore/httpprojectstore.cpp
index d28d49d1f..ff7a5f8b9 100644
--- a/src/zenserver/projectstore/httpprojectstore.cpp
+++ b/src/zenserver/projectstore/httpprojectstore.cpp
@@ -973,6 +973,7 @@ HttpProjectService::HandleOplogOpNewRequest(HttpRouterRequest& Req)
 		std::filesystem::path AttachmentPath = Oplog.TempPath() / AttachmentId.ToHexString();
 		if (IoBuffer Data = IoBufferBuilder::MakeFromTemporaryFile(AttachmentPath))
 		{
+			Data.SetDeleteOnClose(true);
 			return SharedBuffer(std::move(Data));
 		}
 		else
diff --git a/src/zenserver/projectstore/jupiterremoteprojectstore.cpp b/src/zenserver/projectstore/jupiterremoteprojectstore.cpp
index 4be58256c..2bfa6851b 100644
--- a/src/zenserver/projectstore/jupiterremoteprojectstore.cpp
+++ b/src/zenserver/projectstore/jupiterremoteprojectstore.cpp
@@ -15,16 +15,18 @@ using namespace std::literals;
 class JupiterRemoteStore : public RemoteProjectStore
 {
 public:
-	JupiterRemoteStore(Ref<CloudCacheClient>&& CloudClient,
-					   std::string_view		   Namespace,
-					   std::string_view		   Bucket,
-					   const IoHash&		   Key,
-					   bool					   ForceDisableBlocks,
-					   bool					   ForceDisableTempBlocks)
+	JupiterRemoteStore(Ref<CloudCacheClient>&&		CloudClient,
+					   std::string_view				Namespace,
+					   std::string_view				Bucket,
+					   const IoHash&				Key,
+					   bool							ForceDisableBlocks,
+					   bool							ForceDisableTempBlocks,
+					   const std::filesystem::path& TempFilePath)
 	: m_CloudClient(std::move(CloudClient))
 	, m_Namespace(Namespace)
 	, m_Bucket(Bucket)
 	, m_Key(Key)
+	, m_TempFilePath(TempFilePath)
 	{
 		if (ForceDisableBlocks)
 		{
@@ -52,6 +54,10 @@ public:
 			for (int32_t Attempt = 0; Attempt < MaxAttempts && !PutResult.Success; Attempt++)
 			{
 				PutResult = Session.PutRef(m_Namespace, m_Bucket, m_Key, Payload, ZenContentType::kCbObject);
+				if (!PutResult.Success)
+				{
+					Sleep(100 * (Attempt + 1));
+				}
 			}
 		}
 
@@ -77,6 +83,10 @@ public:
 			for (int32_t Attempt = 0; Attempt < MaxAttempts && !PutResult.Success; Attempt++)
 			{
 				PutResult = Session.PutCompressedBlob(m_Namespace, RawHash, Payload);
+				if (!PutResult.Success)
+				{
+					Sleep(100 * (Attempt + 1));
+				}
 			}
 		}
 
@@ -116,6 +126,10 @@ public:
 			for (int32_t Attempt = 0; Attempt < MaxAttempts && !FinalizeResult.Success; Attempt++)
 			{
 				FinalizeResult = Session.FinalizeRef(m_Namespace, m_Bucket, m_Key, RawHash);
+				if (!FinalizeResult.Success)
+				{
+					Sleep(100 * (Attempt + 1));
+				}
 			}
 		}
 		Result Result{ConvertResult(FinalizeResult)};
@@ -140,6 +154,10 @@ public:
 			for (int32_t Attempt = 0; Attempt < MaxAttempts && !GetResult.Success; Attempt++)
 			{
 				GetResult = Session.GetRef(m_Namespace, m_Bucket, m_Key, ZenContentType::kCbObject);
+				if (!GetResult.Success)
+				{
+					Sleep(100 * (Attempt + 1));
+				}
 			}
 		}
 
@@ -179,7 +197,11 @@ public:
 			CloudCacheSession Session(m_CloudClient.Get());
 			for (int32_t Attempt = 0; Attempt < MaxAttempts && !GetResult.Success; Attempt++)
 			{
-				GetResult = Session.GetCompressedBlob(m_Namespace, RawHash);
+				GetResult = Session.GetCompressedBlob(m_Namespace, RawHash, m_TempFilePath);
+				if (!GetResult.Success)
+				{
+					Sleep(100 * (Attempt + 1));
+				}
 			}
 		}
 		LoadAttachmentResult Result{ConvertResult(GetResult), std::move(GetResult.Response)};
@@ -236,12 +258,13 @@ private:
 	const std::string	  m_Namespace;
 	const std::string	  m_Bucket;
 	const IoHash		  m_Key;
+	std::filesystem::path m_TempFilePath;
 	bool				  m_EnableBlocks  = true;
 	bool				  m_UseTempBlocks = true;
 };
 
 std::unique_ptr<RemoteProjectStore>
-CreateJupiterRemoteStore(const JupiterRemoteStoreOptions& Options)
+CreateJupiterRemoteStore(const JupiterRemoteStoreOptions& Options, const std::filesystem::path& TempFilePath)
 {
 	std::string Url = Options.Url;
 	if (Url.find("://"sv) == std::string::npos)
@@ -281,7 +304,8 @@ CreateJupiterRemoteStore(const JupiterRemoteStoreOptions& Options)
 																						   Options.Bucket,
 																						   Options.Key,
 																						   Options.ForceDisableBlocks,
-																						   Options.ForceDisableTempBlocks);
+																						   Options.ForceDisableTempBlocks,
+																						   TempFilePath);
 	return RemoteStore;
 }
 
diff --git a/src/zenserver/projectstore/jupiterremoteprojectstore.h b/src/zenserver/projectstore/jupiterremoteprojectstore.h
index f2375a730..d6fced91b 100644
--- a/src/zenserver/projectstore/jupiterremoteprojectstore.h
+++ b/src/zenserver/projectstore/jupiterremoteprojectstore.h
@@ -22,6 +22,7 @@ struct JupiterRemoteStoreOptions : RemoteStoreOptions
 	bool		AssumeHttp2			   = false;
 };
 
-std::unique_ptr<RemoteProjectStore> CreateJupiterRemoteStore(const JupiterRemoteStoreOptions& Options);
+std::unique_ptr<RemoteProjectStore> CreateJupiterRemoteStore(const JupiterRemoteStoreOptions& Options,
+															 const std::filesystem::path&	  TempFilePath);
 
 }  // namespace zen
diff --git a/src/zenserver/projectstore/projectstore.cpp b/src/zenserver/projectstore/projectstore.cpp
index e42afdb67..7754f61cd 100644
--- a/src/zenserver/projectstore/projectstore.cpp
+++ b/src/zenserver/projectstore/projectstore.cpp
@@ -72,10 +72,11 @@ namespace {
 		} while (true);
 	}
 
-	std::pair<std::unique_ptr<RemoteProjectStore>, std::string> CreateRemoteStore(CbObjectView Params,
-																				  AuthMgr&	   AuthManager,
-																				  size_t	   MaxBlockSize,
-																				  size_t	   MaxChunkEmbedSize)
+	std::pair<std::unique_ptr<RemoteProjectStore>, std::string> CreateRemoteStore(CbObjectView				   Params,
+																				  AuthMgr&					   AuthManager,
+																				  size_t					   MaxBlockSize,
+																				  size_t					   MaxChunkEmbedSize,
+																				  const std::filesystem::path& TempFilePath)
 	{
 		using namespace std::literals;
 
@@ -162,7 +163,7 @@ namespace {
 												 ForceDisableBlocks,
 												 ForceDisableTempBlocks,
 												 AssumeHttp2};
-			RemoteStore						  = CreateJupiterRemoteStore(Options);
+			RemoteStore						  = CreateJupiterRemoteStore(Options, TempFilePath);
 		}
 
 		if (CbObjectView Zen = Params["zen"sv].AsObjectView(); Zen)
@@ -2700,7 +2701,7 @@ ProjectStore::Export(ProjectStore::Project& Project, ProjectStore::Oplog& Oplog,
 	bool   EmbedLooseFile	 = Params["embedloosefiles"sv].AsBool(false);
 
 	std::pair<std::unique_ptr<RemoteProjectStore>, std::string> RemoteStoreResult =
-		CreateRemoteStore(Params, AuthManager, MaxBlockSize, MaxChunkEmbedSize);
+		CreateRemoteStore(Params, AuthManager, MaxBlockSize, MaxChunkEmbedSize, Oplog.TempPath());
 
 	if (RemoteStoreResult.first == nullptr)
 	{
@@ -2742,7 +2743,7 @@ ProjectStore::Import(ProjectStore::Project& Project, ProjectStore::Oplog& Oplog,
 	bool   Force			 = Params["force"sv].AsBool(false);
 
 	std::pair<std::unique_ptr<RemoteProjectStore>, std::string> RemoteStoreResult =
-		CreateRemoteStore(Params, AuthManager, MaxBlockSize, MaxChunkEmbedSize);
+		CreateRemoteStore(Params, AuthManager, MaxBlockSize, MaxChunkEmbedSize, Oplog.TempPath());
 
 	if (RemoteStoreResult.first == nullptr)
 	{
diff --git a/src/zenserver/projectstore/remoteprojectstore.cpp b/src/zenserver/projectstore/remoteprojectstore.cpp
index bbf3a9f32..f10d7da63 100644
--- a/src/zenserver/projectstore/remoteprojectstore.cpp
+++ b/src/zenserver/projectstore/remoteprojectstore.cpp
@@ -298,9 +298,9 @@ BuildContainer(CidStore&																	ChunkStore,
 										Offset += Buffer.GetSize();
 									}
 									void* FileHandle = BlockFile.Detach();
-									AttachmentBuffer = IoBuffer(IoBuffer::File, FileHandle, 0, Offset);
+									AttachmentBuffer = IoBuffer(IoBuffer::File, FileHandle, 0, Offset, /*IsWholeFile*/ true);
 
-									AttachmentBuffer.MarkAsDeleteOnClose();
+									AttachmentBuffer.SetDeleteOnClose(true);
 									ZEN_DEBUG("Saved temp attachment {}, {}", DataHash, NiceBytes(PayloadSize));
 								}
 								OutLooseAttachments->insert_or_assign(DataHash, AttachmentBuffer);
@@ -389,13 +389,24 @@ BuildContainer(CidStore&																	ChunkStore,
 			return ChunkStore.FindChunkByCid(AttachmentHash);
 		};
 
-		for (const auto& It : Attachments)
+		// Sort attachments so we get predictable blocks for the same oplog upload
+		std::vector<IoHash> SortedAttachments;
+		SortedAttachments.reserve(Attachments.size());
+		for (const auto It : Attachments)
 		{
-			const IoHash& AttachmentHash(It.first);
-			IoBuffer	  Payload = GetPayload(AttachmentHash);
+			SortedAttachments.push_back(It.first);
+		}
+		std::sort(SortedAttachments.begin(), SortedAttachments.end());
+
+		for (const IoHash& AttachmentHash : SortedAttachments)
+		{
+			IoBuffer Payload = GetPayload(AttachmentHash);
 			if (!Payload)
 			{
-				std::optional<CbObject> Op = Oplog.GetOpByIndex(It.second);
+				auto It = Attachments.find(AttachmentHash);
+				ZEN_ASSERT(It != Attachments.end());
+
+				std::optional<CbObject> Op = Oplog.GetOpByIndex(It->second);
 				ZEN_ASSERT(Op.has_value());
 				ExtendableStringBuilder<1024> Sb;
 				Sb.Append("Failed to find attachment '");
@@ -672,7 +683,7 @@ SaveOplog(CidStore&				 ChunkStore,
 					Offset += Buffer.GetSize();
 				}
 				void* FileHandle = BlockFile.Detach();
-				BlockBuffer		 = IoBuffer(IoBuffer::File, FileHandle, 0, Offset);
+				BlockBuffer		 = IoBuffer(IoBuffer::File, FileHandle, 0, Offset, /*IsWholeFile*/ true);
 			}
 			catch (std::exception& Ex)
 			{
@@ -682,7 +693,7 @@ SaveOplog(CidStore&				 ChunkStore,
 				return;
 			}
 
-			BlockBuffer.MarkAsDeleteOnClose();
+			BlockBuffer.SetDeleteOnClose(true);
 			{
 				RwLock::ExclusiveLockScope __(AttachmentsLock);
 				CreatedBlocks.insert({BlockHash, std::move(BlockBuffer)});
@@ -757,7 +768,7 @@ SaveOplog(CidStore&				 ChunkStore,
 			ZEN_DEBUG("Saved container in {}", NiceTimeSpanMs(static_cast<uint64_t>(ContainerSaveResult.ElapsedSeconds * 1000)));
 		}
 
-		if (!ContainerSaveResult.Needs.empty())
+		if (!ContainerSaveResult.Needs.empty() || ForceUpload)
 		{
 			ZEN_INFO("Filtering needed attachments...");
 			std::vector<IoHash>						   NeededLargeAttachments;
@@ -766,6 +777,7 @@ SaveOplog(CidStore&				 ChunkStore,
 			NeededOtherAttachments.reserve(CreatedBlocks.size());
 			if (ForceUpload)
 			{
+				// TODO: Check ForceUpload - it should add all attachments and blocks regardless if Needs is empty or not
 				NeededLargeAttachments.insert(NeededLargeAttachments.end(), LargeAttachments.begin(), LargeAttachments.end());
 			}
 			else
@@ -791,59 +803,63 @@ SaveOplog(CidStore&				 ChunkStore,
 					{
 						break;
 					}
-					SaveAttachmentsLatch.AddCount(1);
-					WorkerPool.ScheduleWork(
-						[&ChunkStore, &RemoteStore, &SaveAttachmentsLatch, &RemoteResult, RawHash, &CreatedBlocks, &TempAttachments]() {
-							auto _ = MakeGuard([&SaveAttachmentsLatch] { SaveAttachmentsLatch.CountDown(); });
-							if (RemoteResult.IsError())
-							{
-								return;
-							}
+					IoBuffer Payload;
+					if (auto BlockIt = CreatedBlocks.find(RawHash); BlockIt != CreatedBlocks.end())
+					{
+						Payload = std::move(BlockIt->second);
+					}
+					else if (auto LooseTmpFileIt = TempAttachments.find(RawHash); LooseTmpFileIt != TempAttachments.end())
+					{
+						Payload = LooseTmpFileIt->second;
+						TempAttachments.erase(LooseTmpFileIt);
+					}
 
-							IoBuffer Payload;
-							if (auto BlockIt = CreatedBlocks.find(RawHash); BlockIt != CreatedBlocks.end())
-							{
-								Payload = std::move(BlockIt->second);
-							}
-							else if (auto LooseTmpFileIt = TempAttachments.find(RawHash); LooseTmpFileIt != TempAttachments.end())
-							{
-								Payload = std::move(LooseTmpFileIt->second);
-							}
-							else
-							{
-								Payload = ChunkStore.FindChunkByCid(RawHash);
-							}
-							if (!Payload)
-							{
-								RemoteResult.SetError(gsl::narrow<int>(HttpResponseCode::NotFound),
-													  fmt::format("Failed to find attachment {}", RawHash),
-													  {});
-								ZEN_ERROR("Failed to build container ({}). Reason: '{}'",
-										  RemoteResult.GetErrorReason(),
-										  RemoteResult.GetError());
-								return;
-							}
+					SaveAttachmentsLatch.AddCount(1);
+					WorkerPool.ScheduleWork([&ChunkStore,
+											 &RemoteStore,
+											 &SaveAttachmentsLatch,
+											 &RemoteResult,
+											 RawHash,
+											 &CreatedBlocks,
+											 TempPayload = std::move(Payload)]() {
+						auto _ = MakeGuard([&SaveAttachmentsLatch] { SaveAttachmentsLatch.CountDown(); });
+						if (RemoteResult.IsError())
+						{
+							return;
+						}
+						IoBuffer Payload = TempPayload ? TempPayload : ChunkStore.FindChunkByCid(RawHash);
+						if (!Payload)
+						{
+							RemoteResult.SetError(gsl::narrow<int>(HttpResponseCode::NotFound),
+												  fmt::format("Failed to find attachment {}", RawHash),
+												  {});
+							ZEN_ERROR("Failed to build container ({}). Reason: '{}'",
+									  RemoteResult.GetErrorReason(),
+									  RemoteResult.GetError());
+							return;
+						}
 
-							RemoteProjectStore::SaveAttachmentResult Result =
-								RemoteStore.SaveAttachment(CompositeBuffer(SharedBuffer(Payload)), RawHash);
-							if (Result.ErrorCode)
-							{
-								RemoteResult.SetError(Result.ErrorCode, Result.Reason, Result.Text);
-								ZEN_ERROR("Failed to save attachment '{}', {} ({}). Reason: '{}'",
-										  RawHash,
-										  NiceBytes(Payload.GetSize()),
-										  RemoteResult.GetError(),
-										  RemoteResult.GetErrorReason());
-								return;
-							}
-							ZEN_DEBUG("Saved attachment {}, {} in {}",
+						RemoteProjectStore::SaveAttachmentResult Result =
+							RemoteStore.SaveAttachment(CompositeBuffer(SharedBuffer(Payload)), RawHash);
+						if (Result.ErrorCode)
+						{
+							RemoteResult.SetError(Result.ErrorCode, Result.Reason, Result.Text);
+							ZEN_ERROR("Failed to save attachment '{}', {} ({}). Reason: '{}'",
 									  RawHash,
 									  NiceBytes(Payload.GetSize()),
-									  NiceTimeSpanMs(static_cast<uint64_t>(Result.ElapsedSeconds * 1000)));
+									  RemoteResult.GetError(),
+									  RemoteResult.GetErrorReason());
 							return;
-						});
+						}
+						ZEN_DEBUG("Saved attachment {}, {} in {}",
+								  RawHash,
+								  NiceBytes(Payload.GetSize()),
+								  NiceTimeSpanMs(static_cast<uint64_t>(Result.ElapsedSeconds * 1000)));
+						return;
+					});
 				}
 			}
+			TempAttachments.clear();
 
 			if (!CreatedBlocks.empty())
 			{
@@ -967,7 +983,6 @@ SaveOplog(CidStore&				 ChunkStore,
 				ZEN_INFO("Saving attachments, {} remaining...", SaveAttachmentsLatch.Remaining());
 			}
 			SaveAttachmentsLatch.Wait();
-			TempAttachments.clear();
 		}
 
 		if (!RemoteResult.IsError())
@@ -1224,6 +1239,10 @@ LoadOplog(CidStore& ChunkStore, RemoteProjectStore& RemoteStore, ProjectStore::O
 
 	RemoteProjectStore::Result Result =
 		SaveOplogContainer(Oplog, LoadContainerResult.ContainerObject, HasAttachment, OnNeedBlock, OnNeedAttachment);
+	if (!Attachments.empty())
+	{
+		ZEN_INFO("Found {} attachments to download", Attachments.size());
+	}
 
 	AttachmentsWorkLatch.CountDown();
 	while (!AttachmentsWorkLatch.Wait(1000))