Creates archiving for reward batching

This adds a new table to the batching schema to copy the accrued
balances every so often. This means that we can jump back to a
previous point without popping blocks.

The archiving is triggered in sql every 100 blocks and added to the
archive table, then pruned from the archive table at a later time to
ensure the size is kept small. Rebuilding 100 blocks is pretty
reasonable and should be less than 10s.

For longer distance pop_blocks and blockchain detached every 10k blocks
is kept in the archiving table. This takes longer to rebuild but is
better than rebuilding from scratch.

The blockchain detached function is also added to our regular blockchain
detached hooks so that it gets called every time the blockchain
detaches. Which appears to have caused some issues previously when some
of the modules would detach but batching would be stuck in an advanced
state.

src/blockchain_db/sqlite/db_sqlite.cpp

@@ -170,6 +170,43 @@ namespace cryptonote {
 
       transaction.commit();
     }
+
+    const auto archive_table_count = prepared_get<int64_t>("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='batched_payments_accrued_archive';");
+    if(archive_table_count == 0)
+    {
+      log::info(logcat, "Adding archiving to batching db");
+      auto& netconf = get_config(m_nettype);
+      SQLite::Transaction transaction{
+        db,
+        SQLite::TransactionBehavior::IMMEDIATE
+      };
+      db.exec(fmt::format(R"(
+        CREATE TABLE batched_payments_accrued_archive(
+          address VARCHAR NOT NULL,
+          amount BIGINT NOT NULL,
+          payout_offset INTEGER NOT NULL,
+          archive_height BIGINT NOT NULL,
+          CHECK(amount >= 0),
+          CHECK(archive_height >= 0)
+        );
+
+        CREATE INDEX batched_payments_accrued_archive_height_idx ON batched_payments_accrued_archive(archive_height);
+
+        DROP TRIGGER IF EXISTS make_archive;
+        CREATE TRIGGER make_archive AFTER UPDATE ON batch_db_info
+        FOR EACH ROW WHEN (NEW.height % 100) = 0 AND NEW.height > OLD.height BEGIN
+            INSERT INTO batched_payments_accrued_archive SELECT *, NEW.height FROM batched_payments_accrued;
+            DELETE FROM batched_payments_accrued_archive WHERE archive_height < NEW.height - {1} AND archive_height % {0} != 0;
+        END;
+
+        DROP TRIGGER IF EXISTS clear_archive;
+        CREATE TRIGGER clear_archive AFTER UPDATE ON batch_db_info
+        FOR EACH ROW WHEN NEW.height < OLD.height BEGIN
+            DELETE FROM batched_payments_accrued_archive WHERE archive_height >= NEW.height;
+        END;
+        )", netconf.STORE_LONG_TERM_STATE_INTERVAL, 500));
+      transaction.commit();
+    }
   }
 
   void BlockchainSQLite::reset_database() {
@@ -178,6 +215,8 @@ namespace cryptonote {
     db.exec(R"(
       DROP TABLE IF EXISTS batched_payments_accrued;
 
+      DROP TABLE IF EXISTS batched_payments_accrued_archive;
+
       DROP VIEW IF EXISTS batched_payments_paid;
 
       DROP TABLE IF EXISTS batched_payments_raw;
@@ -186,7 +225,7 @@ namespace cryptonote {
     )");
 
     create_schema();
-
+    upgrade_schema();
     log::debug(logcat, "Database reset complete");
   }
 
@@ -208,6 +247,39 @@ namespace cryptonote {
     update_height(height - 1);
   }
 
+  void BlockchainSQLite::blockchain_detached(uint64_t new_height)
+  {
+    if (height < new_height)
+      return;
+    int64_t revert_to_height = new_height - 1;
+    auto maybe_prev_interval = prepared_maybe_get<int64_t>(
+        "SELECT DISTINCT archive_height FROM batched_payments_accrued_archive WHERE archive_height <= ? ORDER BY archive_height DESC LIMIT 1",
+        revert_to_height);
+
+    if (!maybe_prev_interval)
+    {
+      auto fork_height = cryptonote::get_hard_fork_heights(m_nettype, hf::hf19_reward_batching);
+      reset_database();
+      update_height(fork_height.first.value_or(0));
+      return;
+    }
+    const auto prev_interval = *maybe_prev_interval;
+
+    db.exec(fmt::format(R"(
+      DELETE FROM batched_payments_raw WHERE height_paid > {0};
+
+      DELETE FROM batched_payments_accrued;
+
+      INSERT INTO batched_payments_accrued
+        SELECT address, amount, payout_offset
+        FROM batched_payments_accrued_archive WHERE archive_height = {0};
+
+      DELETE FROM batched_payments_accrued_archive WHERE archive_height >= {0};
+      )", prev_interval));
+    update_height(prev_interval);
+    return;
+  }
+
   // Must be called with the address_str_cache_mutex held!
   const std::string& BlockchainSQLite::get_address_str(const account_public_address& addr)
   {
@@ -502,6 +574,7 @@ namespace cryptonote {
         SQLite::TransactionBehavior::IMMEDIATE
       };
 
+
       if (!reward_handler(block, service_nodes_state, /*add=*/ false))
         return false;
 

src/blockchain_db/sqlite/db_sqlite.h

@@ -58,6 +58,7 @@ public:
   void increment_height();
   void decrement_height();
 
+  void blockchain_detached(uint64_t new_height);
 
   // add_sn_payments/subtract_sn_payments -> passing an array of addresses and amounts. These will be added or subtracted to the database for each address specified. If the address does not exist it will be created.
   bool add_sn_rewards(const std::vector<cryptonote::batch_sn_payment>& payments);

src/cryptonote_config.h

@@ -383,6 +383,9 @@ namespace config
   // If a node has been online for this amount of blocks they will receive SN rewards
   inline constexpr uint64_t SERVICE_NODE_PAYABLE_AFTER_BLOCKS = 720;
 
+  // batching and SNL will save the state every STORE_LONG_TERM_STATE_INTERVAL blocks
+  inline constexpr uint64_t STORE_LONG_TERM_STATE_INTERVAL = 10000;
+
   namespace testnet
   {
     inline constexpr uint64_t HEIGHT_ESTIMATE_HEIGHT = 339767;
@@ -482,6 +485,8 @@ struct network_config
 
   uint64_t HARDFORK_DEREGISTRATION_GRACE_PERIOD;
 
+  uint64_t STORE_LONG_TERM_STATE_INTERVAL;
+
 
   inline constexpr std::string_view governance_wallet_address(hf hard_fork_version) const {
     const auto wallet_switch =
@@ -517,6 +522,7 @@ inline constexpr network_config mainnet_config{
   config::LIMIT_BATCH_OUTPUTS,
   config::SERVICE_NODE_PAYABLE_AFTER_BLOCKS,
   config::HARDFORK_DEREGISTRATION_GRACE_PERIOD,
+  config::STORE_LONG_TERM_STATE_INTERVAL,
 };
 inline constexpr network_config testnet_config{
   network_type::TESTNET,
@@ -544,6 +550,7 @@ inline constexpr network_config testnet_config{
   config::LIMIT_BATCH_OUTPUTS,
   config::testnet::SERVICE_NODE_PAYABLE_AFTER_BLOCKS,
   config::HARDFORK_DEREGISTRATION_GRACE_PERIOD,
+  config::STORE_LONG_TERM_STATE_INTERVAL,
 };
 inline constexpr network_config devnet_config{
   network_type::DEVNET,
@@ -598,6 +605,7 @@ inline constexpr network_config fakenet_config{
   config::LIMIT_BATCH_OUTPUTS,
   config::testnet::SERVICE_NODE_PAYABLE_AFTER_BLOCKS,
   config::HARDFORK_DEREGISTRATION_GRACE_PERIOD,
+  config::STORE_LONG_TERM_STATE_INTERVAL,
 };
 
 inline constexpr const network_config& get_config(network_type nettype)

src/cryptonote_core/blockchain.cpp

@@ -724,8 +724,6 @@ void Blockchain::pop_blocks(uint64_t nblocks)
   detached_info hook_data{m_db->height(), /*by_pop_blocks=*/true};
   for (const auto& hook : m_blockchain_detached_hooks)
     hook(hook_data);
-  if (!pop_batching_rewards)
-    m_service_node_list.reset_batching_to_latest_height();
   load_missing_blocks_into_oxen_subsystems();
 
   if (stop_batch)

src/cryptonote_core/cryptonote_core.cpp

@@ -774,6 +774,8 @@ namespace cryptonote
     m_blockchain_storage.hook_init([this] { m_service_node_list.init(); });
     m_blockchain_storage.hook_validate_miner_tx([this] (const auto& info) { m_service_node_list.validate_miner_tx(info); });
     m_blockchain_storage.hook_alt_block_add([this] (const auto& info) { m_service_node_list.alt_block_add(info); });
+    
+    m_blockchain_storage.hook_blockchain_detached([this] (const auto& info) { m_blockchain_storage.sqlite_db()->blockchain_detached(info.height); });
 
     // NOTE: There is an implicit dependency on service node lists being hooked first!
     m_blockchain_storage.hook_init([this] { m_quorum_cop.init(); });

src/cryptonote_core/service_node_list.cpp

@@ -1734,18 +1734,13 @@ namespace service_nodes
     }
   }
 
-  void service_node_list::reset_batching_to_latest_height()
-  {
-    m_blockchain.sqlite_db()->reset_database();
-    m_blockchain.sqlite_db()->update_height(0);
-  }
-
   bool service_node_list::state_history_exists(uint64_t height)
   {
     auto it = m_transient.state_history.find(height);
     return it != m_transient.state_history.end();
   }
 
+
   bool service_node_list::process_batching_rewards(const cryptonote::block& block)
   {
     uint64_t block_height = cryptonote::get_block_height(block);

src/cryptonote_core/service_node_list.h

@@ -424,7 +424,6 @@ namespace service_nodes
     service_node_list &operator=(const service_node_list &) = delete;
 
     void block_add(const cryptonote::block& block, const std::vector<cryptonote::transaction>& txs, const cryptonote::checkpoint_t* checkpoint);
-    void reset_batching_to_latest_height();
     bool state_history_exists(uint64_t height);
     bool process_batching_rewards(const cryptonote::block& block);
     bool pop_batching_rewards_block(const cryptonote::block& block);