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

3 files changed, 23 insertions, 20 deletions

diff --git a/src/zencore/filesystem.cpp b/src/zencore/filesystem.cpp
index e17d83895..3311ba1b9 100644
--- a/src/zencore/filesystem.cpp
+++ b/src/zencore/filesystem.cpp
@@ -758,7 +758,7 @@ ReadFile(std::filesystem::path Path)
 #endif
 
 	FileContents Contents;
-	Contents.Data.emplace_back(IoBuffer(IoBuffer::File, Handle, 0, FileSizeBytes));
+	Contents.Data.emplace_back(IoBuffer(IoBuffer::File, Handle, 0, FileSizeBytes, /*IsWholeFile*/ true));
 	return Contents;
 }
 
diff --git a/src/zencore/include/zencore/iobuffer.h b/src/zencore/include/zencore/iobuffer.h
index bbc346f9b..fef78741f 100644
--- a/src/zencore/include/zencore/iobuffer.h
+++ b/src/zencore/include/zencore/iobuffer.h
@@ -270,20 +270,20 @@ struct IoBufferExtendedCore : public IoBufferCore
 
 	enum ExtendedFlags
 	{
-		kOwnsFile = 1 << 16,
-		kOwnsMmap = 1 << 17
+		kOwnsFile	   = 1 << 16,
+		kOwnsMmap	   = 1 << 17,
+		kDeleteOnClose = 1 << 18
 	};
 
 	void Materialize() const;
 	bool GetFileReference(IoBufferFileReference& OutRef) const;
-	void MarkAsDeleteOnClose();
+	void SetDeleteOnClose(bool DeleteOnClose);
 
 private:
 	void*		  m_FileHandle	  = nullptr;
 	uint64_t	  m_FileOffset	  = 0;
 	mutable void* m_MmapHandle	  = nullptr;
 	mutable void* m_MappedPointer = nullptr;
-	bool		  m_DeleteOnClose = false;
 };
 
 inline IoBufferExtendedCore*
@@ -362,7 +362,7 @@ public:
 		memcpy(const_cast<void*>(m_Core->DataPointer()), DataPtr, SizeBytes);
 	}
 
-	ZENCORE_API IoBuffer(EFileTag, void* FileHandle, uint64_t ChunkFileOffset, uint64_t ChunkSize);
+	ZENCORE_API IoBuffer(EFileTag, void* FileHandle, uint64_t ChunkFileOffset, uint64_t ChunkSize, bool IsWholeFile);
 	ZENCORE_API IoBuffer(EBorrowedFileTag, void* FileHandle, uint64_t ChunkFileOffset, uint64_t ChunkSize);
 
 	inline explicit						operator bool() const { return !m_Core->IsNull(); }
@@ -379,7 +379,7 @@ public:
 	inline void							SetContentType(ZenContentType ContentType) { m_Core->SetContentType(ContentType); }
 	[[nodiscard]] inline ZenContentType GetContentType() const { return m_Core->GetContentType(); }
 	[[nodiscard]] ZENCORE_API bool		GetFileReference(IoBufferFileReference& OutRef) const;
-	void								MarkAsDeleteOnClose();
+	void								SetDeleteOnClose(bool DeleteOnClose);
 
 	inline MemoryView		 GetView() const { return MemoryView(m_Core->DataPointer(), m_Core->DataBytes()); }
 	inline MutableMemoryView GetMutableView() { return MutableMemoryView(m_Core->MutableDataPointer(), m_Core->DataBytes()); }
diff --git a/src/zencore/iobuffer.cpp b/src/zencore/iobuffer.cpp
index 22bc61395..efec06f7f 100644
--- a/src/zencore/iobuffer.cpp
+++ b/src/zencore/iobuffer.cpp
@@ -217,7 +217,7 @@ IoBufferExtendedCore::~IoBufferExtendedCore()
 
 	if (LocalFlags & kOwnsFile)
 	{
-		if (m_DeleteOnClose)
+		if (LocalFlags & kDeleteOnClose)
 		{
 #if ZEN_PLATFORM_WINDOWS
 			// Mark file for deletion when final handle is closed
@@ -438,9 +438,16 @@ IoBufferExtendedCore::GetFileReference(IoBufferFileReference& OutRef) const
 }
 
 void
-IoBufferExtendedCore::MarkAsDeleteOnClose()
+IoBufferExtendedCore::SetDeleteOnClose(bool DeleteOnClose)
 {
-	m_DeleteOnClose = true;
+	if (DeleteOnClose && (m_Flags & kOwnsFile))
+	{
+		m_Flags.fetch_or(kDeleteOnClose, std::memory_order_release);
+	}
+	else
+	{
+		m_Flags.fetch_and(~static_cast<uint32_t>(kDeleteOnClose), std::memory_order_release);
+	}
 }
 
 //////////////////////////////////////////////////////////////////////////
@@ -475,9 +482,10 @@ IoBuffer::IoBuffer(const IoBuffer& OuterBuffer, size_t Offset, size_t Size)
 	}
 }
 
-IoBuffer::IoBuffer(EFileTag, void* FileHandle, uint64_t ChunkFileOffset, uint64_t ChunkSize)
+IoBuffer::IoBuffer(EFileTag, void* FileHandle, uint64_t ChunkFileOffset, uint64_t ChunkSize, bool IsWholeFile)
 : m_Core(new IoBufferExtendedCore(FileHandle, ChunkFileOffset, ChunkSize, /* owned */ true))
 {
+	m_Core->SetIsWholeFile(IsWholeFile);
 }
 
 IoBuffer::IoBuffer(EBorrowedFileTag, void* FileHandle, uint64_t ChunkFileOffset, uint64_t ChunkSize)
@@ -506,11 +514,11 @@ IoBuffer::GetFileReference(IoBufferFileReference& OutRef) const
 }
 
 void
-IoBuffer::MarkAsDeleteOnClose()
+IoBuffer::SetDeleteOnClose(bool DeleteOnClose)
 {
 	if (IoBufferExtendedCore* ExtCore = m_Core->ExtendedCore())
 	{
-		ExtCore->MarkAsDeleteOnClose();
+		ExtCore->SetDeleteOnClose(DeleteOnClose);
 	}
 }
 
@@ -615,9 +623,7 @@ IoBufferBuilder::MakeFromFile(const std::filesystem::path& FileName, uint64_t Of
 #if ZEN_PLATFORM_WINDOWS
 		void* Fd = DataFile.Detach();
 #endif
-		IoBuffer Iob(IoBuffer::File, (void*)uintptr_t(Fd), Offset, Size);
-		Iob.m_Core->SetIsWholeFile(Offset == 0 && Size == FileSize);
-		return Iob;
+		return IoBuffer(IoBuffer::File, (void*)uintptr_t(Fd), Offset, Size, Offset == 0 && Size == FileSize);
 	}
 
 #if !ZEN_PLATFORM_WINDOWS
@@ -666,10 +672,7 @@ IoBufferBuilder::MakeFromTemporaryFile(const std::filesystem::path& FileName)
 	Handle = (void*)uintptr_t(Fd);
 #endif	// ZEN_PLATFORM_WINDOWS
 
-	IoBuffer Iob(IoBuffer::File, Handle, 0, FileSize);
-	Iob.m_Core->SetIsWholeFile(true);
-
-	return Iob;
+	return IoBuffer(IoBuffer::File, Handle, 0, FileSize, /*IsWholeFile*/ true);
 }
 
 IoHash