revise oplog block arrangement (#842)

- Improvement: Fixed issue where oplog upload could create blocks larger than the max limit (64Mb)

Refactored remoteprojectstore.cpp to use ParallelWork and exceptions for error handling.

2 files changed, 58 insertions, 34 deletions

diff --git a/src/zenremotestore/include/zenremotestore/chunking/chunkblock.h b/src/zenremotestore/include/zenremotestore/chunking/chunkblock.h
index 931bb2097..e3a5f6539 100644
--- a/src/zenremotestore/include/zenremotestore/chunking/chunkblock.h
+++ b/src/zenremotestore/include/zenremotestore/chunking/chunkblock.h
@@ -31,7 +31,7 @@ ChunkBlockDescription			   ParseChunkBlockDescription(const CbObjectView& BlockO
 std::vector<ChunkBlockDescription> ParseBlockMetadatas(std::span<const CbObject> BlockMetadatas);
 CbObject						   BuildChunkBlockDescription(const ChunkBlockDescription& Block, CbObjectView MetaData);
 ChunkBlockDescription			   GetChunkBlockDescription(const SharedBuffer& BlockPayload, const IoHash& RawHash);
-typedef std::function<std::pair<uint64_t, CompressedBuffer>(const IoHash& RawHash)> FetchChunkFunc;
+typedef std::function<std::pair<uint64_t, CompositeBuffer>(const IoHash& RawHash)> FetchChunkFunc;
 
 CompressedBuffer	  GenerateChunkBlock(std::vector<std::pair<IoHash, FetchChunkFunc>>&& FetchChunks, ChunkBlockDescription& OutBlock);
 bool				  IterateChunkBlock(const SharedBuffer&															BlockPayload,
diff --git a/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h b/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
index 084d975a2..8df892053 100644
--- a/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
+++ b/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
@@ -150,27 +150,51 @@ struct RemoteStoreOptions
 	size_t ChunkFileSizeLimit = DefaultChunkFileSizeLimit;
 };
 
-typedef std::function<IoBuffer(const IoHash& AttachmentHash)> TGetAttachmentBufferFunc;
-
-RemoteProjectStore::LoadContainerResult BuildContainer(
-	CidStore&																	 ChunkStore,
-	ProjectStore::Project&														 Project,
-	ProjectStore::Oplog&														 Oplog,
-	WorkerThreadPool&															 WorkerPool,
-	size_t																		 MaxBlockSize,
-	size_t																		 MaxChunksPerBlock,
-	size_t																		 MaxChunkEmbedSize,
-	size_t																		 ChunkFileSizeLimit,
-	bool																		 BuildBlocks,
-	bool																		 IgnoreMissingAttachments,
-	bool																		 AllowChunking,
-	const std::function<void(CompressedBuffer&&, ChunkBlockDescription&&)>&		 AsyncOnBlock,
-	const std::function<void(const IoHash&, TGetAttachmentBufferFunc&&)>&		 OnLargeAttachment,
-	const std::function<void(std::vector<std::pair<IoHash, FetchChunkFunc>>&&)>& OnBlockChunks,
-	bool																		 EmbedLooseFiles);
+typedef std::function<CompositeBuffer(const IoHash& AttachmentHash)> TGetAttachmentBufferFunc;
+
+CbObject BuildContainer(CidStore&																	 ChunkStore,
+						ProjectStore::Project&														 Project,
+						ProjectStore::Oplog&														 Oplog,
+						WorkerThreadPool&															 WorkerPool,
+						size_t																		 MaxBlockSize,
+						size_t																		 MaxChunksPerBlock,
+						size_t																		 MaxChunkEmbedSize,
+						size_t																		 ChunkFileSizeLimit,
+						bool																		 BuildBlocks,
+						bool																		 IgnoreMissingAttachments,
+						bool																		 AllowChunking,
+						const std::function<void(CompressedBuffer&&, ChunkBlockDescription&&)>&		 AsyncOnBlock,
+						const std::function<void(const IoHash&, TGetAttachmentBufferFunc&&)>&		 OnLargeAttachment,
+						const std::function<void(std::vector<std::pair<IoHash, FetchChunkFunc>>&&)>& OnBlockChunks,
+						bool																		 EmbedLooseFiles);
 
 class JobContext;
 
+class RemoteStoreError : public std::runtime_error
+{
+public:
+	RemoteStoreError(const std::string& Message, int32_t ErrorCode, std::string_view Text)
+	: std::runtime_error(Message)
+	, m_ErrorCode(ErrorCode)
+	, m_Text(Text)
+	{
+	}
+
+	RemoteStoreError(const char* Message, int32_t ErrorCode, std::string_view Text)
+	: std::runtime_error(Message)
+	, m_ErrorCode(ErrorCode)
+	, m_Text(Text)
+	{
+	}
+
+	inline int32_t			GetErrorCode() const { return m_ErrorCode; }
+	inline std::string_view GetText() const { return m_Text; }
+
+private:
+	int32_t		m_ErrorCode = 0;
+	std::string m_Text;
+};
+
 RemoteProjectStore::Result SaveOplogContainer(
 	ProjectStore::Oplog&									Oplog,
 	const CbObject&											ContainerObject,
@@ -181,20 +205,20 @@ RemoteProjectStore::Result SaveOplogContainer(
 	const std::function<void(const ChunkedInfo& Chunked)>& OnChunkedAttachment,
 	JobContext*											   OptionalContext);
 
-RemoteProjectStore::Result SaveOplog(CidStore&				ChunkStore,
-									 RemoteProjectStore&	RemoteStore,
-									 ProjectStore::Project& Project,
-									 ProjectStore::Oplog&	Oplog,
-									 WorkerThreadPool&		NetworkWorkerPool,
-									 WorkerThreadPool&		WorkerPool,
-									 size_t					MaxBlockSize,
-									 size_t					MaxChunksPerBlock,
-									 size_t					MaxChunkEmbedSize,
-									 size_t					ChunkFileSizeLimit,
-									 bool					EmbedLooseFiles,
-									 bool					ForceUpload,
-									 bool					IgnoreMissingAttachments,
-									 JobContext*			OptionalContext);
+void SaveOplog(CidStore&			  ChunkStore,
+			   RemoteProjectStore&	  RemoteStore,
+			   ProjectStore::Project& Project,
+			   ProjectStore::Oplog&	  Oplog,
+			   WorkerThreadPool&	  NetworkWorkerPool,
+			   WorkerThreadPool&	  WorkerPool,
+			   size_t				  MaxBlockSize,
+			   size_t				  MaxChunksPerBlock,
+			   size_t				  MaxChunkEmbedSize,
+			   size_t				  ChunkFileSizeLimit,
+			   bool					  EmbedLooseFiles,
+			   bool					  ForceUpload,
+			   bool					  IgnoreMissingAttachments,
+			   JobContext*			  OptionalContext);
 
 struct LoadOplogContext
 {
@@ -218,7 +242,7 @@ struct LoadOplogContext
 	JobContext*					   OptionalJobContext			= nullptr;
 };
 
-RemoteProjectStore::Result LoadOplog(LoadOplogContext&& Context);
+void LoadOplog(LoadOplogContext&& Context);
 
 std::vector<IoHash>					   GetBlockHashesFromOplog(CbObjectView ContainerObject);
 std::vector<ThinChunkBlockDescription> GetBlocksFromOplog(CbObjectView ContainerObject, std::span<const IoHash> IncludeBlockHashes);