improved scrubbing of oplogs and filecas (#596)

- Improvement: Scrub command now validates compressed buffer hashes in filecas storage (used for large chunks)
- Improvement: Added --dry, --no-gc and --no-cas options to zen scrub command
- Improvement: Implemented oplog scrubbing (previously was a no-op)
- Improvement: Implemented support for running scrubbint at startup with --scrub=<options>

6 files changed, 113 insertions, 46 deletions

diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index 96ab65a5f..b21f9f8d8 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -254,6 +254,12 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 {
 	ZEN_TRACE_CPU("CasContainer::ScrubStorage");
 
+	if (Ctx.IsSkipCas())
+	{
+		ZEN_INFO("SKIPPED scrubbing: '{}'", m_BlocksBasePath);
+		return;
+	}
+
 	ZEN_INFO("scrubbing '{}'", m_BlocksBasePath);
 
 	std::vector<IoHash>				BadKeys;
@@ -297,21 +303,12 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 			uint64_t RawSize;
 			if (CompressedBuffer::ValidateCompressedHeader(Buffer, RawHash, RawSize))
 			{
-				if (RawHash != Hash)
+				if (RawHash == Hash)
 				{
-					// Hash mismatch
-					BadKeys.push_back(Hash);
+					// TODO: this should also hash the (decompressed) contents
 					return;
 				}
-				return;
-			}
-#if ZEN_WITH_TESTS
-			IoHash ComputedHash = IoHash::HashBuffer(Data, Size);
-			if (ComputedHash == Hash)
-			{
-				return;
 			}
-#endif
 			BadKeys.push_back(Hash);
 		};
 
@@ -326,26 +323,15 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 
 			IoHash	 RawHash;
 			uint64_t RawSize;
-			// TODO: Add API to verify compressed buffer without having to memorymap the whole file
+			// TODO: Add API to verify compressed buffer without having to memory-map the whole file
 			if (CompressedBuffer::ValidateCompressedHeader(Buffer, RawHash, RawSize))
 			{
-				if (RawHash != Hash)
+				if (RawHash == Hash)
 				{
-					// Hash mismatch
-					BadKeys.push_back(Hash);
+					// TODO: this should also hash the (decompressed) contents
 					return;
 				}
-				return;
-			}
-#if ZEN_WITH_TESTS
-			IoHashStream Hasher;
-			File.StreamByteRange(Offset, Size, [&](const void* Data, size_t Size) { Hasher.Append(Data, Size); });
-			IoHash ComputedHash = Hasher.GetHash();
-			if (ComputedHash == Hash)
-			{
-				return;
 			}
-#endif
 			BadKeys.push_back(Hash);
 		};
 
@@ -398,7 +384,7 @@ CasContainerStrategy::ScrubStorage(ScrubContext& Ctx)
 		Ctx.ReportBadCidChunks(BadKeys);
 	}
 
-	ZEN_INFO("compact cas scrubbed: {} chunks ({})", ChunkCount, NiceBytes(ChunkBytes));
+	ZEN_INFO("scrubbed {} chunks ({}) in '{}'", ChunkCount, NiceBytes(ChunkBytes), m_RootDirectory / m_ContainerBaseName);
 }
 
 void
diff --git a/src/zenstore/filecas.cpp b/src/zenstore/filecas.cpp
index 5da612e30..f18509758 100644
--- a/src/zenstore/filecas.cpp
+++ b/src/zenstore/filecas.cpp
@@ -846,6 +846,13 @@ FileCasStrategy::ScrubStorage(ScrubContext& Ctx)
 {
 	ZEN_TRACE_CPU("FileCas::ScrubStorage");
 
+	if (Ctx.IsSkipCas())
+	{
+		ZEN_INFO("SKIPPED scrubbing: '{}'", m_RootDirectory);
+		return;
+	}
+
+	Stopwatch Timer;
 	ZEN_INFO("scrubbing file CAS @ '{}'", m_RootDirectory);
 
 	ZEN_ASSERT(m_IsInitialized);
@@ -853,6 +860,8 @@ FileCasStrategy::ScrubStorage(ScrubContext& Ctx)
 	std::vector<IoHash> BadHashes;
 	uint64_t			ChunkCount{0}, ChunkBytes{0};
 
+	int DiscoveredFilesNotInIndex = 0;
+
 	{
 		std::vector<FileCasStrategy::FileCasIndexEntry> ScannedEntries = FileCasStrategy::ScanFolderForCasFiles(m_RootDirectory);
 		RwLock::ExclusiveLockScope						_(m_Lock);
@@ -862,10 +871,13 @@ FileCasStrategy::ScrubStorage(ScrubContext& Ctx)
 			{
 				m_TotalSize.fetch_add(static_cast<uint64_t>(Entry.Size), std::memory_order::relaxed);
 				m_CasLog.Append({.Key = Entry.Key, .Size = Entry.Size});
+				++DiscoveredFilesNotInIndex;
 			}
 		}
 	}
 
+	ZEN_INFO("discovered {} files @ '{}' ({} not in index), scrubbing", m_Index.size(), m_RootDirectory, DiscoveredFilesNotInIndex);
+
 	IterateChunks([&](const IoHash& Hash, IoBuffer&& Payload) {
 		if (!Payload)
 		{
@@ -875,25 +887,65 @@ FileCasStrategy::ScrubStorage(ScrubContext& Ctx)
 		++ChunkCount;
 		ChunkBytes += Payload.GetSize();
 
+		IoBuffer InMemoryBuffer = IoBufferBuilder::ReadFromFileMaybe(Payload);
+
 		IoHash	 RawHash;
 		uint64_t RawSize;
-		if (CompressedBuffer::ValidateCompressedHeader(Payload, RawHash, RawSize))
+		if (CompressedBuffer::ValidateCompressedHeader(Payload, /* out */ RawHash, /* out */ RawSize))
 		{
-			if (RawHash != Hash)
+			if (RawHash == Hash)
 			{
-				// Hash mismatch
-				BadHashes.push_back(Hash);
-				return;
+				// Header hash matches the file name, full validation requires that
+				// we check that the decompressed data hash also matches
+
+				CompressedBuffer CompBuffer = CompressedBuffer::FromCompressedNoValidate(std::move(InMemoryBuffer));
+
+				OodleCompressor		  Compressor;
+				OodleCompressionLevel CompressionLevel;
+				uint64_t			  BlockSize;
+				if (CompBuffer.TryGetCompressParameters(Compressor, CompressionLevel, BlockSize))
+				{
+					if (BlockSize == 0)
+					{
+						BlockSize = 256 * 1024;
+					}
+					else if (BlockSize < (1024 * 1024))
+					{
+						BlockSize = BlockSize * (1024 * 1024 / BlockSize);
+					}
+
+					std::unique_ptr<uint8_t[]> DecompressionBuffer(new uint8_t[BlockSize]);
+
+					IoHashStream Hasher;
+
+					uint64_t RawOffset = 0;
+					while (RawSize)
+					{
+						const uint64_t DecompressedBlockSize = Min(BlockSize, RawSize);
+
+						bool Ok = CompBuffer.TryDecompressTo(MutableMemoryView((void*)DecompressionBuffer.get(), DecompressedBlockSize),
+															 RawOffset);
+
+						if (Ok)
+						{
+							Hasher.Append(DecompressionBuffer.get(), DecompressedBlockSize);
+						}
+
+						RawSize -= DecompressedBlockSize;
+						RawOffset += DecompressedBlockSize;
+					}
+
+					const IoHash FinalHash = Hasher.GetHash();
+
+					if (FinalHash == Hash)
+					{
+						// all good
+						return;
+					}
+				}
 			}
-			return;
-		}
-#if ZEN_WITH_TESTS
-		IoHash ComputedHash = IoHash::HashBuffer(CompositeBuffer(SharedBuffer(std::move(Payload))));
-		if (ComputedHash == Hash)
-		{
-			return;
 		}
-#endif
+
 		BadHashes.push_back(Hash);
 	});
 
@@ -901,7 +953,7 @@ FileCasStrategy::ScrubStorage(ScrubContext& Ctx)
 
 	if (!BadHashes.empty())
 	{
-		ZEN_WARN("file CAS scrubbing: {} bad chunks found", BadHashes.size());
+		ZEN_WARN("file CAS scrubbing: {} bad chunks found @ '{}'", BadHashes.size(), m_RootDirectory);
 
 		if (Ctx.RunRecovery())
 		{
@@ -914,10 +966,14 @@ FileCasStrategy::ScrubStorage(ScrubContext& Ctx)
 
 				if (Ec)
 				{
-					ZEN_WARN("failed to delete file for chunk {}", Hash);
+					ZEN_WARN("failed to delete file for chunk {}: {}", Hash, Ec.message());
 				}
 			}
 		}
+		else
+		{
+			ZEN_WARN("recovery: NOT deleting backing files for {} bad chunks", BadHashes.size());
+		}
 	}
 
 	// Let whomever it concerns know about the bad chunks. This could
@@ -925,7 +981,11 @@ FileCasStrategy::ScrubStorage(ScrubContext& Ctx)
 	// than a full validation pass might be able to do
 	Ctx.ReportBadCidChunks(BadHashes);
 
-	ZEN_INFO("file CAS scrubbed: {} chunks ({})", ChunkCount, NiceBytes(ChunkBytes));
+	ZEN_INFO("file CAS @ '{}' scrubbed: {} chunks ({}), took {}",
+			 m_RootDirectory,
+			 ChunkCount,
+			 NiceBytes(ChunkBytes),
+			 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
 }
 
 void
diff --git a/src/zenstore/gc.cpp b/src/zenstore/gc.cpp
index 2660c2643..de653b0e3 100644
--- a/src/zenstore/gc.cpp
+++ b/src/zenstore/gc.cpp
@@ -1712,12 +1712,18 @@ GcScheduler::SchedulerThread()
 					DoGc = false;
 				}
 
+				if (m_TriggerScrubParams->SkipCas)
+				{
+					SkipCid = true;
+				}
+
+				DoDelete	   = !m_TriggerScrubParams->SkipDelete;
 				ScrubTimeslice = m_TriggerScrubParams->MaxTimeslice;
 			}
 
 			if (DoScrubbing)
 			{
-				ScrubStorage(DoDelete, ScrubTimeslice);
+				ScrubStorage(DoDelete, SkipCid, ScrubTimeslice);
 				m_TriggerScrubParams.reset();
 			}
 
@@ -1961,7 +1967,7 @@ GcScheduler::SchedulerThread()
 }
 
 void
-GcScheduler::ScrubStorage(bool DoDelete, std::chrono::seconds TimeSlice)
+GcScheduler::ScrubStorage(bool DoDelete, bool SkipCid, std::chrono::seconds TimeSlice)
 {
 	const std::chrono::steady_clock::time_point TimeNow	 = std::chrono::steady_clock::now();
 	std::chrono::steady_clock::time_point		Deadline = TimeNow + TimeSlice;
@@ -1972,13 +1978,14 @@ GcScheduler::ScrubStorage(bool DoDelete, std::chrono::seconds TimeSlice)
 	}
 
 	Stopwatch Timer;
-	ZEN_INFO("scrubbing STARTING (delete mode => {})", DoDelete);
+	ZEN_INFO("scrubbing STARTING (delete mode => {}, skip CID => {})", DoDelete, SkipCid);
 
 	WorkerThreadPool& ThreadPool = GetSmallWorkerPool();
 	ScrubContext	  Ctx{ThreadPool, Deadline};
 
 	try
 	{
+		Ctx.SetSkipCas(SkipCid);
 		Ctx.SetShouldDelete(DoDelete);
 		m_GcManager.ScrubStorage(Ctx);
 	}
diff --git a/src/zenstore/include/zenstore/gc.h b/src/zenstore/include/zenstore/gc.h
index 698b0d4e8..30dd97ce8 100644
--- a/src/zenstore/include/zenstore/gc.h
+++ b/src/zenstore/include/zenstore/gc.h
@@ -492,6 +492,8 @@ public:
 	{
 		bool				 SkipGc		  = false;
 		std::chrono::seconds MaxTimeslice = std::chrono::seconds::max();
+		bool				 SkipDelete	  = false;
+		bool				 SkipCas	  = false;
 	};
 
 	bool TriggerScrub(const TriggerScrubParams& Params);
@@ -508,7 +510,7 @@ private:
 								GcVersion				  UseGCVersion,
 								uint32_t				  CompactBlockUsageThresholdPercent,
 								bool					  Verbose);
-	void		 ScrubStorage(bool DoDelete, std::chrono::seconds TimeSlice);
+	void		 ScrubStorage(bool DoDelete, bool SkipCid, std::chrono::seconds TimeSlice);
 	LoggerRef	 Log() { return m_Log; }
 	virtual bool AreDiskWritesAllowed() const override { return !m_AreDiskWritesBlocked.load(); }
 	DiskSpace	 CheckDiskSpace();
diff --git a/src/zenstore/include/zenstore/scrubcontext.h b/src/zenstore/include/zenstore/scrubcontext.h
index cefaf0888..2f28cfec7 100644
--- a/src/zenstore/include/zenstore/scrubcontext.h
+++ b/src/zenstore/include/zenstore/scrubcontext.h
@@ -38,15 +38,20 @@ public:
 	inline uint64_t ScrubbedBytes() const { return m_ByteCount; }
 
 	HashKeySet BadCids() const;
+	bool	   IsBadCid(const IoHash& Cid) const;
 
 	inline bool RunRecovery() const { return m_Recover; }
 	inline void SetShouldDelete(bool DoDelete) { m_Recover = DoDelete; }
 
+	inline bool IsSkipCas() const { return m_SkipCas; }
+	inline void SetSkipCas(bool DoSkipCas) { m_SkipCas = DoSkipCas; }
+
 	inline WorkerThreadPool& ThreadPool() { return m_WorkerThreadPool; }
 
 private:
 	uint64_t							  m_ScrubTime = GetHifreqTimerValue();
 	bool								  m_Recover	  = true;
+	bool								  m_SkipCas	  = false;
 	std::atomic<uint64_t>				  m_ChunkCount{0};
 	std::atomic<uint64_t>				  m_ByteCount{0};
 	mutable RwLock						  m_Lock;
diff --git a/src/zenstore/scrubcontext.cpp b/src/zenstore/scrubcontext.cpp
index f5a3784c3..fbcd7d33c 100644
--- a/src/zenstore/scrubcontext.cpp
+++ b/src/zenstore/scrubcontext.cpp
@@ -33,6 +33,13 @@ ScrubContext::BadCids() const
 	return m_BadCid;
 }
 
+bool
+ScrubContext::IsBadCid(const IoHash& Cid) const
+{
+	RwLock::SharedLockScope _(m_Lock);
+	return m_BadCid.ContainsHash(Cid);
+}
+
 void
 ScrubContext::ReportBadCidChunks(std::span<IoHash> BadCasChunks)
 {