Wired up scrubbing to more higher level services

Also moved sharding logic for filecas into a function to redduce cut/pasta

1 files changed, 40 insertions, 11 deletions

diff --git a/zenstore/cidstore.cpp b/zenstore/cidstore.cpp
index 5e266f9d3..9c8d4742c 100644
--- a/zenstore/cidstore.cpp
+++ b/zenstore/cidstore.cpp
@@ -12,9 +12,9 @@
 
 namespace zen {
 
-struct CidStore::CidState
+struct CidStore::Impl
 {
-	CidState(CasStore& InCasStore) : m_CasStore(InCasStore) {}
+	Impl(CasStore& InCasStore) : m_CasStore(InCasStore) {}
 
 	struct IndexEntry
 	{
@@ -42,18 +42,26 @@ struct CidStore::CidState
 
 	void AddCompressedCid(const IoHash& DecompressedId, const IoHash& Compressed)
 	{
+		ZEN_ASSERT(Compressed != IoHash::Zero);
+
 		RwLock::ExclusiveLockScope _(m_Lock);
 		m_CidMap.insert_or_assign(DecompressedId, Compressed);
 		// TODO: it's pretty wasteful to log even idempotent updates
 		// however we can't simply use the boolean returned by insert_or_assign
 		// since there's not a 1:1 mapping between compressed and uncompressed
 		// so if we want a last-write-wins policy then we have to log each update
+		LogMapping(DecompressedId, Compressed);
+	}
+
+	void LogMapping(const IoHash& DecompressedId, const IoHash& Compressed)
+	{
 		m_LogFile.Append({.Uncompressed = DecompressedId, .Compressed = Compressed});
 	}
 
 	IoBuffer FindChunkByCid(const IoHash& DecompressedId)
 	{
 		IoHash CompressedHash;
+
 		{
 			RwLock::SharedLockScope _(m_Lock);
 			if (auto It = m_CidMap.find(DecompressedId); It != m_CidMap.end())
@@ -62,12 +70,9 @@ struct CidStore::CidState
 			}
 		}
 
-		if (CompressedHash != IoHash::Zero)
-		{
-			return m_CasStore.FindChunk(CompressedHash);
-		}
+		ZEN_ASSERT(CompressedHash != IoHash::Zero);
 
-		return IoBuffer();
+		return m_CasStore.FindChunk(CompressedHash);
 	}
 
 	bool ContainsChunk(const IoHash& DecompressedId)
@@ -75,7 +80,17 @@ struct CidStore::CidState
 		RwLock::SharedLockScope _(m_Lock);
 		// Note that we do not check CAS here. This is optimistic but usually
 		// what we want.
-		return m_CidMap.find(DecompressedId) != m_CidMap.end();
+		auto It = m_CidMap.find(DecompressedId);
+
+		if (It == m_CidMap.end())
+		{
+			// Not in map, or tombstone
+			return false;
+		}
+
+		ZEN_ASSERT(It->second != IoHash::Zero);
+
+		return true;
 	}
 
 	void InitializeIndex(const std::filesystem::path& RootDir)
@@ -87,6 +102,8 @@ struct CidStore::CidState
 
 		m_LogFile.Open(SlogPath, IsNew);
 
+		uint64_t TombstoneCount = 0;
+
 		m_LogFile.Replay([&](const IndexEntry& Ie) {
 			if (Ie.Compressed != IoHash::Zero)
 			{
@@ -97,16 +114,24 @@ struct CidStore::CidState
 			{
 				// Tombstone
 				m_CidMap.erase(Ie.Uncompressed);
+				++TombstoneCount;
 			}
 		});
 
-		ZEN_DEBUG("CID index initialized: {} entries found", m_CidMap.size());
+		ZEN_INFO("CID index initialized: {} entries found ({} tombstones)", m_CidMap.size(), TombstoneCount);
 	}
 
 	void Flush() { m_LogFile.Flush(); }
 
 	void Scrub(ScrubContext& Ctx)
 	{
+		if (Ctx.ScrubTimestamp() == m_LastScrubTime)
+		{
+			return;
+		}
+
+		m_LastScrubTime = Ctx.ScrubTimestamp();
+
 		CasChunkSet ChunkSet;
 
 		{
@@ -126,7 +151,9 @@ struct CidStore::CidState
 			return;
 		}
 
-		ZEN_ERROR("Scrubbing found that {} cid mappings mapped to non-existent CAS chunks", ChunkSet.GetChunkSet().size());
+		ZEN_ERROR("Scrubbing found that {} cid mappings (out of {}) mapped to non-existent CAS chunks. These mappings will be removed",
+				  ChunkSet.GetChunkSet().size(),
+				  m_CidMap.size());
 
 		// Erase all mappings to chunks which are not present in the underlying CAS store
 		// we do this by removing mappings from the in-memory lookup structure and also
@@ -163,11 +190,13 @@ struct CidStore::CidState
 
 		Ctx.ReportBadChunks(BadChunks);
 	}
+
+	uint64_t m_LastScrubTime = 0;
 };
 
 //////////////////////////////////////////////////////////////////////////
 
-CidStore::CidStore(CasStore& InCasStore, const std::filesystem::path& RootDir) : m_Impl(std::make_unique<CidState>(InCasStore))
+CidStore::CidStore(CasStore& InCasStore, const std::filesystem::path& RootDir) : m_Impl(std::make_unique<Impl>(InCasStore))
 {
 	m_Impl->InitializeIndex(RootDir);
 }