llmq: Clean old DKG contributions up (#4763)

* llmq: Clean old DKG contributions up

* apply review suggestions and a bit more

* use scheduler

src/init.cpp

@@ -78,6 +78,7 @@
 #include <llmq/blockprocessor.h>
 #include <llmq/init.h>
 #include <llmq/quorums.h>
+#include <llmq/dkgsessionmgr.h>
 #include <llmq/signing.h>
 #include <llmq/snapshot.h>
 #include <llmq/utils.h>
@@ -2332,6 +2333,7 @@ bool AppInitMain(InitInterfaces& interfaces)
 
     if (fMasternodeMode) {
         scheduler.scheduleEvery(std::bind(&CCoinJoinServer::DoMaintenance, std::ref(coinJoinServer), std::ref(*g_connman)), 1 * 1000);
+        scheduler.scheduleEvery(std::bind(&llmq::CDKGSessionManager::CleanupOldContributions, std::ref(*llmq::quorumDKGSessionManager)), 60 * 60 * 1000);
     }
 
     if (gArgs.GetBoolArg("-statsenabled", DEFAULT_STATSD_ENABLE)) {

src/llmq/dkgsessionmgr.cpp

@@ -447,6 +447,52 @@ void CDKGSessionManager::CleanupCache() const
     }
 }
 
+void CDKGSessionManager::CleanupOldContributions() const
+{
+    if (db->IsEmpty()) {
+        return;
+    }
+
+    const auto prefixes = {DB_VVEC, DB_SKCONTRIB, DB_ENC_CONTRIB};
+
+    for (const auto& params : Params().GetConsensus().llmqs) {
+        // For how many blocks recent DKG info should be kept
+        const size_t MAX_STORE_DEPTH = 2 * params.signingActiveQuorumCount * params.dkgInterval;
+
+        LogPrint(BCLog::LLMQ, "CDKGSessionManager::%s -- looking for old entries for llmq type %d\n", __func__, uint8_t(params.type));
+
+        CDBBatch batch(*db);
+        size_t cnt_old{0}, cnt_all{0};
+        for (const auto& prefix : prefixes) {
+            std::unique_ptr<CDBIterator> pcursor(db->NewIterator());
+            auto start = std::make_tuple(prefix, params.type, uint256(), uint256());
+            decltype(start) k;
+
+            pcursor->Seek(start);
+            LOCK(cs_main);
+            while (pcursor->Valid()) {
+                if (!pcursor->GetKey(k) || std::get<0>(k) != prefix || std::get<1>(k) != params.type) {
+                    break;
+                }
+                cnt_all++;
+                const CBlockIndex* pindexQuorum = LookupBlockIndex(std::get<2>(k));
+                if (pindexQuorum == nullptr || ::ChainActive().Tip()->nHeight - pindexQuorum->nHeight > MAX_STORE_DEPTH) {
+                    // not found or too old
+                    batch.Erase(k);
+                    cnt_old++;
+                }
+                pcursor->Next();
+            }
+            pcursor.reset();
+        }
+        LogPrint(BCLog::LLMQ, "CDKGSessionManager::%s -- found %lld entries for llmq type %d\n", __func__, cnt_all, uint8_t(params.type));
+        if (cnt_old > 0) {
+            db->WriteBatch(batch);
+            LogPrint(BCLog::LLMQ, "CDKGSessionManager::%s -- removed %lld old entries for llmq type %d\n", __func__, cnt_old, uint8_t(params.type));
+        }
+    }
+}
+
 bool IsQuorumDKGEnabled()
 {
     return sporkManager.IsSporkActive(SPORK_17_QUORUM_DKG_ENABLED);

src/llmq/dkgsessionmgr.h

@@ -71,6 +71,8 @@ public:
     /// Read encrypted (unverified) DKG contributions for the member with the given proTxHash from the llmqDb
     bool GetEncryptedContributions(Consensus::LLMQType llmqType, const CBlockIndex* pQuorumBaseBlockIndex, const std::vector<bool>& validMembers, const uint256& proTxHash, std::vector<CBLSIESEncryptedObject<CBLSSecretKey>>& vecRet) const;
 
+    void CleanupOldContributions() const;
+
 private:
     void MigrateDKG();
     void CleanupCache() const;