builds download progress include validate (#582)

* take validation into account for progress feedback when downloading builds

3 files changed, 90 insertions, 39 deletions

diff --git a/src/zen/cmds/builds_cmd.cpp b/src/zen/cmds/builds_cmd.cpp
index 5633a90b7..33d8cc943 100644
--- a/src/zen/cmds/builds_cmd.cpp
+++ b/src/zen/cmds/builds_cmd.cpp
@@ -2182,10 +2182,15 @@ namespace {
 
 				if (!IsQuiet)
 				{
+					std::string CloneInfo;
+					if (Updater.m_DiskStats.CloneByteCount > 0)
+					{
+						CloneInfo = fmt::format(" ({} cloned)", NiceBytes(Updater.m_DiskStats.CloneByteCount.load()));
+					}
 					ZEN_CONSOLE(
 						"Downloaded build {}, parts:{} in {}\n"
 						"    Download:   {} ({}) {}bits/s\n"
-						"    Write:      {} ({}) {}B/s\n"
+						"    Write:      {} ({}) {}B/s{}\n"
 						"    Clean:      {}\n"
 						"    Finalize:   {}\n"
 						"    Verify:     {}",
@@ -2198,8 +2203,9 @@ namespace {
 						NiceNum(GetBytesPerSecond(Updater.m_WriteChunkStats.DownloadTimeUs, DownloadByteCount * 8)),
 
 						Updater.m_DiskStats.WriteCount.load(),
-						NiceBytes(Updater.m_DiskStats.WriteByteCount.load()),
+						NiceBytes(Updater.m_WrittenChunkByteCount.load()),
 						NiceNum(GetBytesPerSecond(Updater.m_WriteChunkStats.WriteTimeUs, Updater.m_DiskStats.WriteByteCount.load())),
+						CloneInfo,
 
 						NiceTimeSpanMs(Updater.m_RebuildFolderStateStats.CleanFolderElapsedWallTimeUs / 1000),
 
diff --git a/src/zenremotestore/builds/buildstorageoperations.cpp b/src/zenremotestore/builds/buildstorageoperations.cpp
index 431f1caf3..5630e6cb3 100644
--- a/src/zenremotestore/builds/buildstorageoperations.cpp
+++ b/src/zenremotestore/builds/buildstorageoperations.cpp
@@ -415,7 +415,12 @@ StreamDecompress(std::atomic<bool>&			  AbortFlag,
 				 const std::filesystem::path& CacheFolderPath,
 				 const IoHash&				  SequenceRawHash,
 				 CompositeBuffer&&			  CompressedPart,
-				 DiskStatistics&			  DiskStats)
+				 std::atomic<uint64_t>&		  ReadCount,
+				 std::atomic<uint64_t>&		  ReadByteCount,
+				 std::atomic<uint64_t>&		  WriteCount,
+				 std::atomic<uint64_t>&		  WriteByteCount,
+				 std::atomic<uint64_t>&		  WrittenChunkByteCount,
+				 std::atomic<uint64_t>&		  ValidatedChunkByteCount)
 {
 	ZEN_TRACE_CPU("StreamDecompress");
 	const std::filesystem::path TempChunkSequenceFileName = GetTempChunkedSequenceFileName(CacheFolderPath, SequenceRawHash);
@@ -449,16 +454,19 @@ StreamDecompress(std::atomic<bool>&			  AbortFlag,
 									  [&](uint64_t SourceOffset, uint64_t SourceSize, uint64_t Offset, const CompositeBuffer& RangeBuffer) {
 										  ZEN_UNUSED(SourceOffset);
 										  ZEN_TRACE_CPU("StreamDecompress_Write");
-										  DiskStats.ReadByteCount += SourceSize;
+										  ReadCount++;
+										  ReadByteCount += SourceSize;
 										  if (!AbortFlag)
 										  {
 											  for (const SharedBuffer& Segment : RangeBuffer.GetSegments())
 											  {
 												  Hash.Append(Segment.GetView());
+												  ValidatedChunkByteCount += Segment.GetSize();
 												  DecompressedTemp.Write(Segment, Offset);
 												  Offset += Segment.GetSize();
-												  DiskStats.WriteByteCount += Segment.GetSize();
-												  DiskStats.WriteCount++;
+												  WriteByteCount += Segment.GetSize();
+												  WriteCount++;
+												  WrittenChunkByteCount += Segment.GetSize();
 											  }
 											  return true;
 										  }
@@ -1102,6 +1110,8 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 		BytesToWrite += ScavengeCopyOp.RawSize;
 	}
 
+	uint64_t BytesToValidate = BytesToWrite;
+
 	uint64_t			  TotalRequestCount	  = 0;
 	uint64_t			  TotalPartWriteCount = 0;
 	std::atomic<uint64_t> WritePartsComplete  = 0;
@@ -1997,23 +2007,40 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 					CloneDetails = fmt::format(" ({} cloned)", NiceBytes(m_DiskStats.CloneByteCount.load()));
 				}
 				std::string WriteDetails = m_Options.PrimeCacheOnly ? ""
-																	: fmt::format(" {}/{} ({}B/s) written{}.",
-																				  NiceBytes(m_DiskStats.WriteByteCount.load()),
+																	: fmt::format(" {}/{} ({}B/s) written{}",
+																				  NiceBytes(m_WrittenChunkByteCount.load()),
 																				  NiceBytes(BytesToWrite),
 																				  NiceNum(FilteredWrittenBytesPerSecond.GetCurrent()),
 																				  CloneDetails);
-				std::string Details		 = fmt::format("{}/{} ({}{}) downloaded.{}",
-												   m_DownloadStats.RequestsCompleteCount.load(),
-												   TotalRequestCount,
-												   NiceBytes(DownloadedBytes),
-												   DownloadRateString,
-												   WriteDetails);
+
+				std::string Details = fmt::format("{}/{} ({}{}) downloaded.{}",
+												  m_DownloadStats.RequestsCompleteCount.load(),
+												  TotalRequestCount,
+												  NiceBytes(DownloadedBytes),
+												  DownloadRateString,
+												  WriteDetails);
+
+				std::string Task;
+				if (m_Options.PrimeCacheOnly)
+				{
+					Task = "Downloading      ";
+				}
+				else if (m_WrittenChunkByteCount < BytesToWrite)
+				{
+					Task = "Writing chunks   ";
+				}
+				else
+				{
+					Task = "Verifying chunks ";
+				}
+
 				WriteProgressBar.UpdateState(
-					{.Task			 = m_Options.PrimeCacheOnly ? "Downloading      " : "Writing chunks   ",
+					{.Task			 = Task,
 					 .Details		 = Details,
-					 .TotalCount	 = m_Options.PrimeCacheOnly ? TotalRequestCount : BytesToWrite,
+					 .TotalCount	 = m_Options.PrimeCacheOnly ? TotalRequestCount : (BytesToWrite + BytesToValidate),
 					 .RemainingCount = m_Options.PrimeCacheOnly ? (TotalRequestCount - m_DownloadStats.RequestsCompleteCount.load())
-																: (BytesToWrite - m_DiskStats.WriteByteCount.load()),
+																: ((BytesToWrite + BytesToValidate) -
+																   (m_WrittenChunkByteCount.load() + m_ValidatedChunkByteCount.load())),
 					 .Status		 = BuildOpLogOutput::ProgressBar::State::CalculateStatus(IsAborted, IsPaused)},
 					false);
 			});
@@ -2032,6 +2059,9 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 
 		if (!m_Options.PrimeCacheOnly)
 		{
+			ZEN_ASSERT(m_WrittenChunkByteCount == BytesToWrite);
+			ZEN_ASSERT(m_ValidatedChunkByteCount == BytesToValidate);
+
 			uint32_t RawSequencesMissingWriteCount = 0;
 			for (uint32_t SequenceIndex = 0; SequenceIndex < SequenceIndexChunksLeftToWriteCounters.size(); SequenceIndex++)
 			{
@@ -2071,7 +2101,7 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 					   NiceBytes(DownloadedBytes),
 					   NiceNum(GetBytesPerSecond(FilteredDownloadedBytesPerSecond.GetElapsedTimeUS(), DownloadedBytes * 8)),
 					   NiceTimeSpanMs(FilteredDownloadedBytesPerSecond.GetElapsedTimeUS() / 1000),
-					   NiceBytes(m_DiskStats.WriteByteCount.load()),
+					   NiceBytes(m_WrittenChunkByteCount.load()),
 					   NiceNum(GetBytesPerSecond(FilteredWrittenBytesPerSecond.GetElapsedTimeUS(), m_DiskStats.WriteByteCount.load())),
 					   CloneDetails,
 					   NiceTimeSpanMs(FilteredWrittenBytesPerSecond.GetElapsedTimeUS() / 1000),
@@ -2467,20 +2497,17 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 									ZEN_ASSERT_SLOW(IsFileWithRetry(SourceFilePath));
 
 									LOG_OUTPUT_DEBUG(m_LogOutput, "Copying from '{}' -> '{}'", SourceFilePath, FirstTargetFilePath);
-									const uint64_t		  RawSize = m_LocalContent.RawSizes[LocalPathIndex];
-									std::atomic<uint64_t> WriteCount;
-									std::atomic<uint64_t> WriteByteCount;
-									std::atomic<uint64_t> CloneCount;
-									std::atomic<uint64_t> CloneByteCount;
+									const uint64_t RawSize = m_LocalContent.RawSizes[LocalPathIndex];
 									FastCopyFile(m_Options.AllowFileClone,
 												 m_Options.UseSparseFiles,
 												 SourceFilePath,
 												 FirstTargetFilePath,
 												 RawSize,
-												 WriteCount,
-												 WriteByteCount,
-												 CloneCount,
-												 CloneByteCount);
+												 m_DiskStats.WriteCount,
+												 m_DiskStats.WriteByteCount,
+												 m_DiskStats.CloneCount,
+												 m_DiskStats.CloneByteCount);
+
 									m_RebuildFolderStateStats.FinalizeTreeFilesCopiedCount++;
 								}
 								else
@@ -2554,20 +2581,17 @@ BuildsOperationUpdateFolder::Execute(FolderContent& OutLocalFolderState)
 
 									ZEN_ASSERT_SLOW(IsFileWithRetry(FirstTargetFilePath));
 									LOG_OUTPUT_DEBUG(m_LogOutput, "Copying from '{}' -> '{}'", FirstTargetFilePath, TargetFilePath);
-									const uint64_t		  RawSize = m_RemoteContent.RawSizes[RemotePathIndex];
-									std::atomic<uint64_t> WriteCount;
-									std::atomic<uint64_t> WriteByteCount;
-									std::atomic<uint64_t> CloneCount;
-									std::atomic<uint64_t> CloneByteCount;
+									const uint64_t RawSize = m_RemoteContent.RawSizes[RemotePathIndex];
 									FastCopyFile(m_Options.AllowFileClone,
 												 m_Options.UseSparseFiles,
 												 FirstTargetFilePath,
 												 TargetFilePath,
 												 RawSize,
-												 WriteCount,
-												 WriteByteCount,
-												 CloneCount,
-												 CloneByteCount);
+												 m_DiskStats.WriteCount,
+												 m_DiskStats.WriteByteCount,
+												 m_DiskStats.CloneCount,
+												 m_DiskStats.CloneByteCount);
+
 									m_RebuildFolderStateStats.FinalizeTreeFilesCopiedCount++;
 								}
 
@@ -3132,6 +3156,9 @@ BuildsOperationUpdateFolder::WriteScavengedSequenceToCache(const std::filesystem
 
 	const std::filesystem::path CacheFilePath = GetFinalChunkedSequenceFileName(m_CacheFolderPath, RemoteSequenceRawHash);
 	RenameFile(TempFilePath, CacheFilePath);
+
+	m_WrittenChunkByteCount += RawSize;
+	m_ValidatedChunkByteCount += RawSize;
 }
 
 void
@@ -3884,6 +3911,8 @@ BuildsOperationUpdateFolder::WriteLocalChunkToCache(CloneQueryInterface*					  C
 							m_DiskStats.CloneCount++;
 							m_DiskStats.CloneByteCount += ClonableBytes;
 
+							m_WrittenChunkByteCount += ClonableBytes;
+
 							if (PreBytes > 0)
 							{
 								CompositeBuffer ChunkSource = SourceFile.GetRange(Op.CacheFileOffset, PreBytes);
@@ -3948,7 +3977,16 @@ BuildsOperationUpdateFolder::WriteCompressedChunkToCache(
 	{
 		const std::uint32_t SequenceIndex	= ChunkTargetPtrs.front()->SequenceIndex;
 		const IoHash&		SequenceRawHash = m_RemoteContent.ChunkedContent.SequenceRawHashes[SequenceIndex];
-		StreamDecompress(m_AbortFlag, m_CacheFolderPath, SequenceRawHash, CompositeBuffer(std::move(CompressedPart)), m_DiskStats);
+		StreamDecompress(m_AbortFlag,
+						 m_CacheFolderPath,
+						 SequenceRawHash,
+						 CompositeBuffer(std::move(CompressedPart)),
+						 m_DiskStats.ReadCount,
+						 m_DiskStats.ReadByteCount,
+						 m_DiskStats.WriteCount,
+						 m_DiskStats.WriteByteCount,
+						 m_WrittenChunkByteCount,
+						 m_ValidatedChunkByteCount);
 		return false;
 	}
 	else
@@ -4039,7 +4077,7 @@ BuildsOperationUpdateFolder::WriteSequenceChunkToCache(BufferedWriteFileCache::L
 		SingleChunkFile.Write(Chunk, FileOffset);
 
 		m_DiskStats.WriteCount++;
-		m_DiskStats.WriteByteCount += ChunkSize;
+		m_WrittenChunkByteCount += ChunkSize;
 	}
 	else
 	{
@@ -4056,6 +4094,8 @@ BuildsOperationUpdateFolder::WriteSequenceChunkToCache(BufferedWriteFileCache::L
 
 			m_DiskStats.WriteCount++;
 			m_DiskStats.WriteByteCount += ChunkSize;
+
+			m_WrittenChunkByteCount += ChunkSize;
 		}
 		else
 		{
@@ -4071,6 +4111,8 @@ BuildsOperationUpdateFolder::WriteSequenceChunkToCache(BufferedWriteFileCache::L
 
 			m_DiskStats.WriteCount++;
 			m_DiskStats.WriteByteCount += ChunkSize;
+
+			m_WrittenChunkByteCount += ChunkSize;
 		}
 	}
 }
@@ -4531,7 +4573,7 @@ BuildsOperationUpdateFolder::VerifySequence(uint32_t RemoteSequenceIndex)
 												 ExpectedSize));
 		}
 
-		const IoHash VerifyChunkHash = IoHash::HashBuffer(std::move(VerifyBuffer));
+		const IoHash VerifyChunkHash = IoHash::HashBuffer(std::move(VerifyBuffer), &m_ValidatedChunkByteCount);
 		if (VerifyChunkHash != SequenceRawHash)
 		{
 			throw std::runtime_error(
diff --git a/src/zenremotestore/include/zenremotestore/builds/buildstorageoperations.h b/src/zenremotestore/include/zenremotestore/builds/buildstorageoperations.h
index fcf880e48..f200a342c 100644
--- a/src/zenremotestore/include/zenremotestore/builds/buildstorageoperations.h
+++ b/src/zenremotestore/include/zenremotestore/builds/buildstorageoperations.h
@@ -225,6 +225,7 @@ public:
 	DownloadStatistics			 m_DownloadStats;
 	WriteChunkStatistics		 m_WriteChunkStats;
 	RebuildFolderStateStatistics m_RebuildFolderStateStats;
+	std::atomic<uint64_t>		 m_WrittenChunkByteCount;
 
 private:
 	struct BlockWriteOps
@@ -450,6 +451,8 @@ private:
 	const std::filesystem::path				  m_CacheFolderPath;
 	const std::filesystem::path				  m_TempDownloadFolderPath;
 	const std::filesystem::path				  m_TempBlockFolderPath;
+
+	std::atomic<uint64_t> m_ValidatedChunkByteCount;
 };
 
 struct FindBlocksStatistics