← Watch feed
Monitor the internal state of `ReputationRecorder` (#3216)
Public commit record
81/100 · Strong
What the developer wrote
Authored by Bastien Teinturier
Monitor the internal state of `ReputationRecorder` (#3216)
It is quite hard to reason about the internal state of this actor, and
it tracks every pending HTLC. If for some reason we're not removing
HTLCs correctly, this would create a memory leak.
We log the number of pending HTLCs every 5 minutes to help detect if
clean-up is correctly implemented.
✓ Specific, descriptive subject✓ Names a concrete action or component✓ Provides detailed explanatory context✓ Links an issue, advisory, or supporting reference
What changed, and why it matters
This commit adds a periodic log message to an internal bookkeeping actor so developers can spot if pending payment records are piling up unexpectedly. It does not fix a known bug or change how payments are processed; it is purely an observability/monitoring improvement.
Recommended action
No immediate action required. Treat as routine observability improvement. If logs show pending HTLC counts growing unbounded over time, investigate the add/settle event matching logic separately.
Security signals we found
01
Defensive memory-leak monitoring added
02
No logic change to HTLC add/settle handling
03
No bounds or cleanup logic introduced
04
No input validation or access-control changes
Evidence from the diff
The patch instruments ReputationRecorder with a 5-minute timer that logs counts of pending HTLCs held in internal mutable maps. It also threads the ActorContext into the actor class so it can access the logger. The change is defensive: it makes a hypothetical memory-leak scenario visible in logs but does not alter HTLC add/settle logic or close any identified leak.
Changed components
eclair-core/src/main/scala/fr/acinq/eclair/reputation/ReputationRecorder.scalaeclair-core/src/test/scala/fr/acinq/eclair/reputation/ReputationRecorderSpec.scalaInspect captured patch +22 / −10
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/reputation/ReputationRecorder.scala b/eclair-core/src/main/scala/fr/acinq/eclair/reputation/ReputationRecorder.scala
index 58224a5..df52a56 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/reputation/ReputationRecorder.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/reputation/ReputationRecorder.scala
@@ -17,7 +17,7 @@
package fr.acinq.eclair.reputation
import akka.actor.typed.eventstream.EventStream
-import akka.actor.typed.scaladsl.Behaviors
+import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
import akka.actor.typed.{ActorRef, Behavior}
import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
import fr.acinq.eclair.{BlockHeight, CltvExpiry, MilliSatoshi}
@@ -27,6 +27,7 @@ import fr.acinq.eclair.reputation.ReputationRecorder._
import fr.acinq.eclair.wire.protocol.{UpdateAddHtlc, UpdateFailHtlc, UpdateFailMalformedHtlc}
import scala.collection.mutable
+import scala.concurrent.duration.DurationInt
object ReputationRecorder {
// @formatter:off
@@ -34,14 +35,18 @@ object ReputationRecorder {
case class GetConfidence(replyTo: ActorRef[Reputation.Score], upstream: Upstream.Hot, downstream_opt: Option[PublicKey], fee: MilliSatoshi, currentBlockHeight: BlockHeight, expiry: CltvExpiry) extends Command
case class WrappedOutgoingHtlcAdded(added: OutgoingHtlcAdded) extends Command
case class WrappedOutgoingHtlcSettled(settled: OutgoingHtlcSettled) extends Command
+ private case object TickAudit extends Command
// @formatter:on
def apply(config: Reputation.Config): Behavior[Command] =
- Behaviors.setup(context => {
- context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedOutgoingHtlcSettled))
- context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedOutgoingHtlcAdded))
- new ReputationRecorder(config).run()
- })
+ Behaviors.setup { context =>
+ Behaviors.withTimers { timers =>
+ context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedOutgoingHtlcSettled))
+ context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedOutgoingHtlcAdded))
+ timers.startTimerWithFixedDelay(TickAudit, 5 minutes)
+ new ReputationRecorder(context, config).run()
+ }
+ }
/**
* A pending outgoing HTLC.
@@ -53,11 +58,13 @@ object ReputationRecorder {
case class PendingHtlc(add: UpdateAddHtlc, upstream: Upstream.Hot, downstream: PublicKey)
}
-class ReputationRecorder(config: Reputation.Config) {
+class ReputationRecorder(context: ActorContext[ReputationRecorder.Command], config: Reputation.Config) {
private val incomingReputations: mutable.Map[PublicKey, Reputation] = mutable.HashMap.empty.withDefaultValue(Reputation.init(config))
private val outgoingReputations: mutable.Map[PublicKey, Reputation] = mutable.HashMap.empty.withDefaultValue(Reputation.init(config))
private val pending: mutable.Map[HtlcId, PendingHtlc] = mutable.HashMap.empty
+ private val log = context.log
+
def run(): Behavior[Command] =
Behaviors.receiveMessage {
case GetConfidence(replyTo, _: Upstream.Local, _, _, _, _) =>
@@ -128,5 +135,11 @@ class ReputationRecorder(config: Reputation.Config) {
outgoingReputations(p.downstream) = outgoingReputations(p.downstream).settlePendingHtlc(htlcId, isSuccess)
})
Behaviors.same
+
+ case TickAudit =>
+ val totalIncomingPending = incomingReputations.values.map(_.pending.size).sum
+ val totalOutgoingPending = outgoingReputations.values.map(_.pending.size).sum
+ log.info("{} pending HTLCs: {} tracked for incoming reputation and {} tracked for outgoing reputation", pending.size, totalIncomingPending, totalOutgoingPending)
+ Behaviors.same
}
}
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/reputation/ReputationRecorderSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/reputation/ReputationRecorderSpec.scala
index 1cbbe2b..2d310d2 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/reputation/ReputationRecorderSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/reputation/ReputationRecorderSpec.scala
@@ -18,13 +18,12 @@ package fr.acinq.eclair.reputation
import akka.actor.testkit.typed.scaladsl.{ScalaTestWithActorTestKit, TestProbe}
import akka.actor.typed.ActorRef
-import akka.actor.typed.eventstream.EventStream
import akka.testkit.TestKit.awaitCond
import com.typesafe.config.ConfigFactory
import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
import fr.acinq.eclair.channel.{OutgoingHtlcAdded, OutgoingHtlcFailed, OutgoingHtlcFulfilled, Upstream}
import fr.acinq.eclair.reputation.ReputationRecorder._
-import fr.acinq.eclair.wire.protocol.{TlvStream, UpdateAddHtlc, UpdateAddHtlcTlv, UpdateFailHtlc, UpdateFulfillHtlc}
+import fr.acinq.eclair.wire.protocol._
import fr.acinq.eclair.{BlockHeight, CltvExpiry, MilliSatoshi, MilliSatoshiLong, TimestampMilli, randomBytes, randomBytes32, randomKey, randomLong}
import org.scalatest.Outcome
import org.scalatest.funsuite.FixtureAnyFunSuiteLike
@@ -206,7 +205,7 @@ class ReputationRecorderSpec extends ScalaTestWithActorTestKit(ConfigFactory.loa
}, max = 60 seconds)
}
- test("sink attack") {f =>
+ test("sink attack") { f =>
import f._
val (a, b, c) = (randomKey().publicKey, randomKey().publicKey, randomKey().publicKey)
Risk score
Our methodology →Why this scored 17/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.