Join from an existing snapshot

CMakeLists.txt

@@ -535,6 +535,23 @@ if(BUILD_TESTS)
       4000
   )
 
+  add_e2e_test(
+    NAME snapshots_test_suite
+    PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/e2e_suite.py
+    CONSENSUS raft
+    LABEL suite
+    ADDITIONAL_ARGS
+      --test-duration
+      150
+      --enforce-reqs
+      --test-suite
+      snapshots
+      --raft-election-timeout
+      4000
+      --snapshot-tx-interval
+      5
+  )
+
   add_e2e_test(
     NAME full_test_suite
     PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/e2e_suite.py
@@ -571,13 +588,14 @@ if(BUILD_TESTS)
       NAME reconfiguration_test
       PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/reconfiguration.py
       CONSENSUS raft
+      ADDITIONAL_ARGS --raft-election-timeout 4000
     )
 
     add_e2e_test(
       NAME reconfiguration_snapshot_test
       PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/reconfiguration.py
       CONSENSUS raft
-      ADDITIONAL_ARGS --snapshot-tx-interval 10
+      ADDITIONAL_ARGS --snapshot-tx-interval 10 --raft-election-timeout 4000
     )
 
     add_e2e_test(

src/consensus/ledger_enclave.h

@@ -114,5 +114,15 @@ namespace consensus
     {
       RINGBUFFER_WRITE_MESSAGE(consensus::ledger_commit, to_host, idx);
     }
+
+    /**
+     * Initialise ledger at a given index (e.g. after a snapshot)
+     *
+     * @param idx Index to start ledger from
+     */
+    void init(Index idx)
+    {
+      RINGBUFFER_WRITE_MESSAGE(consensus::ledger_init, to_host, idx);
+    }
   };
 }

src/consensus/ledger_enclave_types.h

@@ -28,6 +28,7 @@ namespace consensus
     DEFINE_RINGBUFFER_MSG_TYPE(ledger_append),
     DEFINE_RINGBUFFER_MSG_TYPE(ledger_truncate),
     DEFINE_RINGBUFFER_MSG_TYPE(ledger_commit),
+    DEFINE_RINGBUFFER_MSG_TYPE(ledger_init),
 
     /// Create a new snapshot. Enclave -> Host
     DEFINE_RINGBUFFER_MSG_TYPE(ledger_snapshot),
@@ -45,6 +46,7 @@ DECLARE_RINGBUFFER_MESSAGE_PAYLOAD(
   consensus::ledger_no_entry,
   consensus::Index,
   consensus::LedgerRequestPurpose);
+DECLARE_RINGBUFFER_MESSAGE_PAYLOAD(consensus::ledger_init, consensus::Index);
 DECLARE_RINGBUFFER_MESSAGE_PAYLOAD(
   consensus::ledger_append,
   bool /* committable */,

src/consensus/raft/raft.h

@@ -28,7 +28,7 @@ namespace raft
     std::vector<Index> terms;
 
   public:
-    static constexpr Term InvalidTerm = 0;
+    static constexpr Term InvalidTerm = ccf::VIEW_UNKNOWN;
 
     void initialise(const std::vector<Index>& terms_)
     {
@@ -56,7 +56,9 @@ namespace raft
       }
 
       for (int64_t i = terms.size(); i < term; ++i)
+      {
         terms.push_back(idx);
+      }
       LOG_DEBUG_FMT("Resulting terms: {}", fmt::join(terms, ", "));
     }
 
@@ -66,7 +68,9 @@ namespace raft
 
       // Indices before the index of the first term are unknown
       if (it == terms.begin())
+      {
         return InvalidTerm;
+      }
 
       return (it - terms.begin());
     }
@@ -271,6 +275,23 @@ namespace raft
       become_leader();
     }
 
+    void init_as_follower(Index index, Term term)
+    {
+      // This should only be called when the node resumes from a snapshot and
+      // before it has received any append entries.
+      std::lock_guard<SpinLock> guard(lock);
+
+      last_idx = index;
+      commit_idx = index;
+
+      term_history.update(index, term);
+
+      ledger->init(index);
+      snapshotter->set_last_snapshot_idx(index);
+
+      become_follower(term);
+    }
+
     Index get_last_idx()
     {
       return last_idx;

src/consensus/raft/raft_consensus.h

@@ -54,6 +54,11 @@ namespace raft
       raft->force_become_leader(seqno, view, terms, commit_seqno);
     }
 
+    void init_as_backup(SeqNo seqno, View view) override
+    {
+      raft->init_as_follower(seqno, view);
+    }
+
     bool replicate(const kv::BatchVector& entries, View view) override
     {
       return raft->replicate(entries, view);

src/consensus/raft/test/main.cpp

@@ -160,7 +160,7 @@ DOCTEST_TEST_CASE(
   auto rvc = get<1>(rv);
   DOCTEST_REQUIRE(rvc.term == 1);
   DOCTEST_REQUIRE(rvc.last_commit_idx == 0);
-  DOCTEST_REQUIRE(rvc.last_commit_term == 0);
+  DOCTEST_REQUIRE(rvc.last_commit_term == raft::TermHistory::InvalidTerm);
 
   r1.recv_message(reinterpret_cast<uint8_t*>(&rvc), sizeof(rvc));
 
@@ -172,7 +172,7 @@ DOCTEST_TEST_CASE(
   rvc = get<1>(rv);
   DOCTEST_REQUIRE(rvc.term == 1);
   DOCTEST_REQUIRE(rvc.last_commit_idx == 0);
-  DOCTEST_REQUIRE(rvc.last_commit_term == 0);
+  DOCTEST_REQUIRE(rvc.last_commit_term == raft::TermHistory::InvalidTerm);
 
   r2.recv_message(reinterpret_cast<uint8_t*>(&rvc), sizeof(rvc));
 
@@ -216,7 +216,7 @@ DOCTEST_TEST_CASE(
   DOCTEST_REQUIRE(aec.idx == 0);
   DOCTEST_REQUIRE(aec.term == 1);
   DOCTEST_REQUIRE(aec.prev_idx == 0);
-  DOCTEST_REQUIRE(aec.prev_term == 0);
+  DOCTEST_REQUIRE(aec.prev_term == raft::TermHistory::InvalidTerm);
   DOCTEST_REQUIRE(aec.leader_commit_idx == 0);
 
   ae = r0.channels->sent_append_entries.front();
@@ -226,7 +226,7 @@ DOCTEST_TEST_CASE(
   DOCTEST_REQUIRE(aec.idx == 0);
   DOCTEST_REQUIRE(aec.term == 1);
   DOCTEST_REQUIRE(aec.prev_idx == 0);
-  DOCTEST_REQUIRE(aec.prev_term == 0);
+  DOCTEST_REQUIRE(aec.prev_term == raft::TermHistory::InvalidTerm);
   DOCTEST_REQUIRE(aec.leader_commit_idx == 0);
 }
 
@@ -374,7 +374,7 @@ DOCTEST_TEST_CASE(
         DOCTEST_REQUIRE(msg.idx == 1);
         DOCTEST_REQUIRE(msg.term == 1);
         DOCTEST_REQUIRE(msg.prev_idx == 0);
-        DOCTEST_REQUIRE(msg.prev_term == 0);
+        DOCTEST_REQUIRE(msg.prev_term == raft::TermHistory::InvalidTerm);
         DOCTEST_REQUIRE(msg.leader_commit_idx == 0);
       }));
 
@@ -472,7 +472,7 @@ DOCTEST_TEST_CASE("Multiple nodes late join" * doctest::test_suite("multiple"))
         DOCTEST_REQUIRE(msg.idx == 1);
         DOCTEST_REQUIRE(msg.term == 1);
         DOCTEST_REQUIRE(msg.prev_idx == 0);
-        DOCTEST_REQUIRE(msg.prev_term == 0);
+        DOCTEST_REQUIRE(msg.prev_term == raft::TermHistory::InvalidTerm);
         DOCTEST_REQUIRE(msg.leader_commit_idx == 0);
       }));
 

src/enclave/enclave.h

@@ -319,7 +319,7 @@ namespace enclave
 
         if (start_type == StartType::Join)
         {
-          node->join({ccf_config});
+          node->join(ccf_config);
         }
         else if (start_type == StartType::Recover)
         {

src/enclave/interface.h

@@ -62,7 +62,9 @@ struct CCFConfig
     std::string target_port;
     std::vector<uint8_t> network_cert;
     size_t join_timer;
-    MSGPACK_DEFINE(target_host, target_port, network_cert, join_timer);
+    std::vector<uint8_t> snapshot;
+    MSGPACK_DEFINE(
+      target_host, target_port, network_cert, join_timer, snapshot);
   };
   Joining joining = {};
 

src/host/ledger.h

@@ -647,6 +647,11 @@ namespace asynchost
 
     Ledger(const Ledger& that) = delete;
 
+    void init_idx(size_t idx)
+    {
+      last_idx = idx;
+    }
+
     std::optional<std::vector<uint8_t>> read_entry(size_t idx)
     {
       auto f = get_file_from_idx(idx);
@@ -796,6 +801,12 @@ namespace asynchost
     void register_message_handlers(
       messaging::Dispatcher<ringbuffer::Message>& disp)
     {
+      DISPATCHER_SET_MESSAGE_HANDLER(
+        disp, consensus::ledger_init, [this](const uint8_t* data, size_t size) {
+          auto idx = serialized::read<consensus::Index>(data, size);
+          init_idx(idx);
+        });
+
       DISPATCHER_SET_MESSAGE_HANDLER(
         disp,
         consensus::ledger_append,

src/host/main.cpp

@@ -134,7 +134,11 @@ int main(int argc, char** argv)
     "--rpc-address)");
 
   std::string ledger_dir("ledger");
-  app.add_option("--ledger-dir", ledger_dir, "Ledger and snapshots directory")
+  app.add_option("--ledger-dir", ledger_dir, "Ledger directory")
+    ->capture_default_str();
+
+  std::string snapshot_dir("snapshots");
+  app.add_option("--snapshot-dir", snapshot_dir, "Snapshots directory")
     ->capture_default_str();
 
   size_t ledger_chunk_bytes = 5'000'000;
@@ -354,7 +358,7 @@ int main(int argc, char** argv)
     "key)")
     ->required();
 
-  std::optional<size_t> recovery_threshold;
+  std::optional<size_t> recovery_threshold = std::nullopt;
   start
     ->add_option(
       "--recovery-threshold",
@@ -438,7 +442,8 @@ int main(int argc, char** argv)
     if ((*start || *join) && files::exists(ledger_dir))
     {
       throw std::logic_error(fmt::format(
-        "On start/join, ledger directory should not exist ({})", ledger_dir));
+        "On start and join, ledger directory should not exist ({})",
+        ledger_dir));
     }
     else if (*recover && !files::exists(ledger_dir))
     {
@@ -544,8 +549,8 @@ int main(int argc, char** argv)
   asynchost::Ledger ledger(ledger_dir, writer_factory, ledger_chunk_bytes);
   ledger.register_message_handlers(bp.get_dispatcher());
 
-  asynchost::SnapshotManager snapshot(ledger_dir);
-  snapshot.register_message_handlers(bp.get_dispatcher());
+  asynchost::SnapshotManager snapshots(snapshot_dir);
+  snapshots.register_message_handlers(bp.get_dispatcher());
 
   // Begin listening for node-to-node and RPC messages.
   // This includes DNS resolution and potentially dynamic port assignment (if
@@ -634,6 +639,21 @@ int main(int argc, char** argv)
     ccf_config.joining.target_port = target_rpc_address.port;
     ccf_config.joining.network_cert = files::slurp(network_cert_file);
     ccf_config.joining.join_timer = join_timer;
+
+    auto snapshot_file = snapshots.find_latest_snapshot();
+    if (snapshot_file.has_value())
+    {
+      ccf_config.joining.snapshot = files::slurp(snapshot_file.value());
+      LOG_INFO_FMT(
+        "Found latest snapshot file: {} (size: {})",
+        snapshot_file.value(),
+        ccf_config.joining.snapshot.size());
+    }
+    else
+    {
+      LOG_INFO_FMT(
+        "No snapshot found, node will request transactions from the beginning");
+    }
   }
   else if (*recover)
   {

src/host/snapshot.h

@@ -7,6 +7,7 @@
 #include <filesystem>
 #include <fstream>
 #include <iostream>
+#include <optional>
 
 namespace fs = std::filesystem;
 
@@ -49,7 +50,7 @@ namespace asynchost
       if (fs::is_directory(snapshot_dir))
       {
         LOG_INFO_FMT(
-          "Snapshots will be stored in existing directory {}", snapshot_dir);
+          "Snapshots will be stored in existing directory: {}", snapshot_dir);
       }
       else if (!fs::create_directory(snapshot_dir))
       {
@@ -58,6 +59,36 @@ namespace asynchost
       }
     }
 
+    std::optional<std::string> find_latest_snapshot()
+    {
+      std::optional<std::string> snapshot_file = std::nullopt;
+      size_t latest_idx = 0;
+
+      for (auto& f : fs::directory_iterator(snapshot_dir))
+      {
+        auto file_name = f.path().filename().string();
+        auto pos = file_name.find(fmt::format("{}.", snapshot_file_prefix));
+        if (pos == std::string::npos)
+        {
+          LOG_FAIL_FMT(
+            "Ignoring \"{}\" because it does not start with {}",
+            file_name,
+            snapshot_file_prefix);
+          continue;
+        }
+
+        pos = file_name.find(".");
+        size_t snapshot_idx = std::stol(file_name.substr(pos + 1));
+        if (snapshot_idx > latest_idx)
+        {
+          snapshot_file = f.path().string();
+          latest_idx = snapshot_idx;
+        }
+      }
+
+      return snapshot_file;
+    }
+
     void register_message_handlers(
       messaging::Dispatcher<ringbuffer::Message>& disp)
     {
@@ -71,4 +102,4 @@ namespace asynchost
         });
     }
   };
-}
+}

src/kv/kv_types.h

@@ -266,6 +266,11 @@ namespace kv
       state = Primary;
     }
 
+    virtual void init_as_backup(SeqNo, View)
+    {
+      state = Backup;
+    }
+
     virtual bool replicate(const BatchVector& entries, View view) = 0;
     virtual std::pair<View, SeqNo> get_committed_txid() = 0;
 

src/node/call_types.h

@@ -19,12 +19,4 @@ namespace ccf
       tls::Pem network_enc_pubk;
     };
   };
-
-  struct Join
-  {
-    struct In
-    {
-      CCFConfig config;
-    };
-  };
 }