stop exceptions from leaking on threaded work (#102)

* catch exceptions in threaded work
* don't abort all project file/chunk info fetch for single failure

1 files changed, 33 insertions, 23 deletions

diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index dd92483b6..a00b17912 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -336,29 +336,39 @@ CasContainerStrategy::IterateChunks(std::span<IoHash>												  ChunkHashes,
 
 	Latch			 WorkLatch(1);
 	std::atomic_bool AsyncContinue = true;
-	bool			 Continue	   = m_BlockStore.IterateChunks(FoundChunkLocations, [&](uint32_t, std::span<const size_t> ChunkIndexes) {
-		 if (OptionalWorkerPool)
-		 {
-			 WorkLatch.AddCount(1);
-			 OptionalWorkerPool->ScheduleWork([&, ChunkIndexes = std::vector<size_t>(ChunkIndexes.begin(), ChunkIndexes.end())]() {
-				 auto _ = MakeGuard([&WorkLatch]() { WorkLatch.CountDown(); });
-				 if (!AsyncContinue)
-				 {
-					 return;
-				 }
-				 bool Continue = DoOneBlock(ChunkIndexes);
-				 if (!Continue)
-				 {
-					 AsyncContinue.store(false);
-				 }
-			 });
-			 return AsyncContinue.load();
-		 }
-		 else
-		 {
-			 return DoOneBlock(ChunkIndexes);
-		 }
-	 });
+	bool Continue = m_BlockStore.IterateChunks(FoundChunkLocations, [&](uint32_t BlockIndex, std::span<const size_t> ChunkIndexes) {
+		if (OptionalWorkerPool)
+		{
+			WorkLatch.AddCount(1);
+			OptionalWorkerPool->ScheduleWork([&, ChunkIndexes = std::vector<size_t>(ChunkIndexes.begin(), ChunkIndexes.end())]() {
+				auto _ = MakeGuard([&WorkLatch]() { WorkLatch.CountDown(); });
+				if (!AsyncContinue)
+				{
+					return;
+				}
+				try
+				{
+					bool Continue = DoOneBlock(ChunkIndexes);
+					if (!Continue)
+					{
+						AsyncContinue.store(false);
+					}
+				}
+				catch (const std::exception& Ex)
+				{
+					ZEN_WARN("Failed iterating chunks for cas root path {}, block {}. Reason: '{}'",
+							 m_RootDirectory,
+							 BlockIndex,
+							 Ex.what());
+				}
+			});
+			return AsyncContinue.load();
+		}
+		else
+		{
+			return DoOneBlock(ChunkIndexes);
+		}
+	});
 	WorkLatch.CountDown();
 	WorkLatch.Wait();
 	return AsyncContinue.load() && Continue;