Initialize sqlite_orm and start interacting with it

llarp/CMakeLists.txt

@@ -157,6 +157,7 @@ add_library(liblokinet
   path/pathset.cpp
   path/transit_hop.cpp
   peerstats/peer_db.cpp
+  peerstats/types.cpp
   pow.cpp
   profiling.cpp
   router/outbound_message_handler.cpp

llarp/peerstats/orm.hpp

@@ -0,0 +1,38 @@
+#pragma once
+
+#include <sqlite_orm/sqlite_orm.h>
+
+#include <peerstats/types.hpp>
+
+/// Contains some code to help deal with sqlite_orm in hopes of keeping other headers clean
+
+namespace llarp
+{
+  inline auto
+  initStorage(const std::string& file)
+  {
+    using namespace sqlite_orm;
+    return make_storage(
+        file,
+        make_table(
+            "peerstats",
+            make_column("routerId", &PeerStats::routerIdHex, primary_key(), unique()),
+            make_column("numConnectionAttempts", &PeerStats::numConnectionAttempts),
+            make_column("numConnectionSuccesses", &PeerStats::numConnectionSuccesses),
+            make_column("numConnectionRejections", &PeerStats::numConnectionRejections),
+            make_column("numConnectionTimeouts", &PeerStats::numConnectionTimeouts),
+            make_column("numPathBuilds", &PeerStats::numPathBuilds),
+            make_column("numPacketsAttempted", &PeerStats::numPacketsAttempted),
+            make_column("numPacketsSent", &PeerStats::numPacketsSent),
+            make_column("numPacketsDropped", &PeerStats::numPacketsDropped),
+            make_column("numPacketsResent", &PeerStats::numPacketsResent),
+            make_column("numDistinctRCsReceived", &PeerStats::numDistinctRCsReceived),
+            make_column("numLateRCs", &PeerStats::numLateRCs),
+            make_column("peakBandwidthBytesPerSec", &PeerStats::peakBandwidthBytesPerSec),
+            make_column("longestRCReceiveInterval", &PeerStats::longestRCReceiveIntervalMs),
+            make_column("mostExpiredRC", &PeerStats::mostExpiredRCMs)));
+  }
+
+  using PeerDbStorage = decltype(initStorage(""));
+
+}  // namespace llarp

llarp/peerstats/peer_db.cpp

@@ -1,65 +1,74 @@
 #include <peerstats/peer_db.hpp>
 
+#include <util/logging/logger.hpp>
+#include <util/str.hpp>
+
 namespace llarp
 {
-  PeerStats&
-  PeerStats::operator+=(const PeerStats& other)
+  void
+  PeerDb::loadDatabase(std::optional<std::filesystem::path> file)
   {
-    numConnectionAttempts += other.numConnectionAttempts;
-    numConnectionSuccesses += other.numConnectionSuccesses;
-    numConnectionRejections += other.numConnectionRejections;
-    numConnectionTimeouts += other.numConnectionTimeouts;
+    std::lock_guard gaurd(m_statsLock);
 
-    numPathBuilds += other.numPathBuilds;
-    numPacketsAttempted += other.numPacketsAttempted;
-    numPacketsSent += other.numPacketsSent;
-    numPacketsDropped += other.numPacketsDropped;
-    numPacketsResent += other.numPacketsResent;
+    m_peerStats.clear();
 
-    numDistinctRCsReceived += other.numDistinctRCsReceived;
-    numLateRCs += other.numLateRCs;
+    if (m_storage)
+      throw std::runtime_error("Reloading database not supported");  // TODO
 
-    peakBandwidthBytesPerSec = std::max(peakBandwidthBytesPerSec, other.peakBandwidthBytesPerSec);
-    longestRCReceiveInterval = std::max(longestRCReceiveInterval, other.longestRCReceiveInterval);
-    mostExpiredRC = std::max(mostExpiredRC, other.mostExpiredRC);
+    // sqlite_orm treats empty-string as an indicator to load a memory-backed database, which we'll
+    // use if file is an empty-optional
+    std::string fileString;
+    if (file.has_value())
+      fileString = file.value().native();
 
-    return *this;
+    m_storage = std::make_unique<PeerDbStorage>(initStorage(fileString));
   }
 
-  bool
-  PeerStats::operator==(const PeerStats& other)
+  void
+  PeerDb::flushDatabase()
   {
-    return numConnectionAttempts == other.numConnectionAttempts
-        and numConnectionSuccesses == other.numConnectionSuccesses
-        and numConnectionRejections == other.numConnectionRejections
-        and numConnectionTimeouts == other.numConnectionTimeouts
+    if (not m_storage)
+      throw std::runtime_error("Cannot flush database before it has been loaded");
+
+    decltype(m_peerStats) copy;
 
-        and numPathBuilds == other.numPathBuilds
-        and numPacketsAttempted == other.numPacketsAttempted
-        and numPacketsSent == other.numPacketsSent and numPacketsDropped == other.numPacketsDropped
-        and numPacketsResent == other.numPacketsResent
+    {
+      std::lock_guard gaurd(m_statsLock);
+      copy = m_peerStats;  // expensive deep copy
+    }
 
-        and numDistinctRCsReceived == other.numDistinctRCsReceived
-        and numLateRCs == other.numLateRCs
+    for (const auto& entry : m_peerStats)
+    {
+      // call me paranoid...
+      assert(not entry.second.routerIdHex.empty());
+      assert(entry.first.ToHex() == entry.second.routerIdHex);
 
-        and peakBandwidthBytesPerSec == peakBandwidthBytesPerSec
-        and longestRCReceiveInterval == longestRCReceiveInterval and mostExpiredRC == mostExpiredRC;
+      m_storage->insert(entry.second);
+    }
   }
 
   void
   PeerDb::accumulatePeerStats(const RouterID& routerId, const PeerStats& delta)
   {
+    if (routerId.ToHex() != delta.routerIdHex)
+      throw std::invalid_argument(
+          stringify("routerId ", routerId, " doesn't match ", delta.routerIdHex));
+
     std::lock_guard gaurd(m_statsLock);
-    m_peerStats[routerId] += delta;
+    auto itr = m_peerStats.find(routerId);
+    if (itr == m_peerStats.end())
+      itr = m_peerStats.insert({routerId, delta}).first;
+    else
+      itr->second += delta;
   }
 
-  PeerStats
+  std::optional<PeerStats>
   PeerDb::getCurrentPeerStats(const RouterID& routerId) const
   {
     std::lock_guard gaurd(m_statsLock);
     auto itr = m_peerStats.find(routerId);
     if (itr == m_peerStats.end())
-      return {};
+      return std::nullopt;
     else
       return itr->second;
   }

llarp/peerstats/peer_db.hpp

@@ -1,45 +1,46 @@
 #pragma once
 
 #include <chrono>
+#include <filesystem>
 #include <unordered_map>
 
 #include <sqlite_orm/sqlite_orm.h>
 
 #include <router_id.hpp>
 #include <util/time.hpp>
+#include <peerstats/types.hpp>
+#include <peerstats/orm.hpp>
 
 namespace llarp
 {
-  // Struct containing stats we know about a peer
-  struct PeerStats
+  /// Maintains a database of stats collected about the connections with our Service Node peers.
+  /// This uses a sqlite3 database behind the scenes as persistance, but this database is
+  /// periodically flushed to, meaning that it will become stale as PeerDb accumulates stats without
+  /// a flush.
+  struct PeerDb
   {
-    int32_t numConnectionAttempts = 0;
-    int32_t numConnectionSuccesses = 0;
-    int32_t numConnectionRejections = 0;
-    int32_t numConnectionTimeouts = 0;
-
-    int32_t numPathBuilds = 0;
-    int64_t numPacketsAttempted = 0;
-    int64_t numPacketsSent = 0;
-    int64_t numPacketsDropped = 0;
-    int64_t numPacketsResent = 0;
-
-    int64_t numDistinctRCsReceived = 0;
-    int64_t numLateRCs = 0;
-
-    double peakBandwidthBytesPerSec = 0;
-    std::chrono::milliseconds longestRCReceiveInterval = 0ms;
-    std::chrono::milliseconds mostExpiredRC = 0ms;
+    /// Loads the database from disk using the provided filepath. If the file is equal to
+    /// `std::nullopt`, the database will be loaded into memory (useful for testing).
+    ///
+    /// This must be called prior to calling flushDatabase(), and will truncate any existing data.
+    ///
+    /// This is a blocking call, both in the sense that it blocks on disk/database I/O and that it
+    /// will sit on a mutex while the database is loaded.
+    ///
+    /// @param file is an optional file which doesn't have to exist but must be writable, if a value
+    ///        is provided. If no value is provided, the database will be memory-backed.
+    /// @throws if sqlite_orm/sqlite3 is unable to open or create a database at the given file
+    void
+    loadDatabase(std::optional<std::filesystem::path> file);
 
-    PeerStats&
-    operator+=(const PeerStats& other);
-    bool
-    operator==(const PeerStats& other);
-  };
+    /// Flushes the database. Must be called after loadDatabase(). This call will block during I/O
+    /// and should be called in an appropriate threading context. However, it will make a temporary
+    /// copy of the peer stats so as to avoid sitting on a mutex lock during disk I/O.
+    ///
+    /// @throws if the database could not be written to (esp. if loadDatabase() has not been called)
+    void
+    flushDatabase();
 
-  /// Maintains a database of stats collected about the connections with our Service Node peers
-  struct PeerDb
-  {
     /// Add the given stats to the cummulative stats for the given peer. For cummulative stats, the
     /// stats are added together; for watermark stats, the max is kept.
     ///
@@ -54,16 +55,18 @@ namespace llarp
     accumulatePeerStats(const RouterID& routerId, const PeerStats& delta);
 
     /// Provides a snapshot of the most recent PeerStats we have for the given peer. If we don't
-    /// have any stats for the peer, an empty PeerStats is returned.
+    /// have any stats for the peer, std::nullopt
     ///
     /// @param routerId is the RouterID of the requested peer
     /// @return a copy of the most recent peer stats or an empty one if no such peer is known
-    PeerStats
+    std::optional<PeerStats>
     getCurrentPeerStats(const RouterID& routerId) const;
 
    private:
     std::unordered_map<RouterID, PeerStats, RouterID::Hash> m_peerStats;
     std::mutex m_statsLock;
+
+    std::unique_ptr<PeerDbStorage> m_storage;
   };
 
 }  // namespace llarp

llarp/peerstats/types.cpp

@@ -0,0 +1,56 @@
+#include <peerstats/types.hpp>
+
+namespace llarp
+{
+  PeerStats::PeerStats(const RouterID& routerId)
+  {
+    routerIdHex = routerId.ToHex();
+  }
+
+  PeerStats&
+  PeerStats::operator+=(const PeerStats& other)
+  {
+    numConnectionAttempts += other.numConnectionAttempts;
+    numConnectionSuccesses += other.numConnectionSuccesses;
+    numConnectionRejections += other.numConnectionRejections;
+    numConnectionTimeouts += other.numConnectionTimeouts;
+
+    numPathBuilds += other.numPathBuilds;
+    numPacketsAttempted += other.numPacketsAttempted;
+    numPacketsSent += other.numPacketsSent;
+    numPacketsDropped += other.numPacketsDropped;
+    numPacketsResent += other.numPacketsResent;
+
+    numDistinctRCsReceived += other.numDistinctRCsReceived;
+    numLateRCs += other.numLateRCs;
+
+    peakBandwidthBytesPerSec = std::max(peakBandwidthBytesPerSec, other.peakBandwidthBytesPerSec);
+    longestRCReceiveIntervalMs =
+        std::max(longestRCReceiveIntervalMs, other.longestRCReceiveIntervalMs);
+    mostExpiredRCMs = std::max(mostExpiredRCMs, other.mostExpiredRCMs);
+
+    return *this;
+  }
+
+  bool
+  PeerStats::operator==(const PeerStats& other)
+  {
+    return routerIdHex == other.routerIdHex and numConnectionAttempts == other.numConnectionAttempts
+        and numConnectionSuccesses == other.numConnectionSuccesses
+        and numConnectionRejections == other.numConnectionRejections
+        and numConnectionTimeouts == other.numConnectionTimeouts
+
+        and numPathBuilds == other.numPathBuilds
+        and numPacketsAttempted == other.numPacketsAttempted
+        and numPacketsSent == other.numPacketsSent and numPacketsDropped == other.numPacketsDropped
+        and numPacketsResent == other.numPacketsResent
+
+        and numDistinctRCsReceived == other.numDistinctRCsReceived
+        and numLateRCs == other.numLateRCs
+
+        and peakBandwidthBytesPerSec == other.peakBandwidthBytesPerSec
+        and longestRCReceiveIntervalMs == other.longestRCReceiveIntervalMs
+        and mostExpiredRCMs == other.mostExpiredRCMs;
+  }
+
+};  // namespace llarp

llarp/peerstats/types.hpp

@@ -0,0 +1,44 @@
+#pragma once
+
+#include <chrono>
+#include <unordered_map>
+
+#include <router_id.hpp>
+#include <util/time.hpp>
+
+/// Types stored in our peerstats database are declared here
+
+namespace llarp
+{
+  // Struct containing stats we know about a peer
+  struct PeerStats
+  {
+    std::string routerIdHex;
+
+    int32_t numConnectionAttempts = 0;
+    int32_t numConnectionSuccesses = 0;
+    int32_t numConnectionRejections = 0;
+    int32_t numConnectionTimeouts = 0;
+
+    int32_t numPathBuilds = 0;
+    int64_t numPacketsAttempted = 0;
+    int64_t numPacketsSent = 0;
+    int64_t numPacketsDropped = 0;
+    int64_t numPacketsResent = 0;
+
+    int32_t numDistinctRCsReceived = 0;
+    int32_t numLateRCs = 0;
+
+    double peakBandwidthBytesPerSec = 0;
+    int64_t longestRCReceiveIntervalMs = 0;
+    int64_t mostExpiredRCMs = 0;
+
+    PeerStats(const RouterID& routerId);
+
+    PeerStats&
+    operator+=(const PeerStats& other);
+    bool
+    operator==(const PeerStats& other);
+  };
+
+}  // namespace llarp

test/CMakeLists.txt

@@ -68,7 +68,8 @@ add_executable(catchAll
   util/test_llarp_util_printer.cpp
   util/test_llarp_util_str.cpp
   util/test_llarp_util_decaying_hashset.cpp
-  peerstats/peer_db.cpp
+  peerstats/test_peer_db.cpp
+  peerstats/test_peer_types.cpp
   config/test_llarp_config_definition.cpp
   config/test_llarp_config_output.cpp
   net/test_ip_address.cpp

test/peerstats/peer_db.cpp

@@ -1,46 +0,0 @@
-#include <numeric>
-#include <peerstats/peer_db.hpp>
-
-#include <catch2/catch.hpp>
-
-TEST_CASE("Test PeerStats operator+=", "[PeerStats]")
-{
-  // TODO: test all members
-  llarp::PeerStats stats;
-  stats.numConnectionAttempts = 1;
-  stats.peakBandwidthBytesPerSec = 12;
-
-  llarp::PeerStats delta;
-  delta.numConnectionAttempts = 2;
-  delta.peakBandwidthBytesPerSec = 4;
-
-  stats += delta;
-
-  CHECK(stats.numConnectionAttempts == 3);
-  CHECK(stats.peakBandwidthBytesPerSec == 12);  // should take max(), not add
-}
-
-TEST_CASE("Test PeerDb PeerStats memory storage", "[PeerDb]")
-{
-  const llarp::PeerStats empty = {};
-  const llarp::RouterID id = {};
-
-  llarp::PeerDb db;
-  CHECK(db.getCurrentPeerStats(id) == empty);
-
-  llarp::PeerStats delta;
-  delta.numConnectionAttempts = 4;
-  delta.peakBandwidthBytesPerSec = 5;
-  db.accumulatePeerStats(id, delta);
-  CHECK(db.getCurrentPeerStats(id) == delta);
-
-  delta = {};
-  delta.numConnectionAttempts = 5;
-  delta.peakBandwidthBytesPerSec = 6;
-  db.accumulatePeerStats(id, delta);
-
-  llarp::PeerStats expected;
-  expected.numConnectionAttempts = 9;
-  expected.peakBandwidthBytesPerSec = 6;
-  CHECK(db.getCurrentPeerStats(id) == expected);
-}

test/peerstats/test_peer_db.cpp

@@ -0,0 +1,75 @@
+#include <numeric>
+#include <peerstats/peer_db.hpp>
+
+#include <catch2/catch.hpp>
+
+TEST_CASE("Test PeerDb PeerStats memory storage", "[PeerDb]")
+{
+  const llarp::RouterID id = {};
+  const llarp::PeerStats empty(id);
+
+  llarp::PeerDb db;
+  CHECK(db.getCurrentPeerStats(id).has_value() == false);
+
+  llarp::PeerStats delta(id);
+  delta.numConnectionAttempts = 4;
+  delta.peakBandwidthBytesPerSec = 5;
+  db.accumulatePeerStats(id, delta);
+  CHECK(db.getCurrentPeerStats(id).value() == delta);
+
+  delta = llarp::PeerStats(id);
+  delta.numConnectionAttempts = 5;
+  delta.peakBandwidthBytesPerSec = 6;
+  db.accumulatePeerStats(id, delta);
+
+  llarp::PeerStats expected(id);
+  expected.numConnectionAttempts = 9;
+  expected.peakBandwidthBytesPerSec = 6;
+  CHECK(db.getCurrentPeerStats(id).value() == expected);
+}
+
+TEST_CASE("Test PeerDb flush before load", "[PeerDb]")
+{
+  llarp::PeerDb db;
+  CHECK_THROWS_WITH(db.flushDatabase(), "Cannot flush database before it has been loaded");
+}
+
+TEST_CASE("Test PeerDb load twice", "[PeerDb]")
+{
+  llarp::PeerDb db;
+  CHECK_NOTHROW(db.loadDatabase(std::nullopt));
+  CHECK_THROWS_WITH(db.loadDatabase(std::nullopt), "Reloading database not supported");
+}
+
+TEST_CASE("Test PeerDb nukes stats on load", "[PeerDb]")
+{
+  const llarp::RouterID id = {};
+
+  llarp::PeerDb db;
+
+  llarp::PeerStats stats(id);
+  stats.numConnectionAttempts = 1;
+
+  db.accumulatePeerStats(id, stats);
+  CHECK(db.getCurrentPeerStats(id).value() == stats);
+
+  db.loadDatabase(std::nullopt);
+
+  CHECK(db.getCurrentPeerStats(id).has_value() == false);
+}
+
+/*
+TEST_CASE("Test file-backed database", "[PeerDb]")
+{
+  llarp::PeerDb db;
+  db.loadDatabase(std::nullopt);
+
+  const llarp::RouterID id = {};
+  llarp::PeerStats stats(id);
+  stats.numConnectionAttempts = 42;
+
+  db.accumulatePeerStats(id, stats);
+
+  db.flushDatabase();
+}
+*/

test/peerstats/test_peer_types.cpp

@@ -0,0 +1,23 @@
+#include <numeric>
+#include <peerstats/types.hpp>
+
+#include <catch2/catch.hpp>
+
+TEST_CASE("Test PeerStats operator+=", "[PeerStats]")
+{
+  llarp::RouterID id = {};
+
+  // TODO: test all members
+  llarp::PeerStats stats(id);
+  stats.numConnectionAttempts = 1;
+  stats.peakBandwidthBytesPerSec = 12;
+
+  llarp::PeerStats delta(id);
+  delta.numConnectionAttempts = 2;
+  delta.peakBandwidthBytesPerSec = 4;
+
+  stats += delta;
+
+  CHECK(stats.numConnectionAttempts == 3);
+  CHECK(stats.peakBandwidthBytesPerSec == 12);  // should take max(), not add
+}