safer oplog import (#52)de/safer-oplog-import

* reference cache gc update capture
* When importing oplogs we now import all attachments first and (optionally clean) write the oplog on success

1 files changed, 71 insertions, 8 deletions

diff --git a/src/zenserver/projectstore/projectstore.cpp b/src/zenserver/projectstore/projectstore.cpp
index 65d2730d8..e452c658e 100644
--- a/src/zenserver/projectstore/projectstore.cpp
+++ b/src/zenserver/projectstore/projectstore.cpp
@@ -831,6 +831,38 @@ ProjectStore::Oplog::Update(const std::filesystem::path& MarkerPath)
 	Write();
 }
 
+bool
+ProjectStore::Oplog::Reset()
+{
+	std::filesystem::path MovedDir;
+
+	{
+		RwLock::ExclusiveLockScope OplogLock(m_OplogLock);
+		m_Storage = {};
+		if (!PrepareDirectoryDelete(m_BasePath, MovedDir))
+		{
+			m_Storage			   = new OplogStorage(this, m_BasePath);
+			const bool StoreExists = m_Storage->Exists();
+			m_Storage->Open(/* IsCreate */ !StoreExists);
+			return false;
+		}
+		m_ChunkMap.clear();
+		m_MetaMap.clear();
+		m_FileMap.clear();
+		m_OpAddressMap.clear();
+		m_LatestOpMap.clear();
+		m_Storage = new OplogStorage(this, m_BasePath);
+		m_Storage->Open(true);
+		CleanDirectory(m_TempPath);
+	}
+	// Erase content on disk
+	if (!MovedDir.empty())
+	{
+		OplogStorage::Delete(MovedDir);
+	}
+	return true;
+}
+
 void
 ProjectStore::Oplog::ReplayLog()
 {
@@ -1167,8 +1199,19 @@ void
 ProjectStore::Oplog::EnableUpdateCapture()
 {
 	m_OplogLock.WithExclusiveLock([&]() {
-		m_UpdatedLSNs	   = std::make_unique<std::vector<int>>();
-		m_NonGCAttachments = std::make_unique<std::vector<IoHash>>();
+		if (m_UpdateCaptureRefCounter == 0)
+		{
+			ZEN_ASSERT(!m_UpdatedLSNs);
+			ZEN_ASSERT(!m_NonGCAttachments);
+			m_UpdatedLSNs	   = std::make_unique<std::vector<int>>();
+			m_NonGCAttachments = std::make_unique<std::vector<IoHash>>();
+		}
+		else
+		{
+			ZEN_ASSERT(m_UpdatedLSNs);
+			ZEN_ASSERT(m_NonGCAttachments);
+		}
+		m_UpdateCaptureRefCounter++;
 	});
 }
 
@@ -1176,8 +1219,15 @@ void
 ProjectStore::Oplog::DisableUpdateCapture()
 {
 	m_OplogLock.WithExclusiveLock([&]() {
-		m_UpdatedLSNs.reset();
-		m_NonGCAttachments.reset();
+		ZEN_ASSERT(m_UpdatedLSNs);
+		ZEN_ASSERT(m_NonGCAttachments);
+		ZEN_ASSERT(m_UpdateCaptureRefCounter > 0);
+		m_UpdateCaptureRefCounter--;
+		if (m_UpdateCaptureRefCounter == 0)
+		{
+			m_UpdatedLSNs.reset();
+			m_NonGCAttachments.reset();
+		}
 	});
 }
 
@@ -3581,6 +3631,7 @@ ProjectStore::Import(ProjectStore::Project& Project, ProjectStore::Oplog& Oplog,
 	size_t MaxChunkEmbedSize		= Params["maxchunkembedsize"sv].AsUInt64(RemoteStoreOptions::DefaultMaxChunkEmbedSize);
 	bool   Force					= Params["force"sv].AsBool(false);
 	bool   IgnoreMissingAttachments = Params["ignoremissingattachments"sv].AsBool(false);
+	bool   CleanOplog				= Params["clean"].AsBool(false);
 
 	CreateRemoteStoreResult RemoteStoreResult = CreateRemoteStore(Params, AuthManager, MaxBlockSize, MaxChunkEmbedSize, Oplog.TempPath());
 
@@ -3594,9 +3645,10 @@ ProjectStore::Import(ProjectStore::Project& Project, ProjectStore::Oplog& Oplog,
 	ZEN_INFO("Loading oplog '{}/{}' from {}", Project.Identifier, Oplog.OplogId(), StoreInfo.Description);
 	JobId JobId = m_JobQueue.QueueJob(
 		fmt::format("Import oplog '{}/{}' from {}", Project.Identifier, Oplog.OplogId(), StoreInfo.Description),
-		[this, ActualRemoteStore = std::move(RemoteStore), OplogPtr = &Oplog, Force, IgnoreMissingAttachments](JobContext& Context) {
+		[this, ActualRemoteStore = std::move(RemoteStore), OplogPtr = &Oplog, Force, IgnoreMissingAttachments, CleanOplog](
+			JobContext& Context) {
 			RemoteProjectStore::Result Result =
-				LoadOplog(m_CidStore, *ActualRemoteStore, *OplogPtr, Force, IgnoreMissingAttachments, &Context);
+				LoadOplog(m_CidStore, *ActualRemoteStore, *OplogPtr, Force, IgnoreMissingAttachments, CleanOplog, &Context);
 			auto Response = ConvertResult(Result);
 			ZEN_INFO("LoadOplog: Status: {} '{}'", ToString(Response.first), Response.second);
 			if (!IsHttpSuccessCode(Response.first))
@@ -4258,11 +4310,22 @@ TEST_CASE_TEMPLATE("project.store.export",
 
 	ProjectStore::Oplog* OplogImport = Project->NewOplog("oplog2", {});
 	CHECK(OplogImport != nullptr);
-	RemoteProjectStore::Result ImportResult = LoadOplog(CidStore, *RemoteStore, *OplogImport, false, false, nullptr);
+
+	RemoteProjectStore::Result ImportResult =
+		LoadOplog(CidStore, *RemoteStore, *OplogImport, /*Force*/ false, /*IgnoreMissingAttachments*/ false, /*CleanOplog*/ false, nullptr);
 	CHECK(ImportResult.ErrorCode == 0);
 
-	RemoteProjectStore::Result ImportForceResult = LoadOplog(CidStore, *RemoteStore, *OplogImport, true, false, nullptr);
+	RemoteProjectStore::Result ImportForceResult =
+		LoadOplog(CidStore, *RemoteStore, *OplogImport, /*Force*/ true, /*IgnoreMissingAttachments*/ false, /*CleanOplog*/ false, nullptr);
 	CHECK(ImportForceResult.ErrorCode == 0);
+
+	RemoteProjectStore::Result ImportCleanResult =
+		LoadOplog(CidStore, *RemoteStore, *OplogImport, /*Force*/ false, /*IgnoreMissingAttachments*/ false, /*CleanOplog*/ true, nullptr);
+	CHECK(ImportCleanResult.ErrorCode == 0);
+
+	RemoteProjectStore::Result ImportForceCleanResult =
+		LoadOplog(CidStore, *RemoteStore, *OplogImport, /*Force*/ true, /*IgnoreMissingAttachments*/ false, /*CleanOplog*/ true, nullptr);
+	CHECK(ImportForceCleanResult.ErrorCode == 0);
 }
 
 TEST_CASE("project.store.gc")