Add magic number to WAL and fix version consistency check

Summary: Recovery file is version inconsistent iff it starts with a correct magic number, has at least one integer written after that and that integer differs from kVersion.

Reviewers: mferencevic, ipaljak, vkasljevic, buda

Reviewed By: mferencevic

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D1565

src/durability/hashed_file_reader.hpp

@@ -66,6 +66,9 @@ class HashedFileReader {
   /** Returns the hash of the data read so far from the stream. */
   uint64_t hash() const { return hasher_.hash(); }
 
+  /** Checks whether the end of file is reached. */
+  bool EndOfFile() const { return input_stream_.eof(); }
+
  private:
   Hasher hasher_;
   std::ifstream input_stream_;

src/durability/recovery.cpp

@@ -37,32 +37,27 @@ bool ReadSnapshotSummary(HashedFileReader &buffer, int64_t &vertex_count,
 }
 
 bool VersionConsistency(const fs::path &durability_dir) {
-  const auto snapshot_dir = durability_dir / kSnapshotDir;
+  for (const auto &durability_type : {kSnapshotDir, kWalDir}) {
-  if (fs::exists(snapshot_dir) && fs::is_directory(snapshot_dir)) {
+    auto recovery_dir = durability_dir / durability_type;
-    for (auto &file : fs::directory_iterator(snapshot_dir)) {
+    if (!fs::exists(recovery_dir) || !fs::is_directory(recovery_dir)) continue;
+
+    for (const auto &file : fs::directory_iterator(recovery_dir)) {
       HashedFileReader reader;
       SnapshotDecoder<HashedFileReader> decoder(reader);
 
-      // This is ok because we are only trying to detect version
+      // The following checks are ok because we are only trying to detect
-      // inconsistencies.
+      // version inconsistencies.
       if (!reader.Open(fs::path(file))) continue;
 
-      auto magic_number = durability::kMagicNumber;
+      std::array<uint8_t, 4> target_magic_number =
+          (durability_type == kSnapshotDir) ? durability::kSnapshotMagic
+                                            : durability::kWalMagic;
+      std::array<uint8_t, 4> magic_number;
       if (!reader.Read(magic_number.data(), magic_number.size())) continue;
+      if (magic_number != target_magic_number) continue;
 
-      Value dv;
+      if (reader.EndOfFile()) continue;
-      if (!decoder.ReadValue(&dv, Value::Type::Int) ||
-          dv.ValueInt() != durability::kVersion)
-        return false;
-    }
-  }
 
-  const auto wal_dir = durability_dir / kWalDir;
-  if (fs::exists(snapshot_dir) && fs::is_directory(wal_dir)) {
-    for (auto &file : fs::directory_iterator(wal_dir)) {
-      HashedFileReader reader;
-      communication::bolt::Decoder<HashedFileReader> decoder(reader);
-      if (!reader.Open(fs::path(file))) continue;
       Value dv;
       if (!decoder.ReadValue(&dv, Value::Type::Int) ||
           dv.ValueInt() != durability::kVersion)
@@ -74,25 +69,25 @@ bool VersionConsistency(const fs::path &durability_dir) {
 }
 
 bool ContainsDurabilityFiles(const fs::path &durability_dir) {
-  for (auto &durability_type : {kSnapshotDir, kWalDir}) {
+  for (const auto &durability_type : {kSnapshotDir, kWalDir}) {
-    const auto dtype_dir = durability_dir / durability_type;
+    auto recovery_dir = durability_dir / durability_type;
-    if (fs::exists(dtype_dir) && fs::is_directory(dtype_dir) &&
+    if (fs::exists(recovery_dir) && fs::is_directory(recovery_dir) &&
-        !fs::is_empty(dtype_dir))
+        !fs::is_empty(recovery_dir))
       return true;
   }
   return false;
 }
 
 void MoveToBackup(const fs::path &durability_dir) {
-  const auto backup_dir = durability_dir / kBackupDir;
+  auto backup_dir = durability_dir / kBackupDir;
   utils::CheckDir(backup_dir);
   utils::CheckDir(backup_dir / kSnapshotDir);
   utils::CheckDir(backup_dir / kWalDir);
-  for (auto &durability_type : {kSnapshotDir, kWalDir}) {
+  for (const auto &durability_type : {kSnapshotDir, kWalDir}) {
-    const auto dtype_dir = durability_dir / durability_type;
+    auto recovery_dir = durability_dir / durability_type;
-    if (!fs::exists(dtype_dir) || !fs::is_directory(dtype_dir)) continue;
+    if (!fs::exists(recovery_dir) || !fs::is_directory(recovery_dir)) continue;
-    for (auto &file : fs::directory_iterator(dtype_dir)) {
+    for (const auto &file : fs::directory_iterator(recovery_dir)) {
-      const auto filename = fs::path(file).filename();
+      auto filename = fs::path(file).filename();
       fs::rename(file, backup_dir / durability_type / filename);
     }
   }
@@ -114,9 +109,9 @@ bool RecoverSnapshot(const fs::path &snapshot_file, database::GraphDb *db,
 
   RETURN_IF_NOT(reader.Open(snapshot_file));
 
-  auto magic_number = durability::kMagicNumber;
+  auto magic_number = durability::kSnapshotMagic;
   reader.Read(magic_number.data(), magic_number.size());
-  RETURN_IF_NOT(magic_number == durability::kMagicNumber);
+  RETURN_IF_NOT(magic_number == durability::kSnapshotMagic);
 
   // Read the vertex and edge count, and the hash, from the end of the snapshot.
   int64_t vertex_count;
@@ -323,6 +318,10 @@ bool ApplyOverDeltas(
 
     communication::bolt::Decoder<HashedFileReader> decoder(wal_reader);
 
+    auto magic_number = durability::kWalMagic;
+    wal_reader.Read(magic_number.data(), magic_number.size());
+    if (magic_number != durability::kWalMagic) return false;
+
     Value dv;
     if (!decoder.ReadValue(&dv, Value::Type::Int) ||
         dv.ValueInt() != durability::kVersion)

src/durability/snapshooter.cpp

@@ -27,8 +27,8 @@ bool Encode(const fs::path &snapshot_file, database::GraphDb &db,
     SnapshotEncoder<HashedFileWriter> encoder(buffer);
     int64_t vertex_num = 0, edge_num = 0;
 
-    encoder.WriteRAW(durability::kMagicNumber.data(),
+    encoder.WriteRAW(durability::kSnapshotMagic.data(),
-                     durability::kMagicNumber.size());
+                     durability::kSnapshotMagic.size());
     encoder.WriteInt(durability::kVersion);
 
     // Writes the worker id to snapshot, used to guarantee consistent cluster

src/durability/version.hpp

@@ -11,7 +11,8 @@
 
 namespace durability {
 
-constexpr std::array<uint8_t, 4> kMagicNumber{{'M', 'G', 's', 'n'}};
+constexpr std::array<uint8_t, 4> kSnapshotMagic{{'M', 'G', 's', 'n'}};
+constexpr std::array<uint8_t, 4> kWalMagic{{'M', 'G', 'w', 'l'}};
 
 // The current default version of snapshot and WAL encoding / decoding.
 constexpr int64_t kVersion{6};

src/durability/wal.cpp

@@ -54,6 +54,8 @@ void WriteAheadLog::WalFile::Init() {
     current_wal_file_ = WalFilenameForTransactionId(wal_dir_, worker_id_);
     try {
       writer_.Open(current_wal_file_);
+      encoder_.WriteRAW(durability::kWalMagic.data(),
+                        durability::kWalMagic.size());
       encoder_.WriteInt(durability::kVersion);
     } catch (std::ios_base::failure &) {
       LOG(ERROR) << "Failed to open write-ahead log file: "

tests/manual/snapshot_explorer.cpp

@@ -32,9 +32,9 @@ int main(int argc, char *argv[]) {
 
   CHECK(reader.Open(snapshot_path)) << "Couldn't open snapshot file!";
 
-  auto magic_number = durability::kMagicNumber;
+  auto magic_number = durability::kSnapshotMagic;
   reader.Read(magic_number.data(), magic_number.size());
-  CHECK(magic_number == durability::kMagicNumber) << "Magic number mismatch";
+  CHECK(magic_number == durability::kSnapshotMagic) << "Magic number mismatch";
 
   int64_t vertex_count, edge_count;
   uint64_t hash;

tests/manual/snapshot_generation/snapshot_writer.hpp

@@ -25,8 +25,8 @@ class SnapshotWriter {
                  uint64_t vertex_generator_local_count = 0,
                  uint64_t edge_generator_local_count = 0)
       : worker_id_(worker_id), buffer_(path) {
-    encoder_.WriteRAW(durability::kMagicNumber.data(),
+    encoder_.WriteRAW(durability::kSnapshotMagic.data(),
-                      durability::kMagicNumber.size());
+                      durability::kSnapshotMagic.size());
     encoder_.WriteValue(durability::kVersion);
     encoder_.WriteInt(worker_id_);
     encoder_.WriteInt(vertex_generator_local_count);

tests/manual/wal_explorer.cpp

@@ -8,6 +8,7 @@
 #include "database/state_delta.hpp"
 #include "durability/hashed_file_reader.hpp"
 #include "durability/recovery.hpp"
+#include "durability/version.hpp"
 #include "durability/wal.hpp"
 #include "transactions/type.hpp"
 
@@ -64,6 +65,15 @@ int main(int argc, char *argv[]) {
   CHECK(wal_reader.Open(wal_path)) << "Couldn't open wal file!";
 
   communication::bolt::Decoder<HashedFileReader> decoder(wal_reader);
+
+  auto magic_number = durability::kWalMagic;
+  wal_reader.Read(magic_number.data(), magic_number.size());
+  CHECK(magic_number == durability::kWalMagic) << "Wal magic number mismatch";
+
+  communication::bolt::Value dv;
+  decoder.ReadValue(&dv);
+  CHECK(dv.ValueInt() == durability::kVersion) << "Wal version mismatch";
+
   tx::TransactionId max_observed_tx_id{0};
   tx::TransactionId min_observed_tx_id{std::numeric_limits<uint64_t>::max()};
 

tests/unit/distributed_durability.cpp

@@ -157,6 +157,11 @@ void CheckDeltas(fs::path wal_dir, database::StateDelta::Type op) {
     communication::bolt::Decoder<HashedFileReader> decoder{reader};
     std::vector<database::StateDelta> deltas;
 
+    // check magic number
+    auto magic_number = durability::kWalMagic;
+    reader.Read(magic_number.data(), magic_number.size());
+    ASSERT_EQ(magic_number, durability::kWalMagic);
+
     // check version
     communication::bolt::Value dv;
     decoder.ReadValue(&dv);

tests/unit/durability.cpp

@@ -365,6 +365,11 @@ TEST_F(Durability, WalEncoding) {
   communication::bolt::Decoder<HashedFileReader> decoder{reader};
   std::vector<database::StateDelta> deltas;
 
+  // check magic number
+  auto magic_number = durability::kWalMagic;
+  reader.Read(magic_number.data(), magic_number.size());
+  ASSERT_EQ(magic_number, durability::kWalMagic);
+
   // check version
   communication::bolt::Value dv;
   decoder.ReadValue(&dv);
@@ -448,9 +453,9 @@ TEST_F(Durability, SnapshotEncoding) {
   ASSERT_EQ(vertex_count, 3);
   ASSERT_EQ(edge_count, 2);
 
-  auto magic_number = durability::kMagicNumber;
+  auto magic_number = durability::kSnapshotMagic;
   buffer.Read(magic_number.data(), magic_number.size());
-  ASSERT_EQ(magic_number, durability::kMagicNumber);
+  ASSERT_EQ(magic_number, durability::kSnapshotMagic);
 
   communication::bolt::Value dv;
   decoder.ReadValue(&dv);

tools/src/mg_import_csv/main.cpp

@@ -409,8 +409,8 @@ void Convert(const std::vector<std::string> &nodes,
     //   5) All nodes, sequentially, but not encoded as a list.
     //   6) All relationships, sequentially, but not encoded as a list.
     //   7) Summary with node count, relationship count and hash digest.
-    encoder.WriteRAW(durability::kMagicNumber.data(),
+    encoder.WriteRAW(durability::kSnapshotMagic.data(),
-                     durability::kMagicNumber.size());
+                     durability::kSnapshotMagic.size());
     encoder.WriteValue(durability::kVersion);
 
     encoder.WriteInt(0);  // Worker Id - for this use case it's okay to set to 0