clean up log/index reading and fix incorrect logging about bad log files (#286)

1 files changed, 96 insertions, 82 deletions

diff --git a/src/zenstore/compactcas.cpp b/src/zenstore/compactcas.cpp
index 2974570e5..275410ae6 100644
--- a/src/zenstore/compactcas.cpp
+++ b/src/zenstore/compactcas.cpp
@@ -586,103 +586,93 @@ CasContainerStrategy::MakeIndexSnapshot()
 }
 
 uint64_t
-CasContainerStrategy::ReadIndexFile()
+CasContainerStrategy::ReadIndexFile(const std::filesystem::path& IndexPath, uint32_t& OutVersion)
 {
 	std::vector<CasDiskIndexEntry> Entries;
-	std::filesystem::path		   IndexPath = GetIndexPath(m_RootDirectory, m_ContainerBaseName);
-	if (std::filesystem::is_regular_file(IndexPath))
-	{
-		Stopwatch  Timer;
-		const auto _ = MakeGuard([&] {
-			ZEN_INFO("read store '{}' index containing {} entries in {}",
-					 IndexPath,
-					 Entries.size(),
-					 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
-		});
+	Stopwatch					   Timer;
+	const auto					   _ = MakeGuard([&] {
+		ZEN_INFO("read store '{}' index containing {} entries in {}", IndexPath, Entries.size(), NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+	});
 
-		BasicFile ObjectIndexFile;
-		ObjectIndexFile.Open(IndexPath, BasicFile::Mode::kRead);
-		uint64_t Size = ObjectIndexFile.FileSize();
-		if (Size >= sizeof(CasDiskIndexHeader))
+	BasicFile ObjectIndexFile;
+	ObjectIndexFile.Open(IndexPath, BasicFile::Mode::kRead);
+	uint64_t Size = ObjectIndexFile.FileSize();
+	if (Size >= sizeof(CasDiskIndexHeader))
+	{
+		uint64_t		   ExpectedEntryCount = (Size - sizeof(sizeof(CasDiskIndexHeader))) / sizeof(CasDiskIndexEntry);
+		CasDiskIndexHeader Header;
+		ObjectIndexFile.Read(&Header, sizeof(Header), 0);
+		if ((Header.Magic == CasDiskIndexHeader::ExpectedMagic) && (Header.Version == CasDiskIndexHeader::CurrentVersion) &&
+			(Header.Checksum == CasDiskIndexHeader::ComputeChecksum(Header)) && (Header.PayloadAlignment > 0) &&
+			(Header.EntryCount <= ExpectedEntryCount))
 		{
-			uint64_t		   ExpectedEntryCount = (Size - sizeof(sizeof(CasDiskIndexHeader))) / sizeof(CasDiskIndexEntry);
-			CasDiskIndexHeader Header;
-			ObjectIndexFile.Read(&Header, sizeof(Header), 0);
-			if ((Header.Magic == CasDiskIndexHeader::ExpectedMagic) && (Header.Version == CasDiskIndexHeader::CurrentVersion) &&
-				(Header.Checksum == CasDiskIndexHeader::ComputeChecksum(Header)) && (Header.PayloadAlignment > 0) &&
-				(Header.EntryCount <= ExpectedEntryCount))
-			{
-				Entries.resize(Header.EntryCount);
-				ObjectIndexFile.Read(Entries.data(), Header.EntryCount * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexHeader));
-				m_PayloadAlignment = Header.PayloadAlignment;
+			Entries.resize(Header.EntryCount);
+			ObjectIndexFile.Read(Entries.data(), Header.EntryCount * sizeof(CasDiskIndexEntry), sizeof(CasDiskIndexHeader));
+			m_PayloadAlignment = Header.PayloadAlignment;
 
-				std::string InvalidEntryReason;
-				for (const CasDiskIndexEntry& Entry : Entries)
+			std::string InvalidEntryReason;
+			for (const CasDiskIndexEntry& Entry : Entries)
+			{
+				if (!ValidateEntry(Entry, InvalidEntryReason))
 				{
-					if (!ValidateEntry(Entry, InvalidEntryReason))
-					{
-						ZEN_WARN("skipping invalid entry in '{}', reason: '{}'", IndexPath, InvalidEntryReason);
-						continue;
-					}
-					m_LocationMap[Entry.Key] = Entry.Location;
+					ZEN_WARN("skipping invalid entry in '{}', reason: '{}'", IndexPath, InvalidEntryReason);
+					continue;
 				}
-
-				return Header.LogPosition;
-			}
-			else
-			{
-				ZEN_WARN("skipping invalid index file '{}'", IndexPath);
+				m_LocationMap[Entry.Key] = Entry.Location;
 			}
+
+			OutVersion = CasDiskIndexHeader::CurrentVersion;
+			return Header.LogPosition;
+		}
+		else
+		{
+			ZEN_WARN("skipping invalid index file '{}'", IndexPath);
 		}
 	}
 	return 0;
 }
 
 uint64_t
-CasContainerStrategy::ReadLog(uint64_t SkipEntryCount)
+CasContainerStrategy::ReadLog(const std::filesystem::path& LogPath, uint64_t SkipEntryCount)
 {
-	std::filesystem::path LogPath = GetLogPath(m_RootDirectory, m_ContainerBaseName);
-	if (std::filesystem::is_regular_file(LogPath))
+	size_t	   LogEntryCount = 0;
+	Stopwatch  Timer;
+	const auto _ = MakeGuard([&] {
+		ZEN_INFO("read store '{}' log containing {} entries in {}",
+				 m_RootDirectory / m_ContainerBaseName,
+				 LogEntryCount,
+				 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
+	});
+
+	TCasLogFile<CasDiskIndexEntry> CasLog;
+	CasLog.Open(LogPath, CasLogFile::Mode::kRead);
+	if (CasLog.Initialize())
 	{
-		size_t	   LogEntryCount = 0;
-		Stopwatch  Timer;
-		const auto _ = MakeGuard([&] {
-			ZEN_INFO("read store '{}' log containing {} entries in {}",
-					 m_RootDirectory / m_ContainerBaseName,
-					 LogEntryCount,
-					 NiceTimeSpanMs(Timer.GetElapsedTimeMs()));
-		});
-
-		TCasLogFile<CasDiskIndexEntry> CasLog;
-		CasLog.Open(LogPath, CasLogFile::Mode::kRead);
-		if (CasLog.Initialize())
+		uint64_t EntryCount = CasLog.GetLogCount();
+		if (EntryCount < SkipEntryCount)
 		{
-			uint64_t EntryCount = CasLog.GetLogCount();
-			if (EntryCount < SkipEntryCount)
-			{
-				ZEN_WARN("reading full log at '{}', reason: Log position from index snapshot is out of range", LogPath);
-				SkipEntryCount = 0;
-			}
-			LogEntryCount = EntryCount - SkipEntryCount;
-			CasLog.Replay(
-				[&](const CasDiskIndexEntry& Record) {
-					LogEntryCount++;
-					std::string InvalidEntryReason;
-					if (Record.Flags & CasDiskIndexEntry::kTombstone)
-					{
-						m_LocationMap.erase(Record.Key);
-						return;
-					}
-					if (!ValidateEntry(Record, InvalidEntryReason))
-					{
-						ZEN_WARN("skipping invalid entry in '{}', reason: '{}'", LogPath, InvalidEntryReason);
-						return;
-					}
-					m_LocationMap[Record.Key] = Record.Location;
-				},
-				SkipEntryCount);
-			return LogEntryCount;
+			ZEN_WARN("reading full log at '{}', reason: Log position from index snapshot is out of range", LogPath);
+			SkipEntryCount = 0;
 		}
+		LogEntryCount = EntryCount - SkipEntryCount;
+		CasLog.Replay(
+			[&](const CasDiskIndexEntry& Record) {
+				LogEntryCount++;
+				std::string InvalidEntryReason;
+				if (Record.Flags & CasDiskIndexEntry::kTombstone)
+				{
+					m_LocationMap.erase(Record.Key);
+					return;
+				}
+				if (!ValidateEntry(Record, InvalidEntryReason))
+				{
+					ZEN_WARN("skipping invalid entry in '{}', reason: '{}'", LogPath, InvalidEntryReason);
+					return;
+				}
+				m_LocationMap[Record.Key] = Record.Location;
+			},
+			SkipEntryCount);
+		return LogEntryCount;
 	}
 	return 0;
 }
@@ -706,10 +696,34 @@ CasContainerStrategy::OpenContainer(bool IsNewStore)
 	std::unordered_map<uint32_t, uint64_t> BlockSizes =
 		m_BlockStore.Initialize(m_BlocksBasePath, m_MaxBlockSize, BlockStoreDiskLocation::MaxBlockIndex + 1);
 
-	m_LogFlushPosition	   = ReadIndexFile();
-	uint64_t LogEntryCount = ReadLog(m_LogFlushPosition);
+	std::filesystem::path LogPath	= GetLogPath(m_RootDirectory, m_ContainerBaseName);
+	std::filesystem::path IndexPath = GetIndexPath(m_RootDirectory, m_ContainerBaseName);
+
+	if (std::filesystem::is_regular_file(IndexPath))
+	{
+		uint32_t IndexVersion = 0;
+		m_LogFlushPosition	  = ReadIndexFile(IndexPath, IndexVersion);
+		if (IndexVersion == 0)
+		{
+			ZEN_WARN("removing invalid index file at '{}'", IndexPath);
+			std::filesystem::remove(IndexPath);
+		}
+	}
+
+	uint64_t LogEntryCount = 0;
+	if (std::filesystem::is_regular_file(LogPath))
+	{
+		if (TCasLogFile<CasDiskIndexEntry>::IsValid(LogPath))
+		{
+			LogEntryCount = ReadLog(LogPath, m_LogFlushPosition);
+		}
+		else
+		{
+			ZEN_WARN("removing invalid cas log at '{}'", LogPath);
+			std::filesystem::remove(LogPath);
+		}
+	}
 
-	std::filesystem::path LogPath = GetLogPath(m_RootDirectory, m_ContainerBaseName);
 	m_CasLog.Open(LogPath, CasLogFile::Mode::kWrite);
 
 	std::vector<BlockStoreLocation> KnownLocations;