Make sure we monitor for new project, oplogs, namespaces and buckets during GCv2 (#93)

- Bugfix: Make sure we monitor and include new project/oplogs created during GCv2
- Bugfix: Make sure we monitor and include new namespaces/cache buckets created during GCv2

1 files changed, 235 insertions, 0 deletions

diff --git a/src/zenstore/cache/structuredcachestore.cpp b/src/zenstore/cache/structuredcachestore.cpp
index d9c2d3e59..d7d594af7 100644
--- a/src/zenstore/cache/structuredcachestore.cpp
+++ b/src/zenstore/cache/structuredcachestore.cpp
@@ -375,6 +375,23 @@ ZenCacheNamespace::GetValueDetails(const std::string_view BucketFilter, const st
 	return m_DiskLayer.GetValueDetails(BucketFilter, ValueFilter);
 }
 
+std::vector<RwLock::SharedLockScope>
+ZenCacheNamespace::GetGcReferencerLocks()
+{
+	return m_DiskLayer.GetGcReferencerLocks();
+}
+
+void
+ZenCacheNamespace::EnableUpdateCapture()
+{
+	m_DiskLayer.EnableUpdateCapture();
+}
+void
+ZenCacheNamespace::DisableUpdateCapture()
+{
+	m_DiskLayer.DisableUpdateCapture();
+}
+
 #if ZEN_WITH_TESTS
 void
 ZenCacheNamespace::SetAccessTime(std::string_view Bucket, const IoHash& HashKey, GcClock::TimePoint Time)
@@ -439,11 +456,15 @@ ZenCacheStore::ZenCacheStore(GcManager&					  Gc,
 												m_BasePath / fmt::format("{}{}", NamespaceDiskPrefix, NamespaceName),
 												m_Configuration.NamespaceConfig);
 	}
+	m_Gc.AddGcReferencer(*this);
+	m_Gc.AddGcReferenceLocker(*this);
 }
 
 ZenCacheStore::~ZenCacheStore()
 {
 	ZEN_INFO("closing cache store at '{}'", m_BasePath);
+	m_Gc.RemoveGcReferenceLocker(*this);
+	m_Gc.RemoveGcReferencer(*this);
 	SetLoggingConfig({.EnableWriteLog = false, .EnableAccessLog = false});
 	m_Namespaces.clear();
 }
@@ -844,6 +865,14 @@ ZenCacheStore::GetNamespace(std::string_view Namespace)
 																		  m_JobQueue,
 																		  m_BasePath / fmt::format("{}{}", NamespaceDiskPrefix, Namespace),
 																		  m_Configuration.NamespaceConfig));
+
+	m_UpdateCaptureLock.WithExclusiveLock([&]() {
+		if (m_CapturedNamespaces)
+		{
+			m_CapturedNamespaces->push_back(std::string(Namespace));
+		}
+	});
+
 	return NewNamespace.first->second.get();
 }
 
@@ -1003,6 +1032,212 @@ ZenCacheStore::GetBucketInfo(std::string_view NamespaceName, std::string_view Bu
 	return {};
 }
 
+std::vector<RwLock::SharedLockScope>
+ZenCacheStore::LockState(GcCtx&)
+{
+	std::vector<RwLock::SharedLockScope> Locks;
+	Locks.emplace_back(RwLock::SharedLockScope(m_NamespacesLock));
+	for (auto& NamespaceIt : m_Namespaces)
+	{
+		std::vector<RwLock::SharedLockScope> NamespaceLocks = NamespaceIt.second->GetGcReferencerLocks();
+		for (auto It = std::make_move_iterator(NamespaceLocks.begin()); It != std::make_move_iterator(NamespaceLocks.end()); It++)
+		{
+			Locks.emplace_back(std::move(*It));
+		}
+	}
+	return Locks;
+}
+
+void
+ZenCacheStore::EnableUpdateCapture()
+{
+	m_UpdateCaptureLock.WithExclusiveLock([&]() {
+		if (m_UpdateCaptureRefCounter == 0)
+		{
+			ZEN_ASSERT(!m_CapturedNamespaces);
+			m_CapturedNamespaces = std::make_unique<std::vector<std::string>>();
+		}
+		else
+		{
+			ZEN_ASSERT(m_CapturedNamespaces);
+		}
+		m_UpdateCaptureRefCounter++;
+	});
+	for (auto& NamespaceIt : m_Namespaces)
+	{
+		NamespaceIt.second->EnableUpdateCapture();
+	}
+}
+
+void
+ZenCacheStore::DisableUpdateCapture()
+{
+	for (auto& NamespaceIt : m_Namespaces)
+	{
+		NamespaceIt.second->DisableUpdateCapture();
+	}
+	m_UpdateCaptureLock.WithExclusiveLock([&]() {
+		ZEN_ASSERT(m_CapturedNamespaces);
+		ZEN_ASSERT(m_UpdateCaptureRefCounter > 0);
+		m_UpdateCaptureRefCounter--;
+		if (m_UpdateCaptureRefCounter == 0)
+		{
+			m_CapturedNamespaces.reset();
+		}
+	});
+}
+
+std::vector<std::string>
+ZenCacheStore::GetCapturedNamespaces()
+{
+	RwLock::SharedLockScope _(m_UpdateCaptureLock);
+	if (m_CapturedNamespaces)
+	{
+		return *m_CapturedNamespaces;
+	}
+	return {};
+}
+
+std::string
+ZenCacheStore::GetGcName(GcCtx&)
+{
+	return fmt::format("zencachestore: '{}'", m_BasePath.string());
+}
+
+GcStoreCompactor*
+ZenCacheStore::RemoveExpiredData(GcCtx&, GcStats&)
+{
+	return nullptr;
+}
+
+class CacheStoreReferenceChecker : public GcReferenceChecker
+{
+public:
+	CacheStoreReferenceChecker(ZenCacheStore& InCacheStore) : m_CacheStore(InCacheStore) { m_CacheStore.EnableUpdateCapture(); }
+
+	virtual ~CacheStoreReferenceChecker()
+	{
+		try
+		{
+			m_CacheStore.DisableUpdateCapture();
+		}
+		catch (const std::exception& Ex)
+		{
+			ZEN_ERROR("~CacheStoreReferenceChecker threw exception: '{}'", Ex.what());
+		}
+	}
+
+	virtual std::string GetGcName(GcCtx&) override { return "cachestore"; }
+	virtual void		PreCache(GcCtx&) override {}
+	virtual void		UpdateLockedState(GcCtx& Ctx) override
+	{
+		ZEN_TRACE_CPU("Z$::UpdateLockedState");
+
+		Stopwatch Timer;
+
+		std::vector<ZenCacheDiskLayer::CacheBucket*> AddedBuckets;
+
+		const auto _ = MakeGuard([&] {
+			if (!Ctx.Settings.Verbose)
+			{
+				return;
+			}
+			ZEN_INFO("GCV2: cachestore [LOCKSTATE] '{}': found {} references in {} in {} new buckets",
+					 "cachestore",
+					 m_References.size(),
+					 NiceTimeSpanMs(Timer.GetElapsedTimeMs()),
+					 AddedBuckets.size());
+		});
+
+		std::vector<std::string> AddedNamespaces = m_CacheStore.GetCapturedNamespaces();
+
+		for (const std::string& AddedNamespace : AddedNamespaces)
+		{
+			if (auto It = m_CacheStore.m_Namespaces.find(AddedNamespace); It != m_CacheStore.m_Namespaces.end())
+			{
+				ZenCacheNamespace& Namespace = *It->second;
+				for (auto& BucketKV : Namespace.m_DiskLayer.m_Buckets)
+				{
+					AddedBuckets.push_back(BucketKV.second.get());
+				}
+			}
+		}
+		for (auto& NamepaceKV : m_CacheStore.m_Namespaces)
+		{
+			ZenCacheNamespace&		 Namespace			   = *NamepaceKV.second;
+			std::vector<std::string> NamespaceAddedBuckets = Namespace.m_DiskLayer.GetCapturedBuckets();
+			for (const std::string& AddedBucketName : NamespaceAddedBuckets)
+			{
+				if (auto It = Namespace.m_DiskLayer.m_Buckets.find(AddedBucketName); It != Namespace.m_DiskLayer.m_Buckets.end())
+				{
+					AddedBuckets.push_back(It->second.get());
+				}
+			}
+		}
+
+		for (ZenCacheDiskLayer::CacheBucket* Bucket : AddedBuckets)
+		{
+			bool Continue = Bucket->GetReferencesLocked(Ctx, m_References);
+			if (!Continue)
+			{
+				break;
+			}
+		}
+	}
+
+	virtual void RemoveUsedReferencesFromSet(GcCtx& Ctx, HashSet& IoCids) override
+	{
+		ZEN_TRACE_CPU("Z$::RemoveUsedReferencesFromSet");
+
+		size_t	   InitialCount = IoCids.size();
+		Stopwatch  Timer;
+		const auto _ = MakeGuard([&] {
+			if (!Ctx.Settings.Verbose)
+			{
+				return;
+			}
+			ZEN_INFO("GCV2: projectstore [FILTER REFERENCES] '{}': filtered out {} used references out of {} in {}",
+					 "projectstore",
+					 InitialCount - IoCids.size(),
+					 InitialCount,
+					 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+		});
+
+		for (const IoHash& ReferenceHash : m_References)
+		{
+			if (IoCids.erase(ReferenceHash) == 1)
+			{
+				if (IoCids.empty())
+				{
+					return;
+				}
+			}
+		}
+	}
+
+private:
+	ZenCacheStore&		m_CacheStore;
+	std::vector<IoHash> m_References;
+};
+
+std::vector<GcReferenceChecker*>
+ZenCacheStore::CreateReferenceCheckers(GcCtx& Ctx)
+{
+	ZEN_TRACE_CPU("CacheStore::CreateReferenceCheckers");
+
+	Stopwatch						 Timer;
+	const auto						 _ = MakeGuard([&] {
+		  if (!Ctx.Settings.Verbose)
+		  {
+			  return;
+		  }
+		  ZEN_INFO("GCV2: cachestore [CREATE CHECKERS] '{}': completed in {}", m_BasePath, NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+	  });
+	std::vector<GcReferenceChecker*> Checkers;
+	Checkers.emplace_back(new CacheStoreReferenceChecker(*this));
+	return Checkers;
+}
+
 //////////////////////////////////////////////////////////////////////////
 
 #if ZEN_WITH_TESTS