use partial blocks for oplog import (#780)

Feature: Add --allow-partial-block-requests to zen oplog-import
Improvement: zen oplog-import now uses partial block requests to reduce download size
Improvement: Use latency to Cloud Storage host and Zen Cache host when calculating partial block requests

1 files changed, 27 insertions, 6 deletions

diff --git a/src/zenserver/storage/projectstore/httpprojectstore.cpp b/src/zenserver/storage/projectstore/httpprojectstore.cpp
index 416e2ed69..2b5474d00 100644
--- a/src/zenserver/storage/projectstore/httpprojectstore.cpp
+++ b/src/zenserver/storage/projectstore/httpprojectstore.cpp
@@ -244,6 +244,8 @@ namespace {
 	{
 		std::shared_ptr<RemoteProjectStore> Store;
 		std::string							Description;
+		double								HostLatencySec	= -1.0;
+		double								CacheLatencySec = -1.0;
 	};
 
 	CreateRemoteStoreResult CreateRemoteStore(LoggerRef					   InLog,
@@ -261,6 +263,8 @@ namespace {
 		using namespace std::literals;
 
 		std::shared_ptr<RemoteProjectStore> RemoteStore;
+		double								HostLatencySec	= -1.0;
+		double								CacheLatencySec = -1.0;
 
 		if (CbObjectView File = Params["file"sv].AsObjectView(); File)
 		{
@@ -495,7 +499,9 @@ namespace {
 														 /*Quiet*/ false,
 														 /*Unattended*/ false,
 														 /*Hidden*/ true,
-														 GetTinyWorkerPool(EWorkloadType::Background));
+														 GetTinyWorkerPool(EWorkloadType::Background),
+														 HostLatencySec,
+														 CacheLatencySec);
 		}
 
 		if (!RemoteStore)
@@ -503,7 +509,10 @@ namespace {
 			return {nullptr, "Unknown remote store type"};
 		}
 
-		return {std::move(RemoteStore), ""};
+		return CreateRemoteStoreResult{.Store			= std::move(RemoteStore),
+									   .Description		= "",
+									   .HostLatencySec	= HostLatencySec,
+									   .CacheLatencySec = CacheLatencySec};
 	}
 
 	std::pair<HttpResponseCode, std::string> ConvertResult(const RemoteProjectStore::Result& Result)
@@ -2356,15 +2365,19 @@ HttpProjectService::HandleOplogSaveRequest(HttpRouterRequest& Req)
 		tsl::robin_set<IoHash, IoHash::Hasher> Attachments;
 
 		auto HasAttachment = [this](const IoHash& RawHash) { return m_CidStore.ContainsChunk(RawHash); };
-		auto OnNeedBlock   = [&AttachmentsLock, &Attachments](const IoHash& BlockHash, const std::vector<IoHash>&& ChunkHashes) {
+		auto OnNeedBlock   = [&AttachmentsLock, &Attachments](ThinChunkBlockDescription&& ThinBlockDescription,
+															  std::vector<uint32_t>&&	  NeededChunkIndexes) {
 			  RwLock::ExclusiveLockScope _(AttachmentsLock);
-			  if (BlockHash != IoHash::Zero)
+			  if (ThinBlockDescription.BlockHash != IoHash::Zero)
 			  {
-				  Attachments.insert(BlockHash);
+				  Attachments.insert(ThinBlockDescription.BlockHash);
 			  }
 			  else
 			  {
-				  Attachments.insert(ChunkHashes.begin(), ChunkHashes.end());
+				  for (uint32_t ChunkIndex : NeededChunkIndexes)
+				  {
+					  Attachments.insert(ThinBlockDescription.ChunkRawHashes[ChunkIndex]);
+				  }
 			  }
 		};
 		auto OnNeedAttachment = [&AttachmentsLock, &Attachments](const IoHash& RawHash) {
@@ -2663,6 +2676,8 @@ HttpProjectService::HandleRpcRequest(HttpRouterRequest& Req)
 				bool		 CleanOplog				  = Params["clean"].AsBool(false);
 				bool		 BoostWorkerCount		  = Params["boostworkercount"].AsBool(false);
 				bool		 BoostWorkerMemory		  = Params["boostworkermemory"sv].AsBool(false);
+				EPartialBlockRequestMode PartialBlockRequestMode =
+					PartialBlockRequestModeFromString(Params["partialblockrequestmode"sv].AsString("true"));
 
 				CreateRemoteStoreResult RemoteStoreResult = CreateRemoteStore(Log(),
 																			  Params,
@@ -2688,6 +2703,9 @@ HttpProjectService::HandleRpcRequest(HttpRouterRequest& Req)
 					 Force,
 					 IgnoreMissingAttachments,
 					 CleanOplog,
+					 PartialBlockRequestMode,
+					 HostLatencySec	 = RemoteStoreResult.HostLatencySec,
+					 CacheLatencySec = RemoteStoreResult.CacheLatencySec,
 					 BoostWorkerCount](JobContext& Context) {
 						Context.ReportMessage(fmt::format("Loading oplog '{}/{}' from {}",
 														  Oplog->GetOuterProjectIdentifier(),
@@ -2709,6 +2727,9 @@ HttpProjectService::HandleRpcRequest(HttpRouterRequest& Req)
 																		Force,
 																		IgnoreMissingAttachments,
 																		CleanOplog,
+																		PartialBlockRequestMode,
+																		HostLatencySec,
+																		CacheLatencySec,
 																		&Context);
 						auto					   Response = ConvertResult(Result);
 						ZEN_INFO("LoadOplog: Status: {} '{}'", ToString(Response.first), Response.second);