automatic scrub on startup (#667)

- Improvement: Deeper validation of data when scrub is activated (cas/cache/project)
- Improvement: Enabled more multi threading when running scrub operations
- Improvement: Added means to force a scrub operation at startup with a new release using ZEN_DATA_FORCE_SCRUB_VERSION variable in xmake.lua

1 files changed, 64 insertions, 0 deletions

diff --git a/src/zenstore/scrubcontext.cpp b/src/zenstore/scrubcontext.cpp
index fbcd7d33c..8f8ec09a7 100644
--- a/src/zenstore/scrubcontext.cpp
+++ b/src/zenstore/scrubcontext.cpp
@@ -2,6 +2,10 @@
 
 #include "zenstore/scrubcontext.h"
 
+#include <zencore/compress.h>
+#include <zencore/fmtutils.h>
+#include <zencore/logging.h>
+#include <zencore/string.h>
 #include <zencore/workthreadpool.h>
 
 namespace zen {
@@ -62,4 +66,64 @@ ScrubContext::ThrowIfDeadlineExpired() const
 	throw ScrubDeadlineExpiredException();
 }
 
+bool
+ValidateCompressedBuffer(const CompositeBuffer& Buffer, const IoHash* OptionalExpectedHash)
+{
+	IoHash	 HeaderRawHash;
+	uint64_t RawSize			 = 0;
+	uint64_t TotalCompressedSize = 0;
+	if (!CompressedBuffer::ValidateCompressedHeader(Buffer, HeaderRawHash, RawSize, &TotalCompressedSize))
+	{
+		if (OptionalExpectedHash)
+		{
+			ZEN_SCOPED_WARN("compressed buffer header validation failed for chunk with hash {}", *OptionalExpectedHash);
+		}
+		else
+		{
+			ZEN_SCOPED_WARN("compressed buffer header validation failed");
+		}
+		return false;
+	}
+
+	if (OptionalExpectedHash != nullptr && HeaderRawHash != (*OptionalExpectedHash))
+	{
+		ZEN_SCOPED_WARN("compressed buffer hash {} does not match expected hash {}", HeaderRawHash, *OptionalExpectedHash);
+		return false;
+	}
+
+	if (TotalCompressedSize != Buffer.GetSize())
+	{
+		ZEN_SCOPED_WARN("compressed buffer size does not match total compressed size in header for chunk {}", HeaderRawHash);
+		return false;
+	}
+
+	CompressedBuffer Compressed = CompressedBuffer::FromCompressed(Buffer, /* out */ HeaderRawHash, /* out */ RawSize);
+
+	IoHashStream HashStream;
+	if (!Compressed.DecompressToStream(
+			0,
+			RawSize,
+			[&HashStream](uint64_t SourceOffset, uint64_t SourceSize, uint64_t Offset, const CompositeBuffer& Range) -> bool {
+				ZEN_UNUSED(SourceOffset, SourceSize, Offset);
+				for (const SharedBuffer& Segment : Range.GetSegments())
+				{
+					HashStream.Append(Segment);
+				}
+				return true;
+			}))
+	{
+		ZEN_SCOPED_WARN("compressed buffer could not be decompressed for chunk {}", HeaderRawHash);
+		return false;
+	}
+
+	IoHash DecompressedHash = HashStream.GetHash();
+
+	if (HeaderRawHash != DecompressedHash)
+	{
+		ZEN_SCOPED_WARN("decompressed hash {} differs from header hash {}", DecompressedHash, HeaderRawHash);
+		return false;
+	}
+	return true;
+}
+
 }  // namespace zen