flush important files

add validation on reading logs and index files

1 files changed, 79 insertions, 5 deletions

diff --git a/zenstore/compactcas.cpp b/zenstore/compactcas.cpp
index f80004a92..5628ce8c2 100644
--- a/zenstore/compactcas.cpp
+++ b/zenstore/compactcas.cpp
@@ -210,6 +210,60 @@ namespace {
 		return Entries;
 	}
 
+	bool ValidateLegacyEntry(const LegacyCasDiskIndexEntry& Entry, std::string& OutReason)
+	{
+		if (Entry.Key == IoHash::Zero)
+		{
+			OutReason = fmt::format("Invalid hash key {}", Entry.Key.ToHexString());
+			return false;
+		}
+		if (Entry.ContentType != ZenContentType::kUnknownContentType)
+		{
+			OutReason =
+				fmt::format("Invalid content type {} for entry {}", static_cast<uint8_t>(Entry.ContentType), Entry.Key.ToHexString());
+			return false;
+		}
+		if ((Entry.Flags & ~LegacyCasDiskIndexEntry::kTombstone) != 0)
+		{
+			OutReason = fmt::format("Invalid flags {} for entry {}", Entry.Flags, Entry.Key.ToHexString());
+			return false;
+		}
+		uint64_t Size = Entry.Location.GetSize();
+		if (Size == 0)
+		{
+			OutReason = fmt::format("Invalid size {} for entry {}", Size, Entry.Key.ToHexString());
+			return false;
+		}
+		return true;
+	}
+
+	bool ValidateEntry(const CasDiskIndexEntry& Entry, std::string& OutReason)
+	{
+		if (Entry.Key == IoHash::Zero)
+		{
+			OutReason = fmt::format("Invalid hash key {}", Entry.Key.ToHexString());
+			return false;
+		}
+		if (Entry.ContentType != ZenContentType::kUnknownContentType)
+		{
+			OutReason =
+				fmt::format("Invalid content type {} for entry {}", static_cast<uint8_t>(Entry.ContentType), Entry.Key.ToHexString());
+			return false;
+		}
+		if ((Entry.Flags & ~CasDiskIndexEntry::kTombstone) != 0)
+		{
+			OutReason = fmt::format("Invalid flags {} for entry {}", Entry.Flags, Entry.Key.ToHexString());
+			return false;
+		}
+		uint64_t Size = Entry.Location.GetSize();
+		if (Size == 0)
+		{
+			OutReason = fmt::format("Invalid size {} for entry {}", Size, Entry.Key.ToHexString());
+			return false;
+		}
+		return true;
+	}
+
 	std::vector<CasDiskIndexEntry> ReadLog(const std::filesystem::path& RootDirectory, const std::string& ContainerBaseName)
 	{
 		std::vector<CasDiskIndexEntry> Entries;
@@ -299,11 +353,17 @@ namespace {
 						 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
 			});
 			LegacyCasLog.Replay([&](const LegacyCasDiskIndexEntry& Record) {
+				std::string InvalidEntryReason;
 				if (Record.Flags & LegacyCasDiskIndexEntry::kTombstone)
 				{
 					LegacyDiskIndex.erase(Record.Key);
 					return;
 				}
+				if (!ValidateLegacyEntry(Record, InvalidEntryReason))
+				{
+					ZEN_WARN("skipping invalid entry in {}, {}", LegacyLogPath, InvalidEntryReason);
+					return;
+				}
 				uint64_t EntryEnd = Record.Location.GetOffset() + Record.Location.GetSize();
 				if (EntryEnd > FileSize)
 				{
@@ -1034,6 +1094,7 @@ CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
 			const std::vector<IoHash>&	   DeleteMap  = DeleteChunks[ChunkMapIndex];
 			std::vector<CasDiskIndexEntry> LogEntries = MakeCasDiskEntries({}, DeleteMap);
 			m_CasLog.Append(LogEntries);
+			m_CasLog.Flush();
 			{
 				RwLock::ExclusiveLockScope _i(m_LocationMapLock);
 				Stopwatch				   Timer;
@@ -1065,6 +1126,7 @@ CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
 				uint32_t					   NextBlockIndex = m_WriteBlockIndex.load(std::memory_order::memory_order_relaxed);
 				std::vector<CasDiskIndexEntry> LogEntries	  = MakeCasDiskEntries(MovedBlockChunks, {});
 				m_CasLog.Append(LogEntries);
+				m_CasLog.Flush();
 
 				{
 					RwLock::ExclusiveLockScope __(m_LocationMapLock);
@@ -1139,6 +1201,7 @@ CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
 		const std::vector<IoHash>&	   DeleteMap  = DeleteChunks[ChunkMapIndex];
 		std::vector<CasDiskIndexEntry> LogEntries = MakeCasDiskEntries(MovedBlockChunks, DeleteMap);
 		m_CasLog.Append(LogEntries);
+		m_CasLog.Flush();
 		{
 			RwLock::ExclusiveLockScope __(m_LocationMapLock);
 			Stopwatch				   Timer;
@@ -1199,8 +1262,6 @@ CasContainerStrategy::CollectGarbage(GcContext& GcCtx)
 void
 CasContainerStrategy::MakeIndexSnapshot()
 {
-	// Flush on done
-
 	ZEN_INFO("write store {} snapshot", m_Config.RootDirectory / m_ContainerBaseName);
 	uint64_t   EntryCount = 0;
 	Stopwatch  Timer;
@@ -1233,6 +1294,7 @@ CasContainerStrategy::MakeIndexSnapshot()
 	{
 		RwLock::ExclusiveLockScope __(m_InsertLock);
 		RwLock::ExclusiveLockScope ___(m_LocationMapLock);
+		m_CasLog.Flush();
 		m_CasLog.Close();
 
 		if (fs::is_regular_file(STmplogPath))
@@ -1269,6 +1331,7 @@ CasContainerStrategy::MakeIndexSnapshot()
 		CasDiskIndexHeader Header = {.PayloadAlignment = gsl::narrow<uint32_t>(m_PayloadAlignment), .EntryCount = Entries.size()};
 		ObjectIndexFile.Write(&Header, sizeof(CasDiskIndexEntry), 0);
 		ObjectIndexFile.Write(Entries.data(), Entries.size() * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexEntry));
+		ObjectIndexFile.Flush();
 		ObjectIndexFile.Close();
 		EntryCount = Entries.size();
 	}
@@ -1294,6 +1357,7 @@ CasContainerStrategy::MakeIndexSnapshot()
 			TCasLogFile<CasDiskIndexEntry> RecoveredCasLog;
 			RecoveredCasLog.Open(SRecoveredlogPath, CasLogFile::EMode::kWrite);
 			RecoveredCasLog.Append(Records);
+			RecoveredCasLog.Flush();
 			RecoveredCasLog.Close();
 
 			fs::remove(SlogPath);
@@ -1344,9 +1408,14 @@ CasContainerStrategy::OpenContainer(bool IsNewStore)
 
 	{
 		std::vector<CasDiskIndexEntry> IndexEntries = ReadIndexFile(m_Config.RootDirectory, m_ContainerBaseName, m_PayloadAlignment);
+		std::string					   InvalidEntryReason;
 		for (const CasDiskIndexEntry& Entry : IndexEntries)
 		{
-			// Log and skip entries that don't makes sense
+			if (!ValidateEntry(Entry, InvalidEntryReason))
+			{
+				ZEN_WARN("skipping invalid entry in {}, {}", GetIndexPath(m_Config.RootDirectory, m_ContainerBaseName), InvalidEntryReason);
+				continue;
+			}
 			m_LocationMap[Entry.Key] = Entry.Location;
 		}
 	}
@@ -1354,14 +1423,19 @@ CasContainerStrategy::OpenContainer(bool IsNewStore)
 	bool MakeSnapshot = false;
 	{
 		std::vector<CasDiskIndexEntry> LogEntries = ReadLog(m_Config.RootDirectory, m_ContainerBaseName);
+		std::string					   InvalidEntryReason;
 		for (const CasDiskIndexEntry& Entry : LogEntries)
 		{
-			// Log and skip entries that don't makes sense
 			if (Entry.Flags & CasDiskIndexEntry::kTombstone)
 			{
 				m_LocationMap.erase(Entry.Key);
 				continue;
 			}
+			if (!ValidateEntry(Entry, InvalidEntryReason))
+			{
+				ZEN_WARN("skipping invalid entry in {}, {}", GetLogPath(m_Config.RootDirectory, m_ContainerBaseName), InvalidEntryReason);
+				continue;
+			}
 			m_LocationMap[Entry.Key] = Entry.Location;
 		}
 		MakeSnapshot = !LogEntries.empty();
@@ -1377,9 +1451,9 @@ CasContainerStrategy::OpenContainer(bool IsNewStore)
 		}
 		std::vector<CasDiskIndexEntry> LegacyEntries =
 			MigrateLegacyData(m_Config.RootDirectory, m_ContainerBaseName, m_MaxBlockSize, m_PayloadAlignment, true, ExistingChunks);
+		std::string InvalidEntryReason;
 		for (const CasDiskIndexEntry& Entry : LegacyEntries)
 		{
-			// Log and skip entries that don't makes sense
 			m_LocationMap[Entry.Key] = Entry.Location;
 		}
 		MakeSnapshot |= !LegacyEntries.empty();