Add delays when reading past payment events for stats (#3288)

diff --git a/eclair-core/src/main/resources/reference.conf b/eclair-core/src/main/resources/reference.conf
index e452e9b..7b03f01 100644
--- a/eclair-core/src/main/resources/reference.conf
+++ b/eclair-core/src/main/resources/reference.conf
@@ -671,6 +671,14 @@ eclair {
     top-peers-count = 10
     // A list of node_ids with whom we will try to maintain liquidity.
     top-peers-whitelist = []
+    // On restart, we read all past events from the previous week from the DB, which is expensive.
+    // We do this in 56 3-hour chunks to avoid performance issues, with a delay between each chunk.
+    // We only read the first chunk after an initial delay, since this isn't critical and there are already a lot of DB
+    // reads when restarting an eclair node that have higher priority.
+    past-events {
+      init-delay = 10 minutes
+      chunk-delay = 10 seconds
+    }
     // We can automatically allocate liquidity to our top peers when necessary.
     liquidity {
       // If true, we will automatically fund channels.
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/NodeParams.scala b/eclair-core/src/main/scala/fr/acinq/eclair/NodeParams.scala
index 9e37af0..f7b0e3d 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/NodeParams.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/NodeParams.scala
@@ -33,7 +33,7 @@ import fr.acinq.eclair.message.OnionMessages.OnionMessageConfig
 import fr.acinq.eclair.payment.offer.OffersConfig
 import fr.acinq.eclair.payment.relay.OnTheFlyFunding
 import fr.acinq.eclair.payment.relay.Relayer.{AsyncPaymentsParams, RelayFees, RelayParams}
-import fr.acinq.eclair.profit.PeerScorer
+import fr.acinq.eclair.profit.{PeerScorer, PeerStatsTracker}
 import fr.acinq.eclair.reputation.Reputation
 import fr.acinq.eclair.router.Announcements.AddressException
 import fr.acinq.eclair.router.Graph.HeuristicsConstants
@@ -97,6 +97,7 @@ case class NodeParams(nodeKeyManager: NodeKeyManager,
                       onTheFlyFundingConfig: OnTheFlyFunding.Config,
                       peerStorageConfig: PeerStorageConfig,
                       peerScoringConfig: PeerScorer.Config,
+                      peerStatsTrackerConfig: PeerStatsTracker.Config,
                       offersConfig: OffersConfig) {
   val privateKey: Crypto.PrivateKey = nodeKeyManager.nodeKey.privateKey
 
@@ -750,6 +751,10 @@ object NodeParams extends Logging {
           dailyPaymentVolumeThresholdPercent = config.getDouble("peer-scoring.relay-fees.daily-payment-volume-threshold-percent"),
         )
       ),
+      peerStatsTrackerConfig = PeerStatsTracker.Config(
+        pastEventsInitDelay = FiniteDuration(config.getDuration("peer-scoring.past-events.init-delay").getSeconds, TimeUnit.SECONDS),
+        pastEventsChunkDelay = FiniteDuration(config.getDuration("peer-scoring.past-events.chunk-delay").getSeconds, TimeUnit.SECONDS),
+      ),
       offersConfig = OffersConfig(
         messagePathMinLength = config.getInt("offers.message-path-min-length"),
         paymentPathCount = config.getInt("offers.payment-path-count"),
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/Setup.scala b/eclair-core/src/main/scala/fr/acinq/eclair/Setup.scala
index 771c11c..70cbe8b 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/Setup.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/Setup.scala
@@ -403,7 +403,7 @@ class Setup(val datadir: File,
       balanceActor = system.spawn(BalanceActor(bitcoinClient, nodeParams.channelConf.minDepth, channelsListener, nodeParams.balanceCheckInterval), name = "balance-actor")
       postman = system.spawn(Behaviors.supervise(Postman(nodeParams, switchboard, router.toTyped, register, offerManager)).onFailure(typed.SupervisorStrategy.restart), name = "postman")
       peerScorer_opt = if (nodeParams.peerScoringConfig.enabled) {
-        val statsTracker = system.spawn(Behaviors.supervise(PeerStatsTracker(nodeParams.db.audit, channels)).onFailure(typed.SupervisorStrategy.restart), name = "peer-stats-tracker")
+        val statsTracker = system.spawn(Behaviors.supervise(PeerStatsTracker(nodeParams.peerStatsTrackerConfig, nodeParams.db.audit, channels)).onFailure(typed.SupervisorStrategy.restart), name = "peer-stats-tracker")
         Some(system.spawn(Behaviors.supervise(PeerScorer(nodeParams, bitcoinClient, statsTracker, register)).onFailure(typed.SupervisorStrategy.restart), name = "peer-scorer"))
       } else None
 
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerStatsTracker.scala b/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerStatsTracker.scala
index 91c574d..ebde194 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerStatsTracker.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerStatsTracker.scala
@@ -35,6 +35,7 @@ import scala.concurrent.duration.{DurationInt, FiniteDuration}
  */
 
 object PeerStatsTracker {
+
   // @formatter:off
   sealed trait Command
   case class GetLatestStats(replyTo: ActorRef[LatestStats]) extends Command
@@ -50,10 +51,12 @@ object PeerStatsTracker {
   private[profit] case class WrappedAvailableBalanceChanged(e: AvailableBalanceChanged) extends Command
   // @formatter:on
 
-  def apply(db: AuditDb, channels: Seq[PersistentChannelData]): Behavior[Command] = {
+  case class Config(pastEventsInitDelay: FiniteDuration, pastEventsChunkDelay: FiniteDuration)
+
+  def apply(config: Config, db: AuditDb, channels: Seq[PersistentChannelData]): Behavior[Command] = {
     Behaviors.setup { context =>
       Behaviors.withTimers { timers =>
-        new PeerStatsTracker(db, timers, context).start(channels)
+        new PeerStatsTracker(config, db, timers, context).start(channels)
       }
     }
   }
@@ -323,7 +326,7 @@ object PeerStatsTracker {
 
 }
 
-private class PeerStatsTracker(db: AuditDb, timers: TimerScheduler[PeerStatsTracker.Command], context: ActorContext[PeerStatsTracker.Command]) {
+private class PeerStatsTracker(config: PeerStatsTracker.Config, db: AuditDb, timers: TimerScheduler[PeerStatsTracker.Command], context: ActorContext[PeerStatsTracker.Command]) {
 
   import PeerStatsTracker._
 
@@ -345,7 +348,7 @@ private class PeerStatsTracker(db: AuditDb, timers: TimerScheduler[PeerStatsTrac
     context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedPaymentRelayed))
     context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedPaymentReceived))
     // We start reading past events from the DB.
-    context.self ! LoadPastEvents(before = startedAt)
+    timers.startSingleTimer(LoadPastEvents(before = startedAt), config.pastEventsInitDelay)
     val stats = BucketedPeerStats.empty(peerChannels.peers)
     timers.startTimerWithFixedDelay(RemoveOldBuckets, Bucket.duration)
     listening(stats, peerChannels, startedAt)
@@ -381,13 +384,13 @@ private class PeerStatsTracker(db: AuditDb, timers: TimerScheduler[PeerStatsTrac
           pastEventsLoaded = true
           listening(stats, channels, lastTransactionReadAt)
         } else {
-          log.info("loading past events before {}", before)
+          log.info("loading past events before {}", Instant.ofEpochMilli(before.toLong))
           val stats1 = db.listSent(before - Bucket.duration, before).foldLeft(stats) { case (current, e) => current.addPaymentSent(e) }
           val stats2 = db.listReceived(before - Bucket.duration, before).foldLeft(stats1) { case (current, e) => current.addPaymentReceived(e) }
           val stats3 = db.listRelayed(before - Bucket.duration, before).foldLeft(stats2) { case (current, e) => current.addPaymentRelayed(e) }
           val stats4 = db.listConfirmed(before - Bucket.duration, before).foldLeft(stats3) { case (current, e) => current.addConfirmedTransaction(e) }
           // We continue reading events from past buckets.
-          context.self ! LoadPastEvents(before - Bucket.duration)
+          timers.startSingleTimer(LoadPastEvents(before - Bucket.duration), config.pastEventsChunkDelay)
           listening(stats4, channels, lastTransactionReadAt)
         }
       case RemoveOldBuckets =>
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/TestConstants.scala b/eclair-core/src/test/scala/fr/acinq/eclair/TestConstants.scala
index 94a5eda..20f87d5 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/TestConstants.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/TestConstants.scala
@@ -29,7 +29,7 @@ import fr.acinq.eclair.message.OnionMessages.OnionMessageConfig
 import fr.acinq.eclair.payment.offer.OffersConfig
 import fr.acinq.eclair.payment.relay.OnTheFlyFunding
 import fr.acinq.eclair.payment.relay.Relayer.{AsyncPaymentsParams, RelayFees, RelayParams}
-import fr.acinq.eclair.profit.PeerScorer
+import fr.acinq.eclair.profit.{PeerScorer, PeerStatsTracker}
 import fr.acinq.eclair.reputation.Reputation
 import fr.acinq.eclair.router.Graph.{HeuristicsConstants, MessageWeightRatios}
 import fr.acinq.eclair.router.Router._
@@ -290,6 +290,7 @@ object TestConstants {
           dailyPaymentVolumeThresholdPercent = 0.1,
         )
       ),
+      peerStatsTrackerConfig = PeerStatsTracker.Config(pastEventsInitDelay = 1 millis, pastEventsChunkDelay = 1 millis),
       offersConfig = OffersConfig(messagePathMinLength = 2, paymentPathCount = 2, paymentPathLength = 4, paymentPathCltvExpiryDelta = CltvExpiryDelta(500)),
     )
 
@@ -509,6 +510,7 @@ object TestConstants {
           dailyPaymentVolumeThresholdPercent = 0.1,
         )
       ),
+      peerStatsTrackerConfig = PeerStatsTracker.Config(pastEventsInitDelay = 1 millis, pastEventsChunkDelay = 1 millis),
       offersConfig = OffersConfig(messagePathMinLength = 2, paymentPathCount = 2, paymentPathLength = 4, paymentPathCltvExpiryDelta = CltvExpiryDelta(500)),
     )
 
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerScorerSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerScorerSpec.scala
index 4f0fc96..c4d13c9 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerScorerSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerScorerSpec.scala
@@ -30,7 +30,7 @@ class PeerScorerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load("appli
   private val remoteNodeId3 = PublicKey(hex"028f5be42aa013f9fd2e5a28a152563ac21acc095ef65cab2e835a789d2a4add96")
 
   private val weeklyBuckets = Bucket.bucketsPerDay * 7
-  private val defaultConfig = Config(
+  private val defaultConfig = PeerScorer.Config(
     enabled = true,
     // We set this to 1 day to more easily match daily rate-limits.
     scoringFrequency = 1 day,
@@ -60,7 +60,7 @@ class PeerScorerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load("appli
 
   private case class Fixture(nodeParams: NodeParams, tracker: TestProbe[GetLatestStats], register: TestProbe[Any], wallet: DummyBalanceChecker, scorer: ActorRef[PeerScorer.Command])
 
-  private def withFixture(config: Config = defaultConfig, onChainBalance: BtcAmount = 0 sat)(testFun: Fixture => Any): Unit = {
+  private def withFixture(config: PeerScorer.Config = defaultConfig, onChainBalance: BtcAmount = 0 sat)(testFun: Fixture => Any): Unit = {
     val tracker = TestProbe[GetLatestStats]()
     val register = TestProbe[Any]()
     val wallet = new DummyBalanceChecker(confirmedBalance = onChainBalance.toMilliSatoshi.truncateToSatoshi)
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerStatsTrackerSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerStatsTrackerSpec.scala
index c2b7a95..ecc5eb1 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerStatsTrackerSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerStatsTrackerSpec.scala
@@ -34,6 +34,8 @@ class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load(
   private val remoteNodeId2 = PublicKey(hex"02271ffb6969f6dc4769438637d7d24dc3358098cdef7a772f9ccfd31251470e28")
   private val remoteNodeId3 = PublicKey(hex"028f5be42aa013f9fd2e5a28a152563ac21acc095ef65cab2e835a789d2a4add96")
 
+  private val config = Config(pastEventsInitDelay = 1 millis, pastEventsChunkDelay = 1 millis)
+
   private def commitments(remoteNodeId: PublicKey, toLocal: BtcAmount, toRemote: BtcAmount, announceChannel: Boolean = true): Commitments = {
     CommitmentsSpec.makeCommitments(toLocal.toMilliSatoshi, toRemote.toMilliSatoshi, localNodeId, remoteNodeId, if (announceChannel) Some(dummyChannelAnn) else None)
   }
@@ -107,7 +109,7 @@ class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load(
     val c3b = DATA_CLOSING(commitments(remoteNodeId3, toLocal = 0.2 btc, toRemote = 0.2 btc), BlockHeight(750_000), ByteVector.empty, Nil, ClosingTx(InputInfo(OutPoint(randomTxId(), 2), TxOut(500_000 sat, Script.pay2wpkh(dummyPubKey))), Transaction(2, Nil, TxOut(500_000 sat, Script.pay2wpkh(dummyPubKey)) :: Nil, 0), None) :: Nil)
 
     // We initialize our actor with these existing channels.
-    val tracker = testKit.spawn(PeerStatsTracker(TestDatabases.inMemoryDb().audit, Seq(c1a, c1b, c1c, c1d, c2a, c2b, c3a, c3b)))
+    val tracker = testKit.spawn(PeerStatsTracker(config, TestDatabases.inMemoryDb().audit, Seq(c1a, c1b, c1c, c1d, c2a, c2b, c3a, c3b)))
     // We have relayed payments through channels that have been closed since then.
     tracker.ref ! WrappedPaymentRelayed(ChannelPaymentRelayed(
       paymentHash = randomBytes32(),
@@ -238,7 +240,7 @@ class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load(
   test("keep track of peer statistics") {
     val now = TimestampMilli.now()
     val probe = TestProbe[LatestStats]()
-    val tracker = testKit.spawn(PeerStatsTracker(TestDatabases.inMemoryDb().audit, Nil))
+    val tracker = testKit.spawn(PeerStatsTracker(config, TestDatabases.inMemoryDb().audit, Nil))
 
     // We have channels with 3 peers:
     val c1a = commitments(remoteNodeId1, toLocal = 0.5 btc, toRemote = 0.3 btc)
@@ -411,7 +413,7 @@ class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load(
     // Spawn tracker with channels for both peers.
     val c1 = channel(remoteNodeId1, toLocal = 0.3 btc, toRemote = 0.2 btc)
     val c2 = channel(remoteNodeId2, toLocal = 0.1 btc, toRemote = 0.4 btc)
-    val tracker = testKit.spawn(PeerStatsTracker(db, Seq(c1, c2)))
+    val tracker = testKit.spawn(PeerStatsTracker(config, db, Seq(c1, c2)))
 
     probe.awaitAssert({
       tracker.ref ! GetLatestStats(probe.ref)
@@ -449,7 +451,7 @@ class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load(
 
     // Spawn tracker with a channel for remoteNodeId1 and empty DB.
     val c1 = channel(remoteNodeId1, toLocal = 0.3 btc, toRemote = 0.2 btc)
-    val tracker = testKit.spawn(PeerStatsTracker(db, Seq(c1)))
+    val tracker = testKit.spawn(PeerStatsTracker(config, db, Seq(c1)))
 
     // Wait for initial loading to complete (should be almost instantaneous on an empty DB).
     probe.awaitAssert({