oplog prep gc fix (#216)

- Added option gc-validation to zenserver that does a check for missing references in all oplog post full GC. Enabled by default.
- Feature: Added option gc-validation to zen gc command to control reference validation. Enabled by default.
- Added more details in post GC log.
- Fixed race condition in oplog writes which could cause used attachments to be incorrectly removed by GC

1 files changed, 35 insertions, 18 deletions

diff --git a/src/zenstore/cache/structuredcachestore.cpp b/src/zenstore/cache/structuredcachestore.cpp
index 578929198..512f1d7f2 100644
--- a/src/zenstore/cache/structuredcachestore.cpp
+++ b/src/zenstore/cache/structuredcachestore.cpp
@@ -364,6 +364,7 @@ ZenCacheNamespace::EnableUpdateCapture()
 {
 	m_DiskLayer.EnableUpdateCapture();
 }
+
 void
 ZenCacheNamespace::DisableUpdateCapture()
 {
@@ -850,12 +851,10 @@ ZenCacheStore::GetNamespace(std::string_view Namespace)
 																		  m_BasePath / fmt::format("{}{}", NamespaceDiskPrefix, Namespace),
 																		  m_Configuration.NamespaceConfig));
 
-	m_UpdateCaptureLock.WithExclusiveLock([&]() {
-		if (m_CapturedNamespaces)
-		{
-			m_CapturedNamespaces->push_back(std::string(Namespace));
-		}
-	});
+	if (m_CapturedNamespaces)
+	{
+		m_CapturedNamespaces->push_back(std::string(Namespace));
+	}
 
 	return NewNamespace.first->second.get();
 }
@@ -1039,7 +1038,8 @@ ZenCacheStore::LockState(GcCtx& Ctx)
 void
 ZenCacheStore::EnableUpdateCapture()
 {
-	m_UpdateCaptureLock.WithExclusiveLock([&]() {
+	std::vector<ZenCacheNamespace*> Namespaces;
+	m_NamespacesLock.WithExclusiveLock([&]() {
 		if (m_UpdateCaptureRefCounter == 0)
 		{
 			ZEN_ASSERT(!m_CapturedNamespaces);
@@ -1050,21 +1050,24 @@ ZenCacheStore::EnableUpdateCapture()
 			ZEN_ASSERT(m_CapturedNamespaces);
 		}
 		m_UpdateCaptureRefCounter++;
+		Namespaces.reserve(m_Namespaces.size());
+		for (auto& NamespaceIt : m_Namespaces)
+		{
+			Namespaces.push_back(NamespaceIt.second.get());
+		}
 	});
-	for (auto& NamespaceIt : m_Namespaces)
+
+	for (ZenCacheNamespace* Namespace : Namespaces)
 	{
-		NamespaceIt.second->EnableUpdateCapture();
+		Namespace->EnableUpdateCapture();
 	}
 }
 
 void
 ZenCacheStore::DisableUpdateCapture()
 {
-	for (auto& NamespaceIt : m_Namespaces)
-	{
-		NamespaceIt.second->DisableUpdateCapture();
-	}
-	m_UpdateCaptureLock.WithExclusiveLock([&]() {
+	std::vector<ZenCacheNamespace*> Namespaces;
+	m_NamespacesLock.WithExclusiveLock([&]() {
 		ZEN_ASSERT(m_CapturedNamespaces);
 		ZEN_ASSERT(m_UpdateCaptureRefCounter > 0);
 		m_UpdateCaptureRefCounter--;
@@ -1072,13 +1075,21 @@ ZenCacheStore::DisableUpdateCapture()
 		{
 			m_CapturedNamespaces.reset();
 		}
+		Namespaces.reserve(m_Namespaces.size());
+		for (auto& NamespaceIt : m_Namespaces)
+		{
+			Namespaces.push_back(NamespaceIt.second.get());
+		}
 	});
+	for (ZenCacheNamespace* Namespace : Namespaces)
+	{
+		Namespace->DisableUpdateCapture();
+	}
 }
 
 std::vector<std::string>
-ZenCacheStore::GetCapturedNamespaces()
+ZenCacheStore::GetCapturedNamespacesLocked()
 {
-	RwLock::SharedLockScope _(m_UpdateCaptureLock);
 	if (m_CapturedNamespaces)
 	{
 		return *m_CapturedNamespaces;
@@ -1149,7 +1160,7 @@ public:
 					 AddedBuckets.size());
 		});
 
-		std::vector<std::string> AddedNamespaces = m_CacheStore.GetCapturedNamespaces();
+		std::vector<std::string> AddedNamespaces = m_CacheStore.GetCapturedNamespacesLocked();
 
 		for (const std::string& AddedNamespace : AddedNamespaces)
 		{
@@ -1165,7 +1176,7 @@ public:
 		for (auto& NamepaceKV : m_CacheStore.m_Namespaces)
 		{
 			ZenCacheNamespace&		 Namespace			   = *NamepaceKV.second;
-			std::vector<std::string> NamespaceAddedBuckets = Namespace.m_DiskLayer.GetCapturedBuckets();
+			std::vector<std::string> NamespaceAddedBuckets = Namespace.m_DiskLayer.GetCapturedBucketsLocked();
 			for (const std::string& AddedBucketName : NamespaceAddedBuckets)
 			{
 				if (auto It = Namespace.m_DiskLayer.m_Buckets.find(AddedBucketName); It != Namespace.m_DiskLayer.m_Buckets.end())
@@ -1244,6 +1255,12 @@ ZenCacheStore::CreateReferenceCheckers(GcCtx& Ctx)
 	return Checkers;
 }
 
+std::vector<GcReferenceValidator*>
+ZenCacheStore::CreateReferenceValidators(GcCtx& /*Ctx*/)
+{
+	return {};
+}
+
 //////////////////////////////////////////////////////////////////////////
 
 #if ZEN_WITH_TESTS