From 1ee79880438c32c857f6e9a7059b298448e2f35d Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Tue, 18 Aug 2026 13:40:34 +0000 Subject: [PATCH 1/6] reputation: add decaying average and revenue aggregation Add the numeric primitives underlying local reputation scoring, following the "Decaying Average" and "Revenue Threshold Aggregation" sections of BOLT #1280, plus a package README describing the subsystem: - saturatedI64: int64 arithmetic that clamps rather than wraps, so the long-window fee accumulators never silently flip sign. - decayingAverage: a value decaying as e^(-elapsed/window) per the spec's decay_rate. - aggregatedWindowAverage: a decaying average over several windows with the spec's exponential warm-up factor. --- reputation/README.md | 78 +++++++++++++ reputation/decaying_average.go | 71 ++++++++++++ reputation/decaying_average_test.go | 119 +++++++++++++++++++ reputation/revenue.go | 92 +++++++++++++++ reputation/saturated.go | 83 ++++++++++++++ reputation/saturated_test.go | 172 ++++++++++++++++++++++++++++ 6 files changed, 615 insertions(+) create mode 100644 reputation/README.md create mode 100644 reputation/decaying_average.go create mode 100644 reputation/decaying_average_test.go create mode 100644 reputation/revenue.go create mode 100644 reputation/saturated.go create mode 100644 reputation/saturated_test.go diff --git a/reputation/README.md b/reputation/README.md new file mode 100644 index 00000000000..2400ec639ed --- /dev/null +++ b/reputation/README.md @@ -0,0 +1,78 @@ +reputation +========== + +[![Build Status](http://img.shields.io/travis/lightningnetwork/lnd.svg)](https://travis-ci.org/lightningnetwork/lnd) +[![MIT licensed](https://img.shields.io/badge/license-MIT-blue.svg)](https://github.com/lightningnetwork/lnd/blob/master/LICENSE) +[![GoDoc](https://img.shields.io/badge/godoc-reference-blue.svg)](http://godoc.org/github.com/lightningnetwork/lnd/reputation) + +The reputation package implements local reputation tracking to help mitigate +channel jamming, following the scoring recommended in [BOLT +\#1280](https://github.com/lightning/bolts/pull/1280) (local resource +conservation). A forwarding node uses it to build an unforgeable history of how +each channel has behaved as an outgoing peer, so that it can later distinguish +peers that are likely being used to jam its channels from those that are not. + +The package is **observational only**: it watches the HTLCs the node forwards, +maintains a per-channel reputation score, and logs the decision it would make +for each HTLC. It never affects forwarding, alters the wire, or writes to disk. + +## Reputation scoring + +Every forwarded HTLC contributes an `effective_fee` to its outgoing channel's +reputation, adjusted for how long it was held: HTLCs that resolve within a +`resolution_period` contribute their full fee, while slower ones are penalised +by an `opportunity_cost` that grows with the overrun. Unaccountable HTLCs can +only ever help reputation, never harm it. + +Three quantities determine whether a channel has sufficient reputation for a +given HTLC: + + * **Outgoing channel reputation**: the sum of effective fees the outgoing + channel has earned over a long rolling window, tracked as a decaying + average. + * **Incoming channel revenue threshold**: the routing revenue the incoming + channel has generated over a shorter window, aggregated over several + windows so a peer cannot cheaply move its own threshold. + * **In-flight risk**: the worst-case opportunity cost of the HTLC assuming it + is held until just before its incoming CLTV expiry. + +An HTLC's outgoing channel is considered to have sufficient reputation when: + + outgoing_channel_reputation - in_flight_risk >= incoming_revenue_threshold + +This is evaluated two ways for each forward: against the HTLC's own risk alone, +and against that plus the risk of the accountable HTLCs already in flight on the +outgoing channel. Both verdicts are logged. + +The rolling windows are implemented as decaying averages to avoid storing +per-HTLC history; see `decaying_average.go`. + +## Integration with the switch + +The subsystem observes forwarding through three read-only hooks the switch +calls at the circuit layer: `OnForward` when it commits to forwarding an HTLC, +and `OnSettle`/`OnFail` when the HTLC resolves. The hooks run synchronously and +do only a handful of map lookups and floating-point operations, so they sit on +the forwarding path without a background worker. + +When the subsystem is disabled the switch skips the hooks behind a nil check. +When it is enabled the manager is wrapped in a panic boundary before being +handed to the switch, so a bug in this (log-only) package can never take down +HTLC forwarding. + +Every resolution removes its own pending HTLC, so a pending entry that outlives +the worst case time it could be held for means a resolution was never reported +to us. Such entries are logged as a warning and deliberately left in place +rather than swept away, so the underlying bug stays visible. + +## Operational notes + +The subsystem is enabled by default and can be disabled with the +`routing.no-reputation` configuration flag. It holds no persisted state, so +reputation resets on restart and re-accrues from live forwarding traffic. + +## Installation and Updating + +```shell +$ go get -u github.com/lightningnetwork/lnd/reputation +``` diff --git a/reputation/decaying_average.go b/reputation/decaying_average.go new file mode 100644 index 00000000000..896464dba77 --- /dev/null +++ b/reputation/decaying_average.go @@ -0,0 +1,71 @@ +package reputation + +import ( + "errors" + "math" + "time" +) + +// errBackwardsTime is returned when a decaying average is asked to evaluate at +// a timestamp earlier than its last update. The algorithm assumes monotonic +// time. +var errBackwardsTime = errors.New("timestamp precedes last update") + +// decayingAverage tracks a value that decays exponentially over a rolling +// window. The running value saturates rather than wrapping (see saturatedI64). +type decayingAverage struct { + value saturatedI64 + lastUpdated time.Time + decayRate float64 +} + +// newDecayingAverage creates a decaying average that starts at zero as of the +// provided start time, decaying over the given window. +func newDecayingAverage(start time.Time, + window time.Duration) *decayingAverage { + + return &decayingAverage{ + lastUpdated: start, + decayRate: decayRateForWindow(window), + } +} + +// decayRateForWindow computes the per-second decay rate for the given window. +// BOLT #1280 defines decay_rate = (1/2)^(1/(ln2 * window)); raised to elapsed +// seconds this is e^(-elapsed/window), so the value decays to 1/e of itself +// over a full window. +func decayRateForWindow(window time.Duration) float64 { + return math.Pow(0.5, 1.0/(math.Ln2*window.Seconds())) +} + +// valueAt returns the stored value decayed forward to the given time. It is +// read-only: the internal state is only mutated by add, so that frequent reads +// do not accumulate rounding error. It errors if the time is before the last +// update. +func (d *decayingAverage) valueAt(ts time.Time) (int64, error) { + if ts.Before(d.lastUpdated) { + return 0, errBackwardsTime + } + + elapsed := ts.Sub(d.lastUpdated).Seconds() + decayed := satFromFloat( + math.Round(float64(d.value.Int64()) * + math.Pow(d.decayRate, elapsed)), + ) + + return decayed.Int64(), nil +} + +// add decays the value to the given time and then adds the provided (possibly +// negative) value. This is the only operation that mutates the stored value. +func (d *decayingAverage) add(value int64, ts time.Time) (int64, error) { + decayed, err := d.valueAt(ts) + if err != nil { + return 0, err + } + + d.value = satFromInt(decayed).Add(satFromInt(value)) + d.lastUpdated = ts + + return d.value.Int64(), nil +} diff --git a/reputation/decaying_average_test.go b/reputation/decaying_average_test.go new file mode 100644 index 00000000000..b893877fd5e --- /dev/null +++ b/reputation/decaying_average_test.go @@ -0,0 +1,119 @@ +package reputation + +import ( + "math" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// testStart is the base time used by the tests that drive the averages +// directly. +var testStart = time.Unix(1_000_000, 0) + +// TestDecayingAverageDecay verifies the decay e^(-elapsed/window) and the +// add-then-decay sequencing: the value falls to 1/sqrt(e) of itself after half +// a window and to 1/e after a full window, and an add applies on top of the +// value decayed to the add's timestamp. +func TestDecayingAverageDecay(t *testing.T) { + t.Parallel() + + const window = 100 * time.Second + d := newDecayingAverage(testStart, window) + + _, err := d.add(1000, testStart) + require.NoError(t, err) + + // At half a window (50s): 1000 * e^(-0.5) = 606.5 -> 607. + got, err := d.valueAt(testStart.Add(50 * time.Second)) + require.NoError(t, err) + require.EqualValues(t, 607, got, "half window") + + // At a full window (another 50s): 1000 * e^(-1) = 367.9 -> 368. + got, err = d.valueAt(testStart.Add(100 * time.Second)) + require.NoError(t, err) + require.EqualValues(t, 368, got, "full window") + + // valueAt is read-only, so adding at 50s still works after the reads + // above: the value decays to 607 and the add lands at 1607. + v, err := d.add(1000, testStart.Add(50*time.Second)) + require.NoError(t, err) + require.EqualValues(t, 1607, v) +} + +// TestDecayingAverageBackwardsTime ensures a backwards timestamp errors, since +// the decay assumes monotonic time. +func TestDecayingAverageBackwardsTime(t *testing.T) { + t.Parallel() + + d := newDecayingAverage(testStart, time.Hour) + _, err := d.valueAt(testStart.Add(-50 * time.Second)) + require.ErrorIs(t, err, errBackwardsTime) +} + +// TestDecayingAverageOverflowClamp verifies that evaluating a saturated +// (near-MaxInt64) value does not flip negative. Because float64(MaxInt64) +// rounds up to 2^63, a naive int64(math.Round(...)) cast yields MinInt64; the +// clamp must keep it saturated at MaxInt64. +func TestDecayingAverageOverflowClamp(t *testing.T) { + t.Parallel() + + const window = 100 * time.Second + d := newDecayingAverage(testStart, window) + + // Saturate the running value to MaxInt64. + _, err := d.add(math.MaxInt64, testStart) + require.NoError(t, err) + require.EqualValues( + t, math.MaxInt64, d.value.Int64(), "setup: value not saturated", + ) + + // Evaluating at the same timestamp (no decay) round-trips the value + // through float64; without the clamp this overflows to MinInt64. + got, err := d.valueAt(testStart) + require.NoError(t, err) + require.EqualValues(t, int64(math.MaxInt64), got, "a negative value "+ + "means the float->int64 cast overflowed") +} + +// TestAggregatedWindowWarmup verifies the warm-up factor +// windowCount*(1 - exp(-periods/windowCount)), guarded at 1. +func TestAggregatedWindowWarmup(t *testing.T) { + t.Parallel() + + // window = 100s, windowCount = 6 -> inner window 600s. + a := newAggregatedWindowAverage(100*time.Second, 6, testStart) + + // Add 600 at t=0. periods=0 => warmup factor tends to 0 and is guarded + // to 1, so the value reads back as 600 (no decay at t=0). + _, err := a.add(600, testStart) + require.NoError(t, err) + + got, err := a.valueAt(testStart) + require.NoError(t, err) + require.EqualValues(t, 600, got, "warmup t=0") + + // At t=300 (periods=3), the inner value has decayed by e^(-300/600) to + // 364, and the warm-up factor is 6*(1 - exp(-3/6)) = 2.3608..., so + // 364/2.3608 rounds to 154. + got, err = a.valueAt(testStart.Add(300 * time.Second)) + require.NoError(t, err) + require.EqualValues(t, 154, got, "warmup t=300") +} + +// TestAggregatedWindowBackwardsTime ensures reading an aggregated average +// before its start timestamp errors rather than underflowing the unsigned +// elapsed-time subtraction. +func TestAggregatedWindowBackwardsTime(t *testing.T) { + t.Parallel() + + a := newAggregatedWindowAverage(100*time.Second, 6, testStart) + before := testStart.Add(-50 * time.Second) + + _, err := a.valueAt(before) + require.ErrorIs(t, err, errBackwardsTime) + + _, err = a.windowsTracked(before) + require.ErrorIs(t, err, errBackwardsTime) +} diff --git a/reputation/revenue.go b/reputation/revenue.go new file mode 100644 index 00000000000..d4c2c43a88f --- /dev/null +++ b/reputation/revenue.go @@ -0,0 +1,92 @@ +package reputation + +import ( + "math" + "time" +) + +// aggregatedWindowAverage tracks an average value over multiple rolling +// windows. Aggregating over several windows rather than reading a single one +// smooths out volatility, which makes the average harder to move quickly by +// manipulating recent activity. +// +// It wraps a single decaying average over windowDuration*windowCount and, when +// reading, divides by a warm-up factor so that a brief history does not read as +// an artificially low average (see warmupFactor). +type aggregatedWindowAverage struct { + start time.Time + windowCount uint8 + windowDuration time.Duration + inner *decayingAverage +} + +// newAggregatedWindowAverage creates an aggregated average starting at zero as +// of start, tracking value over windowCount windows each of windowDuration. +func newAggregatedWindowAverage(window time.Duration, windowCount uint8, + start time.Time) *aggregatedWindowAverage { + + return &aggregatedWindowAverage{ + start: start, + windowCount: windowCount, + windowDuration: window, + inner: newDecayingAverage( + start, window*time.Duration(windowCount), + ), + } +} + +// add records a value at the given time. +func (a *aggregatedWindowAverage) add(value int64, + ts time.Time) (int64, error) { + + return a.inner.add(value, ts) +} + +// windowsTracked returns the (fractional) number of windows (periods) elapsed +// since start. It errors if the time precedes start, since a negative number of +// elapsed periods is not meaningful. +func (a *aggregatedWindowAverage) windowsTracked(ts time.Time) (float64, + error) { + + if ts.Before(a.start) { + return 0, errBackwardsTime + } + + return ts.Sub(a.start).Seconds() / a.windowDuration.Seconds(), nil +} + +// warmupFactor returns the warm-up divisor for the number of periods +// (fractional windows) elapsed so far: +// +// warmup = windowCount * (1 - exp(-periods / windowCount)) +// +// As periods grows this converges to windowCount (the steady-state divisor). +// It is guarded at 1 to avoid the periods->0 singularity where the factor tends +// to 0 and would over-inflate the average. +func (a *aggregatedWindowAverage) warmupFactor(periods float64) float64 { + count := float64(a.windowCount) + + warmup := count * (1 - math.Exp(-periods/count)) + if warmup < 1 { + warmup = 1 + } + + return warmup +} + +// valueAt returns the windowed average value as of the given time. +func (a *aggregatedWindowAverage) valueAt(ts time.Time) (int64, error) { + periods, err := a.windowsTracked(ts) + if err != nil { + return 0, err + } + + warmup := a.warmupFactor(periods) + + raw, err := a.inner.valueAt(ts) + if err != nil { + return 0, err + } + + return satFromFloat(math.Round(float64(raw) / warmup)).Int64(), nil +} diff --git a/reputation/saturated.go b/reputation/saturated.go new file mode 100644 index 00000000000..e2dc7b12eac --- /dev/null +++ b/reputation/saturated.go @@ -0,0 +1,83 @@ +package reputation + +import "math" + +// saturatedI64 wraps an int64 that clamps to the int64 range on overflow +// instead of wrapping, so that an accumulating value can never silently flip +// sign once it grows past the int64 bound. It is a struct rather than a named +// int64 so that the raw +/- operators cannot be used on it by accident: all +// arithmetic has to go through Add and Sub. +type saturatedI64 struct { + v int64 +} + +// satFromInt returns a saturatedI64 holding the given value. +func satFromInt(v int64) saturatedI64 { + return saturatedI64{v: v} +} + +// satFromUint converts an unsigned value, clamping to the maximum when it does +// not fit. +func satFromUint(v uint64) saturatedI64 { + if v > math.MaxInt64 { + return saturatedI64{v: math.MaxInt64} + } + + return saturatedI64{v: int64(v)} +} + +// satFromFloat converts a float, clamping to the int64 range. A plain +// float-to-int conversion is undefined out of range in Go (in practice it +// yields MinInt64), so a value that rounds above the maximum would flip +// negative without this guard. +func satFromFloat(f float64) saturatedI64 { + switch { + case f >= float64(math.MaxInt64): + return saturatedI64{v: math.MaxInt64} + + case f <= float64(math.MinInt64): + return saturatedI64{v: math.MinInt64} + + default: + return saturatedI64{v: int64(f)} + } +} + +// Add returns the saturating sum of the two values. +func (s saturatedI64) Add(o saturatedI64) saturatedI64 { + sum := s.v + o.v + switch { + case s.v > 0 && o.v > 0 && sum < 0: + return saturatedI64{v: math.MaxInt64} + + case s.v < 0 && o.v < 0 && sum >= 0: + return saturatedI64{v: math.MinInt64} + + default: + return saturatedI64{v: sum} + } +} + +// Sub returns the saturating difference of the two values. +func (s saturatedI64) Sub(o saturatedI64) saturatedI64 { + diff := s.v - o.v + switch { + // Underflow: subtracting a positive from a negative can only go more + // negative, so a non-negative result means it wrapped. + case s.v < 0 && o.v > 0 && diff > 0: + return saturatedI64{v: math.MinInt64} + + // Overflow: subtracting a negative from a non-negative can only go more + // positive, so a negative result means it wrapped. + case s.v >= 0 && o.v < 0 && diff < 0: + return saturatedI64{v: math.MaxInt64} + + default: + return saturatedI64{v: diff} + } +} + +// Int64 returns the underlying int64. +func (s saturatedI64) Int64() int64 { + return s.v +} diff --git a/reputation/saturated_test.go b/reputation/saturated_test.go new file mode 100644 index 00000000000..3dc0d9663b4 --- /dev/null +++ b/reputation/saturated_test.go @@ -0,0 +1,172 @@ +package reputation + +import ( + "math" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestSatFromUint checks conversion from unsigned values, which must clamp +// rather than wrap when the value does not fit in an int64. +func TestSatFromUint(t *testing.T) { + t.Parallel() + + require.Equal(t, satFromInt(0), satFromUint(0)) + require.Equal(t, satFromInt(1000), satFromUint(1000)) + require.Equal( + t, satFromInt(math.MaxInt64), satFromUint(math.MaxInt64), + ) + + // Anything above MaxInt64 saturates instead of wrapping negative. + require.Equal( + t, satFromInt(math.MaxInt64), satFromUint(math.MaxInt64+1), + ) + require.Equal( + t, satFromInt(math.MaxInt64), satFromUint(math.MaxUint64), + ) +} + +// TestSatFromFloat checks conversion from floats. An out-of-range float-to-int +// conversion is undefined in Go, so the bounds must be clamped explicitly. +func TestSatFromFloat(t *testing.T) { + t.Parallel() + + require.Equal(t, satFromInt(0), satFromFloat(0)) + require.Equal(t, satFromInt(1000), satFromFloat(1000)) + require.Equal(t, satFromInt(-1000), satFromFloat(-1000)) + + // float64(MaxInt64) rounds up to 2^63, so it is out of range and must + // saturate rather than yield MinInt64. + require.Equal( + t, satFromInt(math.MaxInt64), + satFromFloat(float64(math.MaxInt64)), + ) + require.Equal( + t, satFromInt(math.MinInt64), + satFromFloat(float64(math.MinInt64)), + ) + require.Equal(t, satFromInt(math.MaxInt64), satFromFloat(1e30)) + require.Equal(t, satFromInt(math.MinInt64), satFromFloat(-1e30)) +} + +// TestSaturatedAdd checks that addition clamps at both bounds instead of +// wrapping. +func TestSaturatedAdd(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + a, b int64 + expected int64 + }{{ + name: "simple", + a: 5, + b: 3, + expected: 8, + }, { + name: "add negative", + a: 5, + b: -3, + expected: 2, + }, { + name: "overflow saturates", + a: math.MaxInt64, + b: 1, + expected: math.MaxInt64, + }, { + name: "underflow saturates", + a: math.MinInt64, + b: -1, + expected: math.MinInt64, + }, { + name: "opposite extremes cancel", + a: math.MaxInt64, + b: math.MinInt64, + expected: -1, + }} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + got := satFromInt(test.a).Add(satFromInt(test.b)) + require.Equal(t, satFromInt(test.expected), got) + }) + } +} + +// TestSaturatedSub checks that subtraction clamps at both bounds instead of +// wrapping, including the extremes where negating the operand would itself +// overflow. +func TestSaturatedSub(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + a, b int64 + expected int64 + }{{ + name: "simple", + a: 5, + b: 3, + expected: 2, + }, { + name: "subtract negative", + a: 5, + b: -3, + expected: 8, + }, { + name: "result goes negative", + a: 3, + b: 5, + expected: -2, + }, { + // Both operands are negative, so the result is exactly + // representable and must not be clamped. + name: "min plus one minus min", + a: math.MinInt64 + 1, + b: math.MinInt64, + expected: 1, + }, { + // 0 - (-2^63) = 2^63, which exceeds MaxInt64. + name: "overflow saturates", + a: 0, + b: math.MinInt64, + expected: math.MaxInt64, + }, { + name: "underflow saturates", + a: math.MinInt64, + b: 1, + expected: math.MinInt64, + }, { + name: "max minus negative saturates", + a: math.MaxInt64, + b: -1, + expected: math.MaxInt64, + }, { + name: "min minus max saturates", + a: math.MinInt64, + b: math.MaxInt64, + expected: math.MinInt64, + }} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + got := satFromInt(test.a).Sub(satFromInt(test.b)) + require.Equal(t, satFromInt(test.expected), got) + }) + } +} + +// TestSaturatedInt64 checks the conversion back to a plain int64. +func TestSaturatedInt64(t *testing.T) { + t.Parallel() + + require.Equal(t, int64(1000), satFromInt(1000).Int64()) + require.Equal( + t, int64(math.MinInt64), satFromInt(math.MinInt64).Int64(), + ) +} From fd439cc50a807ad05e6ffc4d5aced43faf529c2e Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Tue, 18 Aug 2026 13:40:34 +0000 Subject: [PATCH 2/6] reputation: add HTLC fee and reputation scoring Add the per-channel reputation state and the BOLT #1280 scoring rules built on the decaying-average primitives: - Config: the tunable parameters (resolution period, revenue window, reputation multiplier, revenue window count) with the spec defaults. - effectiveFee/opportunityCost/inFlightRisk: an HTLC's contribution to reputation and its worst-case in-flight risk. - channelReputation: the per-channel outgoing reputation, incoming revenue threshold and pending HTLCs, plus the sufficiency inequality outgoing_reputation - risk >= revenue_threshold. --- reputation/channel.go | 76 +++++++++++++++++++++++++ reputation/config.go | 97 +++++++++++++++++++++++++++++++ reputation/config_test.go | 75 ++++++++++++++++++++++++ reputation/decision.go | 45 +++++++++++++++ reputation/htlc.go | 117 ++++++++++++++++++++++++++++++++++++++ reputation/htlc_test.go | 94 ++++++++++++++++++++++++++++++ 6 files changed, 504 insertions(+) create mode 100644 reputation/channel.go create mode 100644 reputation/config.go create mode 100644 reputation/config_test.go create mode 100644 reputation/decision.go create mode 100644 reputation/htlc.go create mode 100644 reputation/htlc_test.go diff --git a/reputation/channel.go b/reputation/channel.go new file mode 100644 index 00000000000..0b4cd140d71 --- /dev/null +++ b/reputation/channel.go @@ -0,0 +1,76 @@ +package reputation + +import "time" + +// channelReputation holds all per-channel reputation state. A single channel +// plays both roles: as an outgoing link it accrues reputation and holds the +// pending HTLCs it is responsible for; as an incoming link it accrues the +// revenue that sets its reputation threshold. +type channelReputation struct { + // outgoingReputation is the reputation this channel has accrued as an + // outgoing link. + outgoingReputation *decayingAverage + + // incomingRevenue is the revenue this channel has earned us as an + // incoming link. It is aggregated over several windows so that a peer + // cannot cheaply move its own threshold by manipulating recent + // forwarding. + incomingRevenue *aggregatedWindowAverage + + // pendingHTLCs tracks the in-flight HTLCs for which this channel is the + // outgoing link, keyed by their incoming circuit. + pendingHTLCs map[htlcRef]*pendingHTLC +} + +// newChannelReputation builds empty reputation state for a channel as of the +// provided start time. +func newChannelReputation(cfg Config, + start time.Time) *channelReputation { + + return &channelReputation{ + outgoingReputation: newDecayingAverage( + start, cfg.reputationWindow(), + ), + incomingRevenue: newAggregatedWindowAverage( + cfg.RevenueWindow, cfg.RevenueWindowCount, start, + ), + pendingHTLCs: make(map[htlcRef]*pendingHTLC), + } +} + +// inFlightRisk returns the total worst-case opportunity cost of the HTLCs +// already in flight on this channel as an outgoing link. Per BOLT #1280 only +// accountable HTLCs contribute: an unaccountable HTLC was never told it would +// be held liable, so it cannot dock reputation. +func (c *channelReputation) inFlightRisk() saturatedI64 { + var total saturatedI64 + + for _, p := range c.pendingHTLCs { + if !p.accountable { + continue + } + + total = total.Add(satFromUint(p.risk)) + } + + return total +} + +// sufficientReputation evaluates the reputation inequality +// +// outgoing_reputation - risk >= revenue_threshold +// +// against this (incoming) channel's revenue threshold, returning the verdict +// and the threshold value used. The caller chooses which risk to pass in. +func (c *channelReputation) sufficientReputation(risk saturatedI64, + outgoingReputation int64, at time.Time) (bool, int64, error) { + + threshold, err := c.incomingRevenue.valueAt(at) + if err != nil { + return false, 0, err + } + + net := satFromInt(outgoingReputation).Sub(risk) + + return net.Int64() >= threshold, threshold, nil +} diff --git a/reputation/config.go b/reputation/config.go new file mode 100644 index 00000000000..e6f76dfea92 --- /dev/null +++ b/reputation/config.go @@ -0,0 +1,97 @@ +package reputation + +import ( + "fmt" + "time" +) + +const ( + // defaultResolutionPeriod is the amount of time an HTLC is allowed to + // resolve in that classifies as "good" behaviour. The protocol allows + // for a 60s MPP timeout, so BOLT #1280 recommends 90s. + defaultResolutionPeriod = 90 * time.Second + + // defaultRevenueWindow is the largest cltv delta from the current block + // height that a node will allow before failing with expiry_too_far, + // expressed as a duration assuming 10 minute blocks (2016 blocks ~= 2 + // weeks). + defaultRevenueWindow = 2016 * 10 * time.Minute + + // defaultReputationMultiplier is the multiplier applied to the revenue + // window to determine the rolling window over which the outgoing + // channel's forwarding history is considered (default 12 => ~24 weeks). + // This sizes the outgoing-reputation window only. + defaultReputationMultiplier = 12 + + // defaultRevenueWindowCount is the number of rolling windows over which + // the incoming-revenue aggregated average is measured; BOLT #1280 + // (window_total) recommends at least 6. It is distinct from + // defaultReputationMultiplier, which sizes only the outgoing-reputation + // window. + defaultRevenueWindowCount = 6 + + // blockInterval is the assumed time per block, used to convert cltv + // deltas to durations. + blockInterval = 10 * time.Minute +) + +// Config holds the tunable parameters of the reputation subsystem. The +// zero value is not valid; use DefaultConfig and override as needed. +type Config struct { + // ResolutionPeriod is the duration within which an HTLC resolution is + // considered "good" behaviour (no opportunity cost). + ResolutionPeriod time.Duration + + // RevenueWindow is the rolling window over which incoming-channel + // revenue is measured. + RevenueWindow time.Duration + + // ReputationMultiplier scales RevenueWindow to give the (longer) window + // over which outgoing-channel reputation is measured. + ReputationMultiplier uint8 + + // RevenueWindowCount is the number of rolling RevenueWindow-sized + // windows over which the incoming-revenue aggregated average is + // measured. It is distinct from ReputationMultiplier, which sizes only + // the outgoing-reputation window. + RevenueWindowCount uint8 +} + +// DefaultConfig returns the recommended default configuration. +func DefaultConfig() Config { + return Config{ + ResolutionPeriod: defaultResolutionPeriod, + RevenueWindow: defaultRevenueWindow, + ReputationMultiplier: defaultReputationMultiplier, + RevenueWindowCount: defaultRevenueWindowCount, + } +} + +// Validate ensures the configuration is internally consistent. +func (c Config) Validate() error { + if c.ResolutionPeriod <= 0 { + return fmt.Errorf("resolution period must be positive, got %v", + c.ResolutionPeriod) + } + + if c.RevenueWindow <= 0 { + return fmt.Errorf("revenue window must be positive, got %v", + c.RevenueWindow) + } + + if c.ReputationMultiplier == 0 { + return fmt.Errorf("reputation multiplier must be positive") + } + + if c.RevenueWindowCount == 0 { + return fmt.Errorf("revenue window count must be positive") + } + + return nil +} + +// reputationWindow returns the rolling window over which outgoing-channel +// reputation is tracked. +func (c Config) reputationWindow() time.Duration { + return c.RevenueWindow * time.Duration(c.ReputationMultiplier) +} diff --git a/reputation/config_test.go b/reputation/config_test.go new file mode 100644 index 00000000000..880595fb224 --- /dev/null +++ b/reputation/config_test.go @@ -0,0 +1,75 @@ +package reputation + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestConfigValidate exercises the config validation table. +func TestConfigValidate(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cfg Config + wantErr bool + }{ + { + name: "default is valid", + cfg: DefaultConfig(), + }, + { + name: "zero resolution period invalid", + cfg: Config{ + RevenueWindow: time.Hour, + ReputationMultiplier: 12, + RevenueWindowCount: 6, + }, + wantErr: true, + }, + { + name: "zero revenue window invalid", + cfg: Config{ + ResolutionPeriod: time.Second, + ReputationMultiplier: 12, + RevenueWindowCount: 6, + }, + wantErr: true, + }, + { + name: "zero multiplier invalid", + cfg: Config{ + ResolutionPeriod: time.Second, + RevenueWindow: time.Hour, + RevenueWindowCount: 6, + }, + wantErr: true, + }, + { + name: "zero revenue window count invalid", + cfg: Config{ + ResolutionPeriod: time.Second, + RevenueWindow: time.Hour, + ReputationMultiplier: 12, + }, + wantErr: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + err := tc.cfg.Validate() + if tc.wantErr { + require.Error(t, err) + + return + } + + require.NoError(t, err) + }) + } +} diff --git a/reputation/decision.go b/reputation/decision.go new file mode 100644 index 00000000000..112d58fbfa7 --- /dev/null +++ b/reputation/decision.go @@ -0,0 +1,45 @@ +package reputation + +import "fmt" + +// decision is the result of evaluating the reputation inequality for an HTLC: +// +// sufficient = outgoingReputation - risk >= revenueThreshold +// +// It is evaluated two ways. inIsolation scores the HTLC on its own risk, which +// answers "could this HTLC stand on the outgoing channel's reputation if it +// were the only one in flight?". withInFlight additionally subtracts the risk +// of the accountable HTLCs already in flight on that channel, which is the +// verdict BOLT #1280 defines. Both are log-only: neither affects forwarding. +type decision struct { + // inIsolation reports whether the outgoing channel's reputation covers + // this HTLC's risk alone. + inIsolation bool + + // withInFlight reports whether it also covers the risk of the + // accountable HTLCs already in flight on the outgoing channel. + withInFlight bool + + // outgoingReputation is the outgoing channel's reputation at decision + // time. + outgoingReputation int64 + + // htlcRisk is the in-flight risk of this HTLC alone. + htlcRisk uint64 + + // totalRisk is htlcRisk plus the risk of the accountable HTLCs already + // in flight on the outgoing channel. + totalRisk int64 + + // threshold is the incoming channel's revenue threshold the reputation + // was compared against. + threshold int64 +} + +// String returns a human readable description of the decision for logging. +func (d decision) String() string { + return fmt.Sprintf("in_isolation=%v with_in_flight=%v "+ + "(outgoing_reputation=%d - htlc_risk=%d / total_risk=%d vs "+ + "threshold=%d)", d.inIsolation, d.withInFlight, + d.outgoingReputation, d.htlcRisk, d.totalRisk, d.threshold) +} diff --git a/reputation/htlc.go b/reputation/htlc.go new file mode 100644 index 00000000000..94476293bf3 --- /dev/null +++ b/reputation/htlc.go @@ -0,0 +1,117 @@ +package reputation + +import ( + "math" + "time" + + "github.com/lightningnetwork/lnd/graph/db/models" +) + +// htlcRef uniquely identifies an in-flight forwarded HTLC by its incoming +// circuit key. The pending HTLC is stored against its outgoing channel. +type htlcRef = models.CircuitKey + +// pendingHTLC captures, at forward time, the data the resolution path needs to +// score the HTLC that is not carried by the settle/fail hooks (which only +// identify the circuit). +type pendingHTLC struct { + // fee is the fee in millisatoshis that our policy requires to forward + // this HTLC. It is deliberately not the fee the sender chose to pay, + // which may be inflated: scoring on the required fee means an attacker + // has to make multiple payments rather than one over-paying payment to + // move a channel's reputation. + fee uint64 + + // accountable is the accountable signal as this node would forward it + // on the outgoing link, i.e. the bit received on the incoming link and + // only if this node forwards the experimental accountability signal at + // all. + accountable bool + + // addedAt is the time at which the HTLC was forwarded. + addedAt time.Time + + // maxHold is the worst-case duration for which the HTLC can be held, + // derived from its incoming cltv expiry. + maxHold time.Duration + + // risk is the worst-case opportunity cost of this HTLC while it is in + // flight (its opportunity cost over maxHold). It is computed once at + // forward time so that summing a channel's in-flight risk does not + // recompute it per pending HTLC. + risk uint64 +} + +// opportunityCost implements the BOLT #1280 opportunity_cost: +// +// max(0, (resolution_time - resolution_period)/resolution_period) * fees +// +// The spec value is real-valued; since reputation is tracked in integer +// millisatoshis we round to the nearest integer. +func (c Config) opportunityCost(resolutionTime time.Duration, + feeMsat uint64) uint64 { + + period := c.ResolutionPeriod.Seconds() + overrun := (resolutionTime.Seconds() - period) / period + if overrun < 0 { + overrun = 0 + } + + // overrun and feeMsat are both non-negative, so the product cannot be + // negative; clamp the high end where a very long hold on a large fee + // would exceed uint64 (an out-of-range float->uint conversion is + // undefined in Go). + cost := math.Round(overrun * float64(feeMsat)) + if cost >= float64(math.MaxUint64) { + return math.MaxUint64 + } + + return uint64(cost) +} + +// effectiveFee returns the contribution this HTLC makes to the outgoing +// channel's reputation, given its fee, resolution time, accountable signal and +// outcome. +func (c Config) effectiveFee(feeMsat uint64, resolutionTime time.Duration, + accountable, settled bool) int64 { + + fee := satFromUint(feeMsat) + + if accountable { + oc := satFromUint(c.opportunityCost(resolutionTime, feeMsat)) + if settled { + return fee.Sub(oc).Int64() + } + + return satFromInt(0).Sub(oc).Int64() + } + + // Unaccountable HTLCs can only ever help reputation: they earn their + // fee if they settle quickly, and contribute nothing otherwise. + if settled && resolutionTime <= c.ResolutionPeriod { + return fee.Int64() + } + + return 0 +} + +// maxHold returns the worst-case duration for which an HTLC may be held, +// derived from how far its incoming cltv expiry is from the height it was added +// at. A non-positive delta yields zero; callers validate that the incoming +// expiry is in the future before adding an HTLC. +func maxHold(incomingCltv, heightAdded uint32) time.Duration { + var delta uint32 + if incomingCltv > heightAdded { + delta = incomingCltv - heightAdded + } + + return time.Duration(delta) * blockInterval +} + +// inFlightRisk returns the worst-case opportunity cost of an in-flight HTLC, +// assuming it is held until just before its incoming cltv expiry. +func (c Config) inFlightRisk(feeMsat uint64, incomingCltv, + heightAdded uint32) uint64 { + + return c.opportunityCost(maxHold(incomingCltv, heightAdded), feeMsat) +} diff --git a/reputation/htlc_test.go b/reputation/htlc_test.go new file mode 100644 index 00000000000..ded6e9f3847 --- /dev/null +++ b/reputation/htlc_test.go @@ -0,0 +1,94 @@ +package reputation + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestOpportunityCostVectors checks opportunityCost against values computed +// directly from its formula +// max(0, (resolution_time - resolution_period)/resolution_period) * fees, with +// resolution_period = 90s and fee = 100. E.g. 135s -> (135-90)/90*100 = 50. +func TestOpportunityCostVectors(t *testing.T) { + t.Parallel() + + cfg := DefaultConfig() // ResolutionPeriod = 90s. + + tests := []struct { + resolution time.Duration + want uint64 + }{ + {10 * time.Second, 0}, + {90 * time.Second, 0}, + {91 * time.Second, 1}, + {135 * time.Second, 50}, + {180 * time.Second, 100}, + {900 * time.Second, 900}, + } + + for _, tc := range tests { + got := cfg.opportunityCost(tc.resolution, 100) + require.Equalf(t, tc.want, got, "opportunityCost(%v)", + tc.resolution) + } +} + +// TestEffectiveFeeMatrix covers all four branches of the effective-fee matrix. +func TestEffectiveFeeMatrix(t *testing.T) { + t.Parallel() + + cfg := DefaultConfig() + const fee = 1000 + + // Vectors covering the effective_fee matrix. fast (45s) is within the + // resolution period, so opportunity_cost = 0. slow (270s) gives + // opportunity_cost = (270-90)/90*fee = 2*fee = 2000, so the + // failed-accountable branch is -2000. + fast := cfg.ResolutionPeriod / 2 // 45s, within period. + slow := cfg.ResolutionPeriod * 3 // 270s. + + tests := []struct { + name string + resolution time.Duration + accountable bool + settled bool + want int64 + }{ + {"accountable settled fast", fast, true, true, fee}, + {"accountable settled slow", slow, true, true, -fee}, + {"accountable failed fast", fast, true, false, 0}, + {"accountable failed slow", slow, true, false, -2 * fee}, + {"unaccountable settled fast", fast, false, true, fee}, + {"unaccountable settled slow", slow, false, true, 0}, + {"unaccountable failed fast", fast, false, false, 0}, + {"unaccountable failed slow", slow, false, false, 0}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got := cfg.effectiveFee( + fee, tc.resolution, tc.accountable, tc.settled, + ) + require.Equal(t, tc.want, got) + }) + } +} + +// TestInFlightRisk checks the worst-case-hold opportunity cost. +func TestInFlightRisk(t *testing.T) { + t.Parallel() + + cfg := DefaultConfig() + + // cltv delta of 1 block = 600s hold. overrun = (600-90)/90 = 5.666..., + // * fee(100) = 566.67 -> round 567. + require.EqualValues(t, 567, cfg.inFlightRisk(100, 101, 100)) + + // No delta -> zero hold -> zero risk. + require.EqualValues(t, 0, cfg.inFlightRisk(100, 100, 100), + "zero cltv delta must carry no risk") +} From 78bd59567a893893cd68712046b75ce476363089 Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Tue, 18 Aug 2026 13:40:34 +0000 Subject: [PATCH 3/6] reputation: add the reputation manager Add the Manager that ties the scoring together behind the OnForward/OnSettle/ OnFail hooks. The hooks run synchronously under a single lock: OnForward records the pending HTLC and computes (and logs) the reputation decision, both for the HTLC in isolation and against the risk already in flight on its outgoing channel, while OnSettle/OnFail resolve it and update the outgoing reputation and incoming revenue averages. The subsystem is log-only and holds no persisted state, so reputation re-accrues from live traffic after a restart. Every resolution drops its own pending HTLC, so a pending that outlives the worst case time it could be held for means a resolution was never reported to us. A periodic check warns about those and deliberately leaves them in place rather than sweeping them away, so the underlying bug stays visible. Includes unit tests and benchmarks for the per-forward hook cost. --- reputation/bench_test.go | 74 ++++++ reputation/log.go | 30 +++ reputation/manager.go | 389 ++++++++++++++++++++++++++++++++ reputation/manager_test.go | 437 ++++++++++++++++++++++++++++++++++++ reputation/testutil_test.go | 27 +++ 5 files changed, 957 insertions(+) create mode 100644 reputation/bench_test.go create mode 100644 reputation/log.go create mode 100644 reputation/manager.go create mode 100644 reputation/manager_test.go create mode 100644 reputation/testutil_test.go diff --git a/reputation/bench_test.go b/reputation/bench_test.go new file mode 100644 index 00000000000..1dd56617980 --- /dev/null +++ b/reputation/bench_test.go @@ -0,0 +1,74 @@ +package reputation + +import ( + "testing" + "time" + + "github.com/lightningnetwork/lnd/clock" +) + +// BenchmarkForwardResolve measures the full per-HTLC cost the reputation +// subsystem adds to a forward: the synchronous OnForward hook (record pending + +// compute the decision) plus the OnSettle hook (resolve + update the averages). +// When the subsystem is disabled the switch skips these hooks behind a single +// nil check, so this is the "enabled minus disabled" overhead per forwarded +// HTLC. +func BenchmarkForwardResolve(b *testing.B) { + m, err := NewManager( + DefaultConfig(), clock.NewTestClock(time.Unix(1_000_000, 0)), + ) + if err != nil { + b.Fatalf("NewManager: %v", err) + } + + in := circuit(1, 0) + out := scid(2) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + m.OnForward(in, out, 2000, 1000, 1000, 200, 100, false) + m.OnSettle(in) + } +} + +// BenchmarkOnForward measures the cost of the forwarding hook alone (record +// pending + compute the reputation decision), which is the work that sits on +// the switch's forwarding path. Each iteration uses a distinct HTLC id and is +// resolved immediately so the pending map stays bounded. +func BenchmarkOnForward(b *testing.B) { + m, err := NewManager( + DefaultConfig(), clock.NewTestClock(time.Unix(1_000_000, 0)), + ) + if err != nil { + b.Fatalf("NewManager: %v", err) + } + + out := scid(2) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + in := circuit(1, uint64(i)) + m.OnForward(in, out, 2000, 1000, 1000, 200, 100, false) + + b.StopTimer() + m.OnSettle(in) + b.StartTimer() + } +} + +// BenchmarkDecayingAverageAdd measures the core decaying-average update, the +// hot primitive underlying every reputation and revenue mutation. +func BenchmarkDecayingAverageAdd(b *testing.B) { + d := newDecayingAverage(testStart, DefaultConfig().reputationWindow()) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ts := testStart.Add(time.Duration(i) * time.Second) + if _, err := d.add(1000, ts); err != nil { + b.Fatalf("add: %v", err) + } + } +} diff --git a/reputation/log.go b/reputation/log.go new file mode 100644 index 00000000000..48ef1bdac7e --- /dev/null +++ b/reputation/log.go @@ -0,0 +1,30 @@ +package reputation + +import ( + "github.com/btcsuite/btclog/v2" + "github.com/lightningnetwork/lnd/build" +) + +// Subsystem defines the logging code for this subsystem. +const Subsystem = "REPM" + +// log is a logger that is initialized with no output filters. This means the +// package will not perform any logging by default until the caller requests it. +var log btclog.Logger + +// The default amount of logging is none. +func init() { + UseLogger(build.NewSubLogger(Subsystem, nil)) +} + +// DisableLog disables all library log output. Logging output is disabled by +// default until UseLogger is called. +func DisableLog() { + UseLogger(btclog.Disabled) +} + +// UseLogger uses a specified Logger to output package logging info. This should +// be used in preference to SetLogWriter if the caller is also using btclog. +func UseLogger(logger btclog.Logger) { + log = logger +} diff --git a/reputation/manager.go b/reputation/manager.go new file mode 100644 index 00000000000..34b5539ece3 --- /dev/null +++ b/reputation/manager.go @@ -0,0 +1,389 @@ +package reputation + +import ( + "fmt" + "sync" + "time" + + "github.com/lightningnetwork/lnd/clock" + "github.com/lightningnetwork/lnd/graph/db/models" + "github.com/lightningnetwork/lnd/lnwire" +) + +// staleCheckInterval is how often the manager checks for pending HTLCs that +// have outlived the worst case time they could be held for. +const staleCheckInterval = 5 * time.Minute + +// Manager is the local reputation subsystem. It observes forwarded HTLCs via +// its OnForward/OnSettle/OnFail hooks, maintains per-channel reputation state, +// and logs the reputation decision it would make, without ever affecting +// routing (log-only). +// +// The hooks run synchronously on the caller's goroutine: they take the +// manager's lock, update the per-channel state, and return. The work per hook +// is a handful of map lookups and floating-point operations, so it is cheap +// enough to sit on the switch's forwarding path, and computing the decision +// inline (rather than on a background worker) is what a future enforcement step +// will require. Nothing is persisted, so reputation is re-accrued from live +// traffic after a restart. +type Manager struct { + cfg Config + clock clock.Clock + + // mu guards channels. It is held for the duration of each hook. + mu sync.Mutex + + // channels holds per-scid reputation state, created lazily on the first + // HTLC event for a channel. + channels map[uint64]*channelReputation + + // htlcIndex maps each pending HTLC's incoming circuit key to the scid + // of the outgoing channel it is pending on. Resolutions are looked up + // through this index because the switch's resolution paths do not + // reliably know the outgoing channel (a mailbox-failed add, for + // example, never had a keystone set). + htlcIndex map[models.CircuitKey]uint64 + + wg sync.WaitGroup + quit chan struct{} + startOnce sync.Once + stopOnce sync.Once +} + +// NewManager constructs a reputation Manager with the given config and clock. +// The clock is mandatory (production passes clock.NewDefaultClock; tests pass a +// test clock). +func NewManager(cfg Config, clk clock.Clock) (*Manager, error) { + if err := cfg.Validate(); err != nil { + return nil, fmt.Errorf("invalid reputation config: %w", err) + } + + if clk == nil { + return nil, fmt.Errorf("reputation manager requires a clock") + } + + return &Manager{ + cfg: cfg, + clock: clk, + channels: make(map[uint64]*channelReputation), + htlcIndex: make(map[models.CircuitKey]uint64), + quit: make(chan struct{}), + }, nil +} + +// Start launches the periodic stale-pending check. Per-channel state is created +// lazily on the first HTLC event, so there is nothing to load. +func (m *Manager) Start() error { + m.startOnce.Do(func() { + log.Infof("Reputation manager starting (log-only): "+ + "resolution_period=%v revenue_window=%v "+ + "reputation_multiplier=%d revenue_window_count=%d", + m.cfg.ResolutionPeriod, m.cfg.RevenueWindow, + m.cfg.ReputationMultiplier, m.cfg.RevenueWindowCount) + + m.wg.Add(1) + go m.staleCheckLoop() + }) + + return nil +} + +// Stop tears down the subsystem. +func (m *Manager) Stop() error { + m.stopOnce.Do(func() { + close(m.quit) + m.wg.Wait() + + log.Infof("Reputation manager stopped") + }) + + return nil +} + +// OnForward observes a forwarded HTLC at the point the switch commits to +// forwarding it. outgoing identifies the outgoing channel, advertisedFee is the +// total fee the node advertised for this forward (outbound plus inbound +// component, attributed to the outgoing channel), height is the current best +// block height, and accountable is the outgoing accountable signal. +func (m *Manager) OnForward(incoming models.CircuitKey, + outgoing lnwire.ShortChannelID, incomingAmt, outgoingAmt, + advertisedFee lnwire.MilliSatoshi, incomingCltv, height uint32, + accountable bool) { + + at := m.clock.Now() + + m.mu.Lock() + defer m.mu.Unlock() + + d, err := m.addHTLC( + incoming, outgoing, advertisedFee, incomingCltv, height, + accountable, at, + ) + if err != nil { + log.Warnf("Reputation OnForward(in=%v out=%v) error: %v", + incoming, outgoing, err) + + return + } + + // Emit the greppable decision line. This is log-only and never affects + // forwarding. + log.Infof("reputation decision: chan=%v htlc=%v in_isolation=%v "+ + "with_in_flight=%v", outgoing.ToUint64(), incoming, + d.inIsolation, d.withInFlight) + + log.Debugf("Reputation forward in=%v out=%v amt_in=%v amt_out=%v "+ + "advertised_fee=%v accountable=%v height=%d => %s", incoming, + outgoing, incomingAmt, outgoingAmt, advertisedFee, accountable, + height, d) +} + +// OnSettle observes the successful resolution of a forwarded HTLC, identified +// by its incoming circuit key. +func (m *Manager) OnSettle(incoming models.CircuitKey) { + m.resolve(incoming, true) +} + +// OnFail observes the failed resolution of a forwarded HTLC, identified by its +// incoming circuit key. +func (m *Manager) OnFail(incoming models.CircuitKey) { + m.resolve(incoming, false) +} + +// resolve applies an HTLC resolution under the lock. +func (m *Manager) resolve(incoming models.CircuitKey, settled bool) { + at := m.clock.Now() + + m.mu.Lock() + defer m.mu.Unlock() + + if err := m.resolveHTLC(incoming, settled, at); err != nil { + log.Warnf("Reputation resolve(in=%v settled=%v) error: %v", + incoming, settled, err) + } +} + +// getOrCreateChannel returns the reputation state for an scid, creating it +// lazily (zero reputation, initialised as of at) if it does not yet exist. +// Caller must hold mu. +func (m *Manager) getOrCreateChannel(scid uint64, + at time.Time) *channelReputation { + + if c, ok := m.channels[scid]; ok { + return c + } + + c := newChannelReputation(m.cfg, at) + m.channels[scid] = c + + return c +} + +// addHTLC records the pending HTLC and computes the (log-only) decision for it. +// Caller must hold mu. +func (m *Manager) addHTLC(incoming models.CircuitKey, + outgoing lnwire.ShortChannelID, advertisedFee lnwire.MilliSatoshi, + incomingCltv, height uint32, accountable bool, + at time.Time) (decision, error) { + + // The incoming expiry must be in the future; if it is not, something + // is badly wrong upstream (the HTLC should never have been accepted) + // and we cannot bound its hold time, so we refuse to track it. + if incomingCltv <= height { + return decision{}, fmt.Errorf("incoming cltv %d not beyond "+ + "current height %d", incomingCltv, height) + } + + outScid := outgoing.ToUint64() + inScid := incoming.ChanID.ToUint64() + + // Initialise any lazily-created channel state as of the event timestamp + // (a fresh clock read here could be a moment after `at`, causing this + // event to be rejected as backwards time). + outChan := m.getOrCreateChannel(outScid, at) + inChan := m.getOrCreateChannel(inScid, at) + + if _, ok := m.htlcIndex[incoming]; ok { + return decision{}, fmt.Errorf("duplicate htlc %v", incoming) + } + + // BOLT #1280 scores reputation on the fee the node advertised, not the + // offered fee, so a sender cannot inflate or destroy reputation by + // over/under-paying. + fee := uint64(advertisedFee) + risk := m.cfg.inFlightRisk(fee, incomingCltv, height) + htlcRisk := satFromUint(risk) + + // The total risk is this HTLC plus the accountable HTLCs already in + // flight on the outgoing channel, which is the risk BOLT #1280 uses. + totalRisk := htlcRisk.Add(outChan.inFlightRisk()) + + outReputation, err := outChan.outgoingReputation.valueAt(at) + if err != nil { + return decision{}, err + } + + // Score the HTLC both on its own risk and against the channel's total + // in-flight risk. + inIsolation, threshold, err := inChan.sufficientReputation( + htlcRisk, outReputation, at, + ) + if err != nil { + return decision{}, err + } + + withInFlight, _, err := inChan.sufficientReputation( + totalRisk, outReputation, at, + ) + if err != nil { + return decision{}, err + } + + outChan.pendingHTLCs[incoming] = &pendingHTLC{ + fee: fee, + accountable: accountable, + addedAt: at, + maxHold: maxHold(incomingCltv, height), + risk: risk, + } + m.htlcIndex[incoming] = outScid + + return decision{ + inIsolation: inIsolation, + withInFlight: withInFlight, + outgoingReputation: outReputation, + htlcRisk: uint64(htlcRisk.Int64()), + totalRisk: totalRisk.Int64(), + threshold: threshold, + }, nil +} + +// resolveHTLC applies an HTLC resolution to reputation and revenue. The pending +// HTLC is looked up by its incoming circuit key alone: the switch's resolution +// paths do not reliably know the outgoing channel, so it is recovered from the +// index recorded at forward time. Caller must hold mu. +func (m *Manager) resolveHTLC(incoming models.CircuitKey, settled bool, + at time.Time) error { + + outScid, ok := m.htlcIndex[incoming] + if !ok { + // Tolerate: we never saw the forward (e.g. enabled mid-flight). + log.Debugf("Reputation resolve for unmatched htlc %v; ignoring", + incoming) + + return nil + } + + // Drop the pending entry up front. The HTLC has resolved, so whatever + // happens below it must not be left in our in-flight view. + delete(m.htlcIndex, incoming) + + outChan, ok := m.channels[outScid] + if !ok { + return fmt.Errorf("htlc %v indexed to unknown outgoing "+ + "channel %d", incoming, outScid) + } + + pending, ok := outChan.pendingHTLCs[incoming] + if !ok { + return fmt.Errorf("htlc %v indexed to outgoing channel %d "+ + "but not pending on it", incoming, outScid) + } + + delete(outChan.pendingHTLCs, incoming) + + // The resolution cannot predate the add; if it does the clock went + // backwards and we cannot score this HTLC, so leave the averages alone. + if at.Before(pending.addedAt) { + return errBackwardsTime + } + + // Credit the incoming channel's revenue first: it does not depend on + // the reputation update below, so an error there must not starve it. + if settled { + inScid := incoming.ChanID.ToUint64() + inChan := m.getOrCreateChannel(inScid, at) + fee := satFromUint(pending.fee).Int64() + if _, err := inChan.incomingRevenue.add(fee, at); err != nil { + return err + } + } + + effFee := m.cfg.effectiveFee( + pending.fee, at.Sub(pending.addedAt), pending.accountable, + settled, + ) + + newRep, err := outChan.outgoingReputation.add(effFee, at) + if err != nil { + return err + } + + // Log a single greppable line per resolution reporting the reputation + // change, so it can be tracked without matching several phrasings. The + // phrasing is stable: integration tests match on it. + log.Infof("Reputation change: outgoing=%v eff_fee=%d "+ + "new_outgoing_reputation=%d settled=%v", outScid, effFee, + newRep, settled) + + return nil +} + +// staleCheckLoop runs the periodic stale-pending check until the manager is +// stopped. +func (m *Manager) staleCheckLoop() { + defer m.wg.Done() + + ticker := time.NewTicker(staleCheckInterval) + defer ticker.Stop() + + for { + select { + case <-m.quit: + return + + case <-ticker.C: + m.reportStalePendings() + } + } +} + +// reportStalePendings warns about pending HTLCs that have outlived the worst +// case time they could be held for, and returns how many it found. +// +// These entries are deliberately NOT evicted. Every resolution path removes its +// own pending, so a stale entry means the switch never reported a resolution to +// us, i.e. a bug on our side. Quietly sweeping it away would hide that bug, +// so instead it is reported and left in place, where it keeps contributing to +// the channel's in-flight risk and stays visible. +func (m *Manager) reportStalePendings() int { + at := m.clock.Now() + + m.mu.Lock() + defer m.mu.Unlock() + + var stale int + for scid, ch := range m.channels { + for ref, p := range ch.pendingHTLCs { + if at.Before(p.addedAt.Add(p.maxHold)) { + continue + } + + stale++ + + log.Warnf("Reputation has pending htlc %v on outgoing "+ + "channel %d that outlived its maximum hold "+ + "time (added_at=%v, max_hold=%v): its "+ + "resolution was never observed", ref, scid, + p.addedAt, p.maxHold) + } + } + + if stale > 0 { + log.Warnf("Reputation is tracking %d pending HTLC(s) past "+ + "their maximum hold time; the in-flight view has "+ + "diverged from the switch", stale) + } + + return stale +} diff --git a/reputation/manager_test.go b/reputation/manager_test.go new file mode 100644 index 00000000000..cc506e3bc0b --- /dev/null +++ b/reputation/manager_test.go @@ -0,0 +1,437 @@ +package reputation + +import ( + "testing" + "time" + + "github.com/lightningnetwork/lnd/clock" + "github.com/stretchr/testify/require" +) + +// testHeight is the fixed best block height used by the manager tests. Forward +// events use an incoming cltv comfortably beyond it. +const testHeight = uint32(100) + +// buildManager returns a started manager with an injected test clock at +// `start`. Per-channel state (incoming scid=1, outgoing scid=2) is created +// lazily on the first HTLC event. +func buildManager(t *testing.T, start int64) (*Manager, *clock.TestClock) { + t.Helper() + + clk := clock.NewTestClock(time.Unix(start, 0)) + + m, err := NewManager(DefaultConfig(), clk) + require.NoError(t, err, "NewManager") + require.NoError(t, m.Start(), "Start") + t.Cleanup(func() { _ = m.Stop() }) + + return m, clk +} + +// TestManagerStartStop is a smoke test for the lifecycle. +func TestManagerStartStop(t *testing.T) { + t.Parallel() + + m, err := NewManager( + DefaultConfig(), clock.NewTestClock(time.Unix(1000, 0)), + ) + require.NoError(t, err) + require.NoError(t, m.Start()) + + // Hooks on an empty manager must be safe no-ops (other than lazy + // channel creation): they must never panic. + m.OnSettle(circuit(1, 0)) + m.OnFail(circuit(1, 0)) + + require.NoError(t, m.Stop()) +} + +// TestManagerRequiresClock checks that a manager cannot be built without a +// clock, since production and tests must both supply a real time source. +func TestManagerRequiresClock(t *testing.T) { + t.Parallel() + + _, err := NewManager(DefaultConfig(), nil) + require.Error(t, err) +} + +// TestForwardSettleLifecycle exercises the pending lifecycle + reputation +// accrual: an unaccountable HTLC that settles quickly earns its fee. Because +// the hooks are synchronous, the effects are observable as soon as they return. +func TestForwardSettleLifecycle(t *testing.T) { + t.Parallel() + + const start = 1_000_000 + m, clk := buildManager(t, start) + + in := circuit(1, 0) + out := scid(2) + + // Required fee = 1000 (equals in-out here). cltv 200 > height 100. + m.OnForward(in, out, 2000, 1000, 1000, 200, testHeight, false) + + outChan := m.channels[2] + require.Len(t, outChan.pendingHTLCs, 1) + + // Settle 30s later (within resolution period). + advance(clk, 30*time.Second) + m.OnSettle(in) + + require.Empty(t, outChan.pendingHTLCs, "pending not cleared") + require.Empty(t, m.htlcIndex, "htlc index not cleared") + + rep, err := outChan.outgoingReputation.valueAt(clk.Now()) + require.NoError(t, err) + require.EqualValues(t, 1000, rep, "reputation") + + // Incoming channel earned the fee as revenue. + rev, err := m.channels[1].incomingRevenue.valueAt(clk.Now()) + require.NoError(t, err) + require.Positive(t, rev, "revenue") +} + +// TestFailDoesNotEarnRevenue checks that a failed unaccountable HTLC neither +// helps reputation nor adds revenue. +func TestFailDoesNotEarnRevenue(t *testing.T) { + t.Parallel() + + m, clk := buildManager(t, 1_000_000) + in, out := circuit(1, 0), scid(2) + + m.OnForward(in, out, 2000, 1000, 1000, 200, testHeight, false) + advance(clk, 30*time.Second) + m.OnFail(in) + + rep, err := m.channels[2].outgoingReputation.valueAt(clk.Now()) + require.NoError(t, err) + require.EqualValues(t, 0, rep, "reputation after fail") + + rev, err := m.channels[1].incomingRevenue.valueAt(clk.Now()) + require.NoError(t, err) + require.EqualValues(t, 0, rev, "revenue after fail") +} + +// TestAccountableResolution drives accountable HTLCs through the full +// forward/resolve path. Unlike unaccountable HTLCs, accountable ones are +// charged the opportunity cost of the time they held the outgoing slot, so +// they are the only way a channel's reputation can decrease. +func TestAccountableResolution(t *testing.T) { + t.Parallel() + + // The default resolution period is 90s, so resolving at 270s overruns + // it by exactly 2x: opportunity cost = 2 * fee = 2000. + const ( + fee = 1000 + fast = 30 * time.Second + slow = 270 * time.Second + ) + + tests := []struct { + name string + hold time.Duration + settled bool + wantRep int64 + wantRev int64 + }{{ + // Settling within the resolution period costs nothing, so the + // HTLC earns its full fee just like an unaccountable one. + name: "settled fast earns fee", + hold: fast, + settled: true, + wantRep: fee, + wantRev: fee, + }, { + // fee - 2*fee = -fee: holding the slot for too long costs more + // than the forward earned. + name: "settled slow costs reputation", + hold: slow, + settled: true, + wantRep: -fee, + wantRev: fee, + }, { + // A fast failure has no opportunity cost, but earns nothing + // either. + name: "failed fast is neutral", + hold: fast, + settled: false, + wantRep: 0, + wantRev: 0, + }, { + // A slow failure is pure cost: the fee was never earned, so + // only the opportunity cost applies. + name: "failed slow costs reputation", + hold: slow, + settled: false, + wantRep: -2 * fee, + wantRev: 0, + }} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + m, clk := buildManager(t, 1_000_000) + in, out := circuit(1, 0), scid(2) + + m.OnForward( + in, out, 2000, 1000, fee, 200, testHeight, true, + ) + + advance(clk, test.hold) + if test.settled { + m.OnSettle(in) + } else { + m.OnFail(in) + } + + outChan := m.channels[2] + require.Empty( + t, outChan.pendingHTLCs, "pending not cleared", + ) + + outRep := outChan.outgoingReputation + rep, err := outRep.valueAt(clk.Now()) + require.NoError(t, err) + require.Equal(t, test.wantRep, rep, "reputation") + + rev, err := m.channels[1].incomingRevenue.valueAt( + clk.Now(), + ) + require.NoError(t, err) + require.Equal(t, test.wantRev, rev, "revenue") + }) + } +} + +// TestUnmatchedResolveNoop ensures a resolve with no matching forward is a safe +// no-op (tolerating a missed add / mid-flight enable). +func TestUnmatchedResolveNoop(t *testing.T) { + t.Parallel() + + m, _ := buildManager(t, 1_000_000) + + // Should not panic or error fatally. + m.OnSettle(circuit(1, 99)) + m.OnFail(circuit(1, 88)) +} + +// TestForwardRejectsExpiredCltv verifies OnForward refuses to track an HTLC +// whose incoming expiry is not beyond the current height (a condition that +// should never occur for a validly-accepted HTLC), leaving no pending state. +func TestForwardRejectsExpiredCltv(t *testing.T) { + t.Parallel() + + m, _ := buildManager(t, 1_000_000) + in, out := circuit(1, 0), scid(2) + + // incoming cltv == height: not beyond, must be rejected. + m.OnForward(in, out, 2000, 1000, 1000, testHeight, testHeight, false) + + if c := m.channels[2]; c != nil { + require.Empty(t, c.pendingHTLCs, + "expired-cltv forward must not create a pending htlc") + } +} + +// TestStalePendingReported verifies that a pending HTLC which outlives its +// maximum hold time is reported, and deliberately NOT evicted: every resolution +// path removes its own pending, so a stale entry is a bug on our side that must +// stay visible rather than be quietly swept away. +func TestStalePendingReported(t *testing.T) { + t.Parallel() + + m, clk := buildManager(t, 1_000_000) + + // cltv 101 at height 100 => max hold 600s. + _, err := m.addHTLC( + circuit(1, 0), scid(2), 1000, 101, testHeight, false, clk.Now(), + ) + require.NoError(t, err, "addHTLC") + + // Before the maximum hold elapses nothing is stale. + require.Zero(t, m.reportStalePendings()) + + // Past the maximum hold it is reported, but left in place. + advance(clk, 700*time.Second) + require.Equal(t, 1, m.reportStalePendings()) + require.Len(t, m.channels[2].pendingHTLCs, 1, + "stale pending must not be swept away") +} + +// TestSufficiencyBoundary unit-tests the core reputation inequality at its +// boundary. +func TestSufficiencyBoundary(t *testing.T) { + t.Parallel() + + cfg := DefaultConfig() + c := newChannelReputation(cfg, testStart) + + // Seed an incoming-revenue threshold of 1000 and read it back so the + // aggregated average's warmup divisor is settled. + _, err := c.incomingRevenue.add(1000, testStart) + require.NoError(t, err) + + threshold, err := c.incomingRevenue.valueAt(testStart) + require.NoError(t, err) + + // Reputation exactly at threshold, no in-flight risk => sufficient. + noRisk := satFromInt(0) + + ok, _, err := c.sufficientReputation(noRisk, threshold, testStart) + require.NoError(t, err) + require.True(t, ok, "expected sufficient at threshold") + + // One msat below threshold => insufficient. + ok, _, err = c.sufficientReputation(noRisk, threshold-1, testStart) + require.NoError(t, err) + require.False(t, ok, "expected insufficient below threshold") + + // At threshold but with in-flight risk => insufficient. + ok, _, err = c.sufficientReputation(satFromInt(1), threshold, testStart) + require.NoError(t, err) + require.False(t, ok, "expected insufficient with in-flight risk") +} + +// TestReputationDecision drives the log-only reputation verdict through the +// addHTLC path: zero reputation is insufficient, while ample reputation on the +// outgoing channel is sufficient. +func TestReputationDecision(t *testing.T) { + t.Parallel() + + const start = 1_000_000 + at := time.Unix(start, 0) + + addHTLC := func(m *Manager) (decision, error) { + return m.addHTLC( + circuit(1, 0), scid(2), 1000, 200, testHeight, true, at, + ) + } + + t.Run("zero reputation insufficient", func(t *testing.T) { + t.Parallel() + + m, _ := buildManager(t, start) + + // Give the incoming channel a positive revenue threshold so the + // (zero) outgoing reputation is insufficient. + inChan := m.getOrCreateChannel(1, at) + _, err := inChan.incomingRevenue.add(1_000_000, at) + require.NoError(t, err, "seed revenue") + + d, err := addHTLC(m) + require.NoError(t, err, "addHTLC") + require.False(t, d.inIsolation, "expected insufficient: %s", d) + }) + + t.Run("ample reputation sufficient", func(t *testing.T) { + t.Parallel() + + m, _ := buildManager(t, start) + + // Give the outgoing channel ample reputation. + outChan := m.getOrCreateChannel(2, at) + _, err := outChan.outgoingReputation.add(10_000_000, at) + require.NoError(t, err, "seed reputation") + + d, err := addHTLC(m) + require.NoError(t, err, "addHTLC") + require.True(t, d.inIsolation, "expected sufficient: %s", d) + }) +} + +// TestInFlightRiskDecision checks that withInFlight, the verdict BOLT #1280 +// defines, also charges the risk of the accountable HTLCs already pending on +// the outgoing channel, while unaccountable ones add none. +func TestInFlightRiskDecision(t *testing.T) { + t.Parallel() + + // A cltv delta of 1 block gives a 600s hold, so each HTLC carries a + // risk of round((600-90)/90 * 1000) = 5667. The outgoing reputation of + // 12000 covers one HTLC's risk, and two of them, but not three. + const ( + fee = 1000 + cltv = testHeight + 1 + risk = 5667 + reputation = 12_000 + ) + + tests := []struct { + name string + + // prior holds the accountable bit of each HTLC already in + // flight on the outgoing channel. + prior []bool + + wantTotalRisk int64 + wantWithInFlight bool + }{{ + name: "nothing in flight", + wantTotalRisk: risk, + wantWithInFlight: true, + }, { + name: "accountable in flight", + prior: []bool{true, true}, + wantTotalRisk: 3 * risk, + wantWithInFlight: false, + }, { + // Counting the unaccountable HTLC would make it 3 * risk and + // flip the verdict. + name: "unaccountable in flight adds no risk", + prior: []bool{false, true}, + wantTotalRisk: 2 * risk, + wantWithInFlight: true, + }} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + m, clk := buildManager(t, 1_000_000) + at := clk.Now() + + outChan := m.getOrCreateChannel(2, at) + _, err := outChan.outgoingReputation.add(reputation, at) + require.NoError(t, err, "seed reputation") + + for i, accountable := range test.prior { + _, err := m.addHTLC( + circuit(1, uint64(i)), scid(2), fee, + cltv, testHeight, accountable, at, + ) + require.NoError(t, err, "add prior htlc") + } + + d, err := m.addHTLC( + circuit(1, 99), scid(2), fee, cltv, testHeight, + true, at, + ) + require.NoError(t, err, "addHTLC") + + require.True(t, d.inIsolation, "in isolation: %s", d) + require.Equal(t, test.wantTotalRisk, d.totalRisk) + require.Equal( + t, test.wantWithInFlight, d.withInFlight, + "with in flight: %s", d, + ) + }) + } +} + +// TestDuplicateForwardRejected checks that a second forward for an incoming +// circuit already pending is rejected without touching the first. Otherwise the +// index would move to the new channel and orphan the original pending HTLC, +// which would then never resolve and keep charging its channel's in-flight +// risk. +func TestDuplicateForwardRejected(t *testing.T) { + t.Parallel() + + m, _ := buildManager(t, 1_000_000) + in := circuit(1, 0) + + m.OnForward(in, scid(2), 2000, 1000, 1000, 200, testHeight, true) + m.OnForward(in, scid(3), 2000, 1000, 1000, 200, testHeight, true) + + require.EqualValues(t, 2, m.htlcIndex[in], "index moved") + require.Len(t, m.channels[2].pendingHTLCs, 1, "original pending lost") + require.Empty(t, m.channels[3].pendingHTLCs, "duplicate recorded") +} diff --git a/reputation/testutil_test.go b/reputation/testutil_test.go new file mode 100644 index 00000000000..9d605e22ca6 --- /dev/null +++ b/reputation/testutil_test.go @@ -0,0 +1,27 @@ +package reputation + +import ( + "time" + + "github.com/lightningnetwork/lnd/clock" + "github.com/lightningnetwork/lnd/graph/db/models" + "github.com/lightningnetwork/lnd/lnwire" +) + +// circuit builds a CircuitKey from an scid int and htlc id. +func circuit(scidInt, htlcID uint64) models.CircuitKey { + return models.CircuitKey{ + ChanID: lnwire.NewShortChanIDFromInt(scidInt), + HtlcID: htlcID, + } +} + +// scid builds a ShortChannelID from its integer representation. +func scid(v uint64) lnwire.ShortChannelID { + return lnwire.NewShortChanIDFromInt(v) +} + +// advance moves a test clock forward by d. +func advance(c *clock.TestClock, d time.Duration) { + c.SetTime(c.Now().Add(d)) +} From 8915d228ae38a94fdedf5f3751f306888c28f4ec Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Tue, 18 Aug 2026 13:40:34 +0000 Subject: [PATCH 4/6] htlcswitch+lnd: connect reputation manager to the switch Feed forwarded HTLCs to the reputation subsystem through a read-only seam on the switch. The switch calls OnForward/OnSettle/OnFail at the circuit layer behind a nil check, so the subsystem is skipped entirely when disabled. The manager is wrapped in a panic boundary before being handed to the switch: a bug in the (log-only) subsystem can never take down HTLC forwarding. Only the outgoing channel is reported to the subsystem, not an outgoing circuit key: at forward time the switch has not yet handed the packet to the outgoing link, so no outgoing HTLC ID exists yet. The subsystem is enabled by default and can be disabled with the new routing.no-reputation flag. Includes unit tests for the switch seam: each hook fires once with the right keys, a nil manager is a no-op, local sends are skipped, a hook panic is absorbed by the guard, and a non-strict forward reports the channel the HTLC actually went out on for both the add and its resolution. --- htlcswitch/interfaces.go | 44 ++ htlcswitch/link.go | 47 +- htlcswitch/link_test.go | 40 ++ htlcswitch/mock.go | 10 + htlcswitch/reputation_guard.go | 68 +++ htlcswitch/reputation_hooks_test.go | 856 ++++++++++++++++++++++++++++ htlcswitch/switch.go | 94 +++ lncfg/routing.go | 2 + log.go | 4 + sample-lnd.conf | 5 + server.go | 48 ++ 11 files changed, 1205 insertions(+), 13 deletions(-) create mode 100644 htlcswitch/reputation_guard.go create mode 100644 htlcswitch/reputation_hooks_test.go diff --git a/htlcswitch/interfaces.go b/htlcswitch/interfaces.go index ef62eb71d93..fe513d265f3 100644 --- a/htlcswitch/interfaces.go +++ b/htlcswitch/interfaces.go @@ -279,6 +279,13 @@ type ChannelLink interface { // policy to govern if it an incoming HTLC should be forwarded or not. UpdateForwardingPolicy(models.ForwardingPolicy) + // AdvertisedFee returns the fee this link's current forwarding policy + // charges to forward the given outgoing amount (base fee plus the + // proportional fee). It is the fee the node advertised for this link, + // as distinct from the (possibly larger) fee actually offered by the + // incoming HTLC. + AdvertisedFee(amtToForward lnwire.MilliSatoshi) lnwire.MilliSatoshi + // CheckHtlcForward should return a nil error if the passed HTLC details // satisfy the current forwarding policy fo the target link. Otherwise, // a LinkError with a valid protocol failure message should be returned @@ -515,6 +522,43 @@ type htlcNotifier interface { info channeldb.FinalHtlcInfo) } +// ReputationManager is the read-only seam through which the switch feeds HTLC +// forwarding lifecycle events to the (optional) local reputation subsystem. +// It is a black box that only observes events to update internal reputation +// state; it never affects forwarding decisions or the wire (log-only). When no +// reputation manager is configured this is nil and the hooks are skipped. +type ReputationManager interface { + // OnForward observes a forwarded HTLC at the point the switch + // commits to forwarding it to the outgoing channel. advertisedFee is + // the total fee the node advertised for this forward (the outgoing + // link's outbound fee plus the incoming link's inbound fee, clamped + // at zero; not the fee offered by the incoming HTLC), height is the + // switch's current best block height, and accountable is the outgoing + // accountable bit as this node would forward it. + // + // Only the outgoing channel is identified, not the outgoing HTLC: at + // this point the switch has not yet handed the packet to the outgoing + // link, so no outgoing HTLC ID has been assigned. HTLCs are therefore + // tracked by their incoming circuit key, which is stable for the whole + // lifecycle. + OnForward(incoming CircuitKey, outgoing lnwire.ShortChannelID, + incomingAmt, outgoingAmt, advertisedFee lnwire.MilliSatoshi, + incomingCltv, height uint32, accountable bool) + + // OnSettle observes the successful resolution of a forwarded HTLC. + // + // Resolutions identify the HTLC by its incoming circuit key alone: + // not every resolution path knows the outgoing channel (an add failed + // back through the outgoing link's mailbox, for example, never had a + // keystone set), so the manager matches resolutions to the forwards + // it recorded by circuit key. + OnSettle(incoming CircuitKey) + + // OnFail observes the failed resolution of a forwarded HTLC. See + // OnSettle for how the HTLC is identified. + OnFail(incoming CircuitKey) +} + // AuxHtlcModifier is an interface that allows the sender to modify the outgoing // HTLC of a payment by changing the amount or the wire message tlv records. type AuxHtlcModifier interface { diff --git a/htlcswitch/link.go b/htlcswitch/link.go index 9e3adf0bbc2..3b427250b82 100644 --- a/htlcswitch/link.go +++ b/htlcswitch/link.go @@ -79,6 +79,24 @@ func ExpectedFee(f models.ForwardingPolicy, return f.BaseFee + (htlcAmt*f.FeeRate)/1000000 } +// TotalForwardingFee returns the total fee this node charges to forward +// amtToForward: the outbound fee the outgoing link charges on the outgoing +// amount, plus the inbound fee of the incoming link, which is charged on the +// outgoing amount plus the outbound fee. +// +// The two components are calculated and rounded separately on purpose. An +// aggregate fee applied to the outgoing amount may round slightly higher than +// the sum of the separately rounded components, which would cause failed +// forwards for senders. The result is signed because the inbound component may +// be a discount. +func TotalForwardingFee(amtToForward, outFee lnwire.MilliSatoshi, + inboundFee models.InboundFee) int64 { + + inFee := inboundFee.CalcFee(amtToForward + outFee) + + return inFee + int64(outFee) +} + // ChannelLinkConfig defines the configuration for the channel link. ALL // elements within the configuration MUST be non-nil for channel link to carry // out its duties. @@ -2482,6 +2500,20 @@ func (l *channelLink) UpdateForwardingPolicy( l.cfg.FwrdingPolicy = newPolicy } +// AdvertisedFee returns the fee this link's current forwarding policy charges +// to forward the given outgoing amount (base fee plus the proportional fee). +// +// NOTE: Part of the ChannelLink interface. +func (l *channelLink) AdvertisedFee( + amtToForward lnwire.MilliSatoshi) lnwire.MilliSatoshi { + + l.RLock() + policy := l.cfg.FwrdingPolicy + l.RUnlock() + + return ExpectedFee(policy, amtToForward) +} + // CheckHtlcForward should return a nil error if the passed HTLC details // satisfy the current forwarding policy fo the target link. Otherwise, // a LinkError with a valid protocol failure message should be returned @@ -2501,20 +2533,9 @@ func (l *channelLink) CheckHtlcForward(payHash [32]byte, incomingHtlcAmt, // Using the outgoing HTLC amount, we'll calculate the outgoing // fee this incoming HTLC must carry in order to satisfy the constraints - // of the outgoing link. + // of the outgoing link, then add the inbound fee we charge on top. outFee := ExpectedFee(policy, amtToForward) - - // Then calculate the inbound fee that we charge based on the sum of - // outgoing HTLC amount and outgoing fee. - inFee := inboundFee.CalcFee(amtToForward + outFee) - - // Add up both fee components. It is important to calculate both fees - // separately. An alternative way of calculating is to first determine - // an aggregate fee and apply that to the outgoing HTLC amount. However, - // rounding may cause the result to be slightly higher than in the case - // of separately rounded fee components. This potentially causes failed - // forwards for senders and is something to be avoided. - expectedFee := inFee + int64(outFee) + expectedFee := TotalForwardingFee(amtToForward, outFee, inboundFee) // If the actual fee is less than our expected fee, then we'll reject // this HTLC as it didn't provide a sufficient amount of fees, or the diff --git a/htlcswitch/link_test.go b/htlcswitch/link_test.go index 28760897551..abfd577065c 100644 --- a/htlcswitch/link_test.go +++ b/htlcswitch/link_test.go @@ -7764,3 +7764,43 @@ func TestLinkQuiescenceExitHopProcessingDeferred(t *testing.T) { // TODO(proofofkeags): make sure these actions are run on resume. } + +// TestTotalForwardingFee checks that the total forwarding fee is the outbound +// fee plus the inbound fee charged on the outgoing amount plus the outbound +// fee, and that an inbound discount yields a signed, possibly negative, total. +func TestTotalForwardingFee(t *testing.T) { + t.Parallel() + + const ( + amt = lnwire.MilliSatoshi(100_000) + outFee = lnwire.MilliSatoshi(1_000) + ) + + tests := []struct { + name string + inboundFee models.InboundFee + want int64 + }{{ + name: "no inbound fee", + want: 1_000, + }, { + // 500 base + 1% of (100_000 + 1_000) = 500 + 1_010. + name: "inbound fee on amount plus outbound fee", + inboundFee: models.InboundFee{Base: 500, Rate: 10_000}, + want: 2_510, + }, { + // A discount larger than the outbound fee goes negative. + name: "inbound discount", + inboundFee: models.InboundFee{Base: -1_500}, + want: -500, + }} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + got := TotalForwardingFee(amt, outFee, test.inboundFee) + require.Equal(t, test.want, got) + }) + } +} diff --git a/htlcswitch/mock.go b/htlcswitch/mock.go index a3079e69628..ac433552a98 100644 --- a/htlcswitch/mock.go +++ b/htlcswitch/mock.go @@ -738,6 +738,10 @@ type mockChannelLink struct { checkHtlcForwardResult *LinkError + // advertisedFee is the fee returned by AdvertisedFee, letting tests + // control the outgoing link's advertised forwarding fee. + advertisedFee lnwire.MilliSatoshi + failAliasUpdate func(sid lnwire.ShortChannelID, incoming bool) *lnwire.ChannelUpdate1 @@ -847,6 +851,12 @@ func (f *mockChannelLink) HandleChannelUpdate(lnwire.Message) { func (f *mockChannelLink) UpdateForwardingPolicy(_ models.ForwardingPolicy) { } + +func (f *mockChannelLink) AdvertisedFee( + _ lnwire.MilliSatoshi) lnwire.MilliSatoshi { + + return f.advertisedFee +} func (f *mockChannelLink) CheckHtlcForward([32]byte, lnwire.MilliSatoshi, lnwire.MilliSatoshi, uint32, uint32, models.InboundFee, uint32, lnwire.ShortChannelID, lnwire.CustomRecords) *LinkError { diff --git a/htlcswitch/reputation_guard.go b/htlcswitch/reputation_guard.go new file mode 100644 index 00000000000..f262a4fa92f --- /dev/null +++ b/htlcswitch/reputation_guard.go @@ -0,0 +1,68 @@ +package htlcswitch + +import ( + "github.com/lightningnetwork/lnd/lnwire" +) + +// guardedReputationManager wraps a ReputationManager so that a panic in any of +// its hooks can never propagate into the switch's forwarding goroutine. The +// reputation subsystem is log-only and MUST NOT be able to degrade forwarding; +// if a hook panics we log it and carry on forwarding. +// +// The hooks run synchronously on the switch's forwarding goroutine, so this +// boundary keeps a subsystem bug, such as a nil deref or an arithmetic panic, +// from taking down the node's HTLC forwarding. +type guardedReputationManager struct { + inner ReputationManager +} + +// NewGuardedReputationManager wraps the given ReputationManager with a panic +// boundary. It returns nil when inner is nil, so the switch's existing nil +// check still short-circuits a disabled subsystem with zero overhead. +func NewGuardedReputationManager(inner ReputationManager) ReputationManager { + if inner == nil { + return nil + } + + return &guardedReputationManager{inner: inner} +} + +// OnForward forwards the observation to the wrapped manager behind a panic +// boundary. +func (g *guardedReputationManager) OnForward(incoming CircuitKey, + outgoing lnwire.ShortChannelID, incomingAmt, outgoingAmt, + advertisedFee lnwire.MilliSatoshi, incomingCltv, height uint32, + accountable bool) { + + defer g.recoverHook("OnForward") + + g.inner.OnForward( + incoming, outgoing, incomingAmt, outgoingAmt, advertisedFee, + incomingCltv, height, accountable, + ) +} + +// OnSettle forwards the observation to the wrapped manager behind a panic +// boundary. +func (g *guardedReputationManager) OnSettle(incoming CircuitKey) { + defer g.recoverHook("OnSettle") + + g.inner.OnSettle(incoming) +} + +// OnFail forwards the observation to the wrapped manager behind a panic +// boundary. +func (g *guardedReputationManager) OnFail(incoming CircuitKey) { + defer g.recoverHook("OnFail") + + g.inner.OnFail(incoming) +} + +// recoverHook recovers from a panic in a reputation hook and logs it, so that a +// bug in the log-only subsystem cannot affect forwarding. +func (g *guardedReputationManager) recoverHook(method string) { + if r := recover(); r != nil { + log.Errorf("Reputation %s hook panicked (forwarding is "+ + "unaffected): %v", method, r) + } +} diff --git a/htlcswitch/reputation_hooks_test.go b/htlcswitch/reputation_hooks_test.go new file mode 100644 index 00000000000..6c7811ecc2d --- /dev/null +++ b/htlcswitch/reputation_hooks_test.go @@ -0,0 +1,856 @@ +package htlcswitch + +import ( + "crypto/sha256" + "sync" + "testing" + "time" + + "github.com/lightningnetwork/lnd/graph/db/models" + "github.com/lightningnetwork/lnd/lnwire" + "github.com/stretchr/testify/require" +) + +// mockReputationManager is a stub ReputationManager that records the hook calls +// the switch makes, used to assert the read-only reputation seam fires exactly +// once per forward/settle/fail with the correct circuit keys. +type mockReputationManager struct { + mu sync.Mutex + forwards []repForward + settles []repResolve + fails []repResolve +} + +type repForward struct { + in CircuitKey + out lnwire.ShortChannelID + inAmt, outAmt lnwire.MilliSatoshi + advertisedFee lnwire.MilliSatoshi + cltv uint32 + height uint32 + accountable bool +} + +type repResolve struct { + in CircuitKey +} + +func (r *mockReputationManager) OnForward(in CircuitKey, + out lnwire.ShortChannelID, inAmt, outAmt, + advertisedFee lnwire.MilliSatoshi, cltv, height uint32, + accountable bool) { + + r.mu.Lock() + defer r.mu.Unlock() + r.forwards = append(r.forwards, repForward{ + in: in, out: out, inAmt: inAmt, outAmt: outAmt, + advertisedFee: advertisedFee, cltv: cltv, height: height, + accountable: accountable, + }) +} + +func (r *mockReputationManager) OnSettle(in CircuitKey) { + r.mu.Lock() + defer r.mu.Unlock() + r.settles = append(r.settles, repResolve{in: in}) +} + +func (r *mockReputationManager) OnFail(in CircuitKey) { + r.mu.Lock() + defer r.mu.Unlock() + r.fails = append(r.fails, repResolve{in: in}) +} + +func (r *mockReputationManager) snapshot() ([]repForward, []repResolve, + []repResolve) { + + r.mu.Lock() + defer r.mu.Unlock() + + return append([]repForward(nil), r.forwards...), + append([]repResolve(nil), r.settles...), + append([]repResolve(nil), r.fails...) +} + +// newReputationTestSwitch builds a switch with the given (possibly nil) +// reputation manager wired in, plus two linked mock channels (alice -> bob). +func newReputationTestSwitch(t *testing.T, repMgr ReputationManager) (*Switch, + *mockChannelLink, *mockChannelLink) { + + t.Helper() + + alicePeer, err := newMockServer( + t, "alice", testStartingHeight, nil, testDefaultDelta, + ) + if err != nil { + t.Fatalf("unable to create alice server: %v", err) + } + bobPeer, err := newMockServer( + t, "bob", testStartingHeight, nil, testDefaultDelta, + ) + if err != nil { + t.Fatalf("unable to create bob server: %v", err) + } + + s, err := initSwitchWithTempDB(t, testStartingHeight) + if err != nil { + t.Fatalf("unable to init switch: %v", err) + } + + // Wire the reputation manager into the switch config before starting. + s.cfg.ReputationManager = repMgr + + if err := s.Start(); err != nil { + t.Fatalf("unable to start switch: %v", err) + } + t.Cleanup(func() { _ = s.Stop() }) + + chanID1, chanID2, aliceChanID, bobChanID := genIDs() + + aliceLink := newMockChannelLink( + s, chanID1, aliceChanID, emptyScid, alicePeer, true, false, + false, false, + ) + bobLink := newMockChannelLink( + s, chanID2, bobChanID, emptyScid, bobPeer, true, false, false, + false, + ) + if err := s.AddLink(aliceLink); err != nil { + t.Fatalf("unable to add alice link: %v", err) + } + if err := s.AddLink(bobLink); err != nil { + t.Fatalf("unable to add bob link: %v", err) + } + + return s, aliceLink, bobLink +} + +// TestSwitchReputationForwardSettle asserts that forwarding then settling an +// HTLC fires OnForward and OnSettle exactly once with the correct circuit keys. +func TestSwitchReputationForwardSettle(t *testing.T) { + t.Parallel() + + repMgr := &mockReputationManager{} + s, aliceLink, bobLink := newReputationTestSwitch(t, repMgr) + + preimage, err := genPreimage() + if err != nil { + t.Fatalf("unable to generate preimage: %v", err) + } + rhash := sha256.Sum256(preimage[:]) + + addPkt := &htlcPacket{ + incomingChanID: aliceLink.ShortChanID(), + incomingHTLCID: 0, + outgoingChanID: bobLink.ShortChanID(), + obfuscator: NewMockObfuscator(), + htlc: &lnwire.UpdateAddHTLC{ + PaymentHash: rhash, + Amount: 1, + }, + } + if err := s.ForwardPackets(nil, addPkt); err != nil { + t.Fatal(err) + } + + select { + case <-bobLink.packets: + if err := bobLink.completeCircuit(addPkt); err != nil { + t.Fatalf("unable to complete circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("add was not propagated to destination") + } + + forwards, _, _ := repMgr.snapshot() + if len(forwards) != 1 { + t.Fatalf("expected 1 OnForward, got %d", len(forwards)) + } + if forwards[0].in.ChanID != aliceLink.ShortChanID() || + forwards[0].out != bobLink.ShortChanID() { + + t.Fatalf("OnForward wrong keys: in=%v out=%v", + forwards[0].in, forwards[0].out) + } + + settlePkt := &htlcPacket{ + outgoingChanID: bobLink.ShortChanID(), + outgoingHTLCID: 0, + amount: 1, + htlc: &lnwire.UpdateFulfillHTLC{ + PaymentPreimage: preimage, + }, + } + if err := s.ForwardPackets(nil, settlePkt); err != nil { + t.Fatal(err) + } + + select { + case pkt := <-aliceLink.packets: + if err := aliceLink.deleteCircuit(pkt); err != nil { + t.Fatalf("unable to remove circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("settle was not propagated upstream") + } + + _, settles, fails := repMgr.snapshot() + require.Len(t, settles, 1, "expected exactly 1 OnSettle") + require.Equal(t, aliceLink.ShortChanID(), settles[0].in.ChanID, + "OnSettle wrong incoming key") + require.Empty(t, fails, "expected 0 OnFail") +} + +// TestSwitchReputationForwardFail asserts that forwarding then failing an HTLC +// fires OnForward and OnFail (not OnSettle). +func TestSwitchReputationForwardFail(t *testing.T) { + t.Parallel() + + repMgr := &mockReputationManager{} + s, aliceLink, bobLink := newReputationTestSwitch(t, repMgr) + + preimage, err := genPreimage() + if err != nil { + t.Fatalf("unable to generate preimage: %v", err) + } + rhash := sha256.Sum256(preimage[:]) + + addPkt := &htlcPacket{ + incomingChanID: aliceLink.ShortChanID(), + incomingHTLCID: 0, + outgoingChanID: bobLink.ShortChanID(), + obfuscator: NewMockObfuscator(), + htlc: &lnwire.UpdateAddHTLC{ + PaymentHash: rhash, + Amount: 1, + }, + } + if err := s.ForwardPackets(nil, addPkt); err != nil { + t.Fatal(err) + } + + select { + case <-bobLink.packets: + if err := bobLink.completeCircuit(addPkt); err != nil { + t.Fatalf("unable to complete circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("add was not propagated to destination") + } + + failPkt := &htlcPacket{ + outgoingChanID: bobLink.ShortChanID(), + outgoingHTLCID: 0, + amount: 1, + htlc: &lnwire.UpdateFailHTLC{}, + } + if err := s.ForwardPackets(nil, failPkt); err != nil { + t.Fatal(err) + } + + select { + case pkt := <-aliceLink.packets: + if err := aliceLink.deleteCircuit(pkt); err != nil { + t.Fatalf("unable to remove circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("fail was not propagated upstream") + } + + forwards, settles, fails := repMgr.snapshot() + require.Len(t, forwards, 1, "expected exactly 1 OnForward") + require.Len(t, fails, 1, "expected exactly 1 OnFail") + require.Equal(t, aliceLink.ShortChanID(), fails[0].in.ChanID, + "OnFail wrong incoming key") + require.Empty(t, settles, "expected 0 OnSettle") +} + +// panicReputationManager is a stub whose hooks always panic, used to prove the +// switch's forwarding path survives a misbehaving (buggy) reputation +// subsystem. It also counts calls, so a test can assert that the guard +// still forwards every hook to it. +type panicReputationManager struct { + mu sync.Mutex + calls int +} + +func (p *panicReputationManager) bump() { + p.mu.Lock() + p.calls++ + p.mu.Unlock() +} + +func (p *panicReputationManager) callCount() int { + p.mu.Lock() + defer p.mu.Unlock() + + return p.calls +} + +func (p *panicReputationManager) OnForward(_ CircuitKey, + _ lnwire.ShortChannelID, _, _, _ lnwire.MilliSatoshi, _, _ uint32, + _ bool) { + + p.bump() + panic("boom from OnForward") +} + +func (p *panicReputationManager) OnSettle(_ CircuitKey) { + p.bump() + panic("boom from OnSettle") +} + +func (p *panicReputationManager) OnFail(_ CircuitKey) { + p.bump() + panic("boom from OnFail") +} + +// mustNotPanic fails the test if fn panics (the guard should absorb it). +func mustNotPanic(t *testing.T, fn func()) { + t.Helper() + defer func() { + if r := recover(); r != nil { + t.Fatalf("panic escaped the reputation guard: %v", r) + } + }() + fn() +} + +// TestGuardedReputationManagerRecovers asserts that the panic boundary around +// the reputation hooks (NewGuardedReputationManager) swallows a hook panic, so +// that a bug in the log-only subsystem can never propagate to the caller. This +// is the unit-level proof; TestSwitchReputationPanicSurvives drives it through +// the live switch. +func TestGuardedReputationManagerRecovers(t *testing.T) { + t.Parallel() + + inner := &panicReputationManager{} + guard := NewGuardedReputationManager(inner) + + in := CircuitKey{ChanID: lnwire.NewShortChanIDFromInt(1)} + out := lnwire.NewShortChanIDFromInt(2) + + // Every hook panics internally; the guard must recover each time so the + // caller's goroutine is unaffected. + mustNotPanic(t, func() { + guard.OnForward(in, out, 1, 1, 0, 100, 90, false) + guard.OnSettle(in) + guard.OnFail(in) + }) + if inner.callCount() != 3 { + t.Fatalf("every hook should have reached the inner manager, "+ + "got %d calls", inner.callCount()) + } +} + +// TestSwitchReputationPanicSurvives drives a forward through a live switch +// whose reputation manager panics in OnForward, and asserts the HTLC is still +// forwarded to the destination, i.e. a subsystem panic cannot take down the +// switch's forwarding goroutine. +func TestSwitchReputationPanicSurvives(t *testing.T) { + t.Parallel() + + guard := NewGuardedReputationManager(&panicReputationManager{}) + s, aliceLink, bobLink := newReputationTestSwitch(t, guard) + + preimage, err := genPreimage() + if err != nil { + t.Fatalf("unable to generate preimage: %v", err) + } + rhash := sha256.Sum256(preimage[:]) + + addPkt := &htlcPacket{ + incomingChanID: aliceLink.ShortChanID(), + incomingHTLCID: 0, + outgoingChanID: bobLink.ShortChanID(), + obfuscator: NewMockObfuscator(), + htlc: &lnwire.UpdateAddHTLC{ + PaymentHash: rhash, + Amount: 1, + }, + } + if err := s.ForwardPackets(nil, addPkt); err != nil { + t.Fatal(err) + } + + // Despite OnForward panicking, the HTLC must still reach the + // destination link, so forwarding is unaffected. + select { + case <-bobLink.packets: + if err := bobLink.completeCircuit(addPkt); err != nil { + t.Fatalf("unable to complete circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("add was not propagated despite reputation panic") + } +} + +// TestSwitchReputationLocalSendSkipped asserts that a locally-originated HTLC +// (this node is the payment source) does NOT invoke the reputation hooks: only +// genuine forwards are observed. A false trigger here would pollute reputation +// with the node's own payments. +func TestSwitchReputationLocalSendSkipped(t *testing.T) { + t.Parallel() + + repMgr := &mockReputationManager{} + + peer, err := newMockServer( + t, "alice", testStartingHeight, nil, testDefaultDelta, + ) + if err != nil { + t.Fatalf("unable to create server: %v", err) + } + + s, err := initSwitchWithTempDB(t, testStartingHeight) + if err != nil { + t.Fatalf("unable to init switch: %v", err) + } + s.cfg.ReputationManager = repMgr + if err := s.Start(); err != nil { + t.Fatalf("unable to start switch: %v", err) + } + t.Cleanup(func() { _ = s.Stop() }) + + chanID, _, aliceChanID, _ := genIDs() + link := newMockChannelLink( + s, chanID, aliceChanID, emptyScid, peer, true, false, false, + true, + ) + if err := s.AddLink(link); err != nil { + t.Fatalf("unable to add link: %v", err) + } + + preimage, err := genPreimage() + if err != nil { + t.Fatalf("unable to generate preimage: %v", err) + } + rhash := sha256.Sum256(preimage[:]) + + // SendHTLC originates a payment from this node (incoming chan is + // hop.Source), so it must not be treated as a forward. + htlc := &lnwire.UpdateAddHTLC{PaymentHash: rhash, Amount: 1} + if err := s.SendHTLC(link.ShortChanID(), 0, htlc); err != nil { + t.Fatalf("unable to send local htlc: %v", err) + } + + // Drain the add from the outgoing link so it is actually dispatched. + select { + case <-link.packets: + case <-time.After(time.Second): + t.Fatal("local add was not dispatched") + } + + forwards, settles, fails := repMgr.snapshot() + if len(forwards) != 0 { + t.Fatalf("local send must not trigger OnForward, got %d", + len(forwards)) + } + if len(settles) != 0 || len(fails) != 0 { + t.Fatalf("local send must not trigger resolutions, got "+ + "%d settles %d fails", len(settles), len(fails)) + } +} + +// TestSwitchReputationNilManagerNoop asserts that with no reputation manager +// configured (the default), forwarding works and nothing panics, i.e. the +// hooks are safely skipped. +func TestSwitchReputationNilManagerNoop(t *testing.T) { + t.Parallel() + + s, aliceLink, bobLink := newReputationTestSwitch(t, nil) + + preimage, err := genPreimage() + if err != nil { + t.Fatalf("unable to generate preimage: %v", err) + } + rhash := sha256.Sum256(preimage[:]) + + addPkt := &htlcPacket{ + incomingChanID: aliceLink.ShortChanID(), + incomingHTLCID: 0, + outgoingChanID: bobLink.ShortChanID(), + obfuscator: NewMockObfuscator(), + htlc: &lnwire.UpdateAddHTLC{ + PaymentHash: rhash, + Amount: 1, + }, + } + if err := s.ForwardPackets(nil, addPkt); err != nil { + t.Fatal(err) + } + + select { + case <-bobLink.packets: + if err := bobLink.completeCircuit(addPkt); err != nil { + t.Fatalf("unable to complete circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("add was not propagated with nil reputation manager") + } +} + +// accountableAddPkt builds a forwarding add packet from alice to bob whose +// incoming HTLC carries the experimental accountable bit set. +func accountableAddPkt(t *testing.T, alice, + bob *mockChannelLink) *htlcPacket { + + t.Helper() + + preimage, err := genPreimage() + if err != nil { + t.Fatalf("unable to generate preimage: %v", err) + } + rhash := sha256.Sum256(preimage[:]) + + return &htlcPacket{ + incomingChanID: alice.ShortChanID(), + incomingHTLCID: 0, + outgoingChanID: bob.ShortChanID(), + obfuscator: NewMockObfuscator(), + htlc: &lnwire.UpdateAddHTLC{ + PaymentHash: rhash, + Amount: 1, + CustomRecords: lnwire.CustomRecords{ + uint64(lnwire.ExperimentalAccountableType): { + lnwire.ExperimentalAccountable, + }, + }, + }, + } +} + +// TestSwitchReputationAdvertisedFee asserts that the switch feeds the fee the +// node ADVERTISED for the forward (not the offered in-out delta) to OnForward: +// the outgoing link's outbound fee plus the incoming link's inbound fee, +// clamped at zero when an inbound discount pushes the total negative. +func TestSwitchReputationAdvertisedFee(t *testing.T) { + t.Parallel() + + const outFee = lnwire.MilliSatoshi(4242) + + tests := []struct { + name string + inboundFee models.InboundFee + wantFee lnwire.MilliSatoshi + }{{ + // No inbound fee: only the outgoing link's advertised fee. + name: "outbound only", + wantFee: outFee, + }, { + // A positive inbound fee is added on top. + name: "inbound fee added", + inboundFee: models.InboundFee{Base: 100}, + wantFee: outFee + 100, + }, { + // An inbound discount larger than the outbound fee clamps the + // total at zero rather than going negative. + name: "inbound discount clamped", + inboundFee: models.InboundFee{Base: -5000}, + wantFee: 0, + }} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + repMgr := &mockReputationManager{} + s, aliceLink, bobLink := newReputationTestSwitch( + t, repMgr, + ) + + // The outgoing (bob) link advertises a fee distinct + // from any in-out delta so we can prove the switch + // sources the advertised value. + bobLink.advertisedFee = outFee + + // The incoming link stamps its inbound fee schedule + // onto the packet; mirror that here. + addPkt := accountableAddPkt(t, aliceLink, bobLink) + addPkt.inboundFee = test.inboundFee + + if err := s.ForwardPackets(nil, addPkt); err != nil { + t.Fatal(err) + } + + select { + case <-bobLink.packets: + err := bobLink.completeCircuit(addPkt) + if err != nil { + t.Fatalf("unable to complete "+ + "circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("add was not propagated to " + + "destination") + } + + forwards, _, _ := repMgr.snapshot() + if len(forwards) != 1 { + t.Fatalf("expected 1 OnForward, got %d", + len(forwards)) + } + if forwards[0].advertisedFee != test.wantFee { + t.Fatalf("advertised fee: got %d, want %d", + forwards[0].advertisedFee, + test.wantFee) + } + }) + } +} + +// TestSwitchReputationAccountabilityGating asserts that the outgoing +// accountable bit fed to OnForward is derived the way the outgoing link derives +// it: even when the incoming HTLC is accountable, a node that does not forward +// the experimental accountability signal reports the forward as unaccountable. +func TestSwitchReputationAccountabilityGating(t *testing.T) { + t.Parallel() + + t.Run("forwarded when enabled", func(t *testing.T) { + t.Parallel() + + repMgr := &mockReputationManager{} + s, aliceLink, bobLink := newReputationTestSwitch(t, repMgr) + s.cfg.ShouldFwdExpAccountability = func() bool { return true } + + addPkt := accountableAddPkt(t, aliceLink, bobLink) + if err := s.ForwardPackets(nil, addPkt); err != nil { + t.Fatal(err) + } + + select { + case <-bobLink.packets: + if err := bobLink.completeCircuit(addPkt); err != nil { + t.Fatalf("complete circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("add was not propagated to destination") + } + + forwards, _, _ := repMgr.snapshot() + if len(forwards) != 1 { + t.Fatalf("expected 1 OnForward, got %d", len(forwards)) + } + if !forwards[0].accountable { + t.Fatalf("expected accountable=true when enabled") + } + }) + + t.Run("gated off when disabled", func(t *testing.T) { + t.Parallel() + + repMgr := &mockReputationManager{} + s, aliceLink, bobLink := newReputationTestSwitch(t, repMgr) + s.cfg.ShouldFwdExpAccountability = func() bool { return false } + + addPkt := accountableAddPkt(t, aliceLink, bobLink) + if err := s.ForwardPackets(nil, addPkt); err != nil { + t.Fatal(err) + } + + select { + case <-bobLink.packets: + if err := bobLink.completeCircuit(addPkt); err != nil { + t.Fatalf("complete circuit: %v", err) + } + case <-time.After(time.Second): + t.Fatal("add was not propagated to destination") + } + + forwards, _, _ := repMgr.snapshot() + if len(forwards) != 1 { + t.Fatalf("expected 1 OnForward, got %d", len(forwards)) + } + if forwards[0].accountable { + t.Fatalf("expected accountable=false when gated off") + } + }) +} + +// TestSwitchReputationNonStrictForward asserts that when the switch forwards an +// HTLC over a channel other than the one the sender asked for (non-strict +// forwarding, where the requested link cannot take the HTLC but another link to +// the same peer can), the reputation manager is told about the channel the HTLC +// actually went out on, and the resolution reports that same channel. +// +// This matters because the manager records the pending HTLC against the +// channel reported at forward time: reporting the requested channel here would +// attribute the HTLC's reputation and in-flight risk to a channel it never +// went out on. +func TestSwitchReputationNonStrictForward(t *testing.T) { + t.Parallel() + + repMgr := &mockReputationManager{} + + alicePeer, err := newMockServer( + t, "alice", testStartingHeight, nil, testDefaultDelta, + ) + require.NoError(t, err, "unable to create alice server") + bobPeer, err := newMockServer( + t, "bob", testStartingHeight, nil, testDefaultDelta, + ) + require.NoError(t, err, "unable to create bob server") + + s, err := initSwitchWithTempDB(t, testStartingHeight) + require.NoError(t, err, "unable to init switch") + + s.cfg.ReputationManager = repMgr + require.NoError(t, s.Start()) + t.Cleanup(func() { _ = s.Stop() }) + + chanID1, aliceChanID := genID() + aliceLink := newMockChannelLink( + s, chanID1, aliceChanID, emptyScid, alicePeer, true, false, + false, false, + ) + + // Bob has two channels with us. The first is the one the sender asked + // for, but it cannot take the HTLC, so the switch must fall back to the + // second one. + chanID2, bobChanID1 := genID() + requestedLink := newMockChannelLink( + s, chanID2, bobChanID1, emptyScid, bobPeer, true, false, false, + false, + ) + requestedLink.checkHtlcForwardResult = NewDetailedLinkError( + lnwire.NewTemporaryChannelFailure(nil), + OutgoingFailureInsufficientBalance, + ) + + chanID3, bobChanID2 := genID() + chosenLink := newMockChannelLink( + s, chanID3, bobChanID2, emptyScid, bobPeer, true, false, false, + false, + ) + + require.NoError(t, s.AddLink(aliceLink)) + require.NoError(t, s.AddLink(requestedLink)) + require.NoError(t, s.AddLink(chosenLink)) + + preimage, err := genPreimage() + require.NoError(t, err, "unable to generate preimage") + rhash := sha256.Sum256(preimage[:]) + + // The packet asks for Bob's first channel, which cannot forward it. + addPkt := &htlcPacket{ + incomingChanID: aliceLink.ShortChanID(), + incomingHTLCID: 0, + outgoingChanID: requestedLink.ShortChanID(), + obfuscator: NewMockObfuscator(), + htlc: &lnwire.UpdateAddHTLC{ + PaymentHash: rhash, + Amount: 1, + }, + } + require.NoError(t, s.ForwardPackets(nil, addPkt)) + + select { + case <-chosenLink.packets: + require.NoError(t, chosenLink.completeCircuit(addPkt)) + + case <-requestedLink.packets: + t.Fatal("htlc went out on the link that cannot forward it") + + case <-time.After(time.Second): + t.Fatal("add was not propagated to destination") + } + + // The forward must be reported against the channel actually used, not + // the one the sender requested. + forwards, _, _ := repMgr.snapshot() + require.Len(t, forwards, 1) + require.Equal(t, chosenLink.ShortChanID(), forwards[0].out, + "forward must report the chosen outgoing channel") + + settlePkt := &htlcPacket{ + outgoingChanID: chosenLink.ShortChanID(), + outgoingHTLCID: 0, + amount: 1, + htlc: &lnwire.UpdateFulfillHTLC{ + PaymentPreimage: preimage, + }, + } + require.NoError(t, s.ForwardPackets(nil, settlePkt)) + + select { + case pkt := <-aliceLink.packets: + require.NoError(t, aliceLink.deleteCircuit(pkt)) + + case <-time.After(time.Second): + t.Fatal("settle was not propagated upstream") + } + + // The resolution matches the add by its incoming circuit key. + _, settles, _ := repMgr.snapshot() + require.Len(t, settles, 1) + require.Equal(t, forwards[0].in, settles[0].in, + "resolve must report the same incoming circuit as the forward") +} + +// TestSwitchReputationMailboxFailAdd asserts that an add failed back through +// the outgoing link's mailbox (mailbox.FailAdd) still reaches the reputation +// manager as a fail with the HTLC's incoming circuit key. +// +// This is the path taken when the outgoing link cannot commit the HTLC +// (channel.AddHTLC fails, the link is flushing, fee exposure is hit) or the +// mailbox delivery deadline elapses. No keystone was ever set for the circuit +// and the mailbox builds a fresh fail packet that carries no outgoing scid, so +// this resolution path cannot name the outgoing channel; the manager must be +// able to match the fail to the forward by circuit key alone. +func TestSwitchReputationMailboxFailAdd(t *testing.T) { + t.Parallel() + + repMgr := &mockReputationManager{} + s, aliceLink, bobLink := newReputationTestSwitch(t, repMgr) + + preimage, err := genPreimage() + require.NoError(t, err, "unable to generate preimage") + rhash := sha256.Sum256(preimage[:]) + + addPkt := &htlcPacket{ + incomingChanID: aliceLink.ShortChanID(), + incomingHTLCID: 0, + outgoingChanID: bobLink.ShortChanID(), + obfuscator: NewMockObfuscator(), + htlc: &lnwire.UpdateAddHTLC{ + PaymentHash: rhash, + Amount: 1, + }, + } + require.NoError(t, s.ForwardPackets(nil, addPkt)) + + // Take the packet out of the outgoing link's mailbox, but do NOT call + // completeCircuit: that is what sets the keystone, and the whole point + // of this path is that the add fails before a keystone exists. + var queued *htlcPacket + select { + case queued = <-bobLink.packets: + case <-time.After(time.Second): + t.Fatal("add was not propagated to the destination link") + } + + forwards, _, _ := repMgr.snapshot() + require.Len(t, forwards, 1) + + // The outgoing link cannot add the HTLC to its commitment, so it fails + // the add back through the mailbox. + bobLink.mailBox.FailAdd(queued) + + select { + case <-aliceLink.packets: + case <-time.After(time.Second): + t.Fatal("fail was not propagated upstream") + } + + // The fail must reach the manager with the same incoming circuit key + // as the forward, so the pending HTLC it recorded can be resolved. + require.Eventually(t, func() bool { + _, _, fails := repMgr.snapshot() + + return len(fails) == 1 + }, 2*time.Second, 10*time.Millisecond) + + _, _, fails := repMgr.snapshot() + require.Equal(t, forwards[0].in, fails[0].in, + "mailbox fail must resolve the same incoming circuit as the "+ + "forward") +} diff --git a/htlcswitch/switch.go b/htlcswitch/switch.go index 2c0bbddb624..fd22fc7992f 100644 --- a/htlcswitch/switch.go +++ b/htlcswitch/switch.go @@ -188,6 +188,21 @@ type Config struct { // events through. HtlcNotifier htlcNotifier + // ReputationManager is an optional, read-only local reputation + // subsystem. When non-nil, the switch feeds it forward/settle/fail + // events for forwarded HTLCs so it can track reputation. It is a black + // box that never affects forwarding (log-only); when nil the hooks are + // skipped. + ReputationManager ReputationManager + + // ShouldFwdExpAccountability reports whether this node forwards the + // experimental accountability signal. It mirrors the per-link closure + // of the same name and is used by the reputation hooks to derive the + // outgoing accountable bit the way the outgoing link would (so a peer + // that was never told an HTLC was accountable is not penalised). It may + // be nil, in which case accountability is treated as forwarded. + ShouldFwdExpAccountability func() bool + // FwdEventTicker is a signal that instructs the htlcswitch to flush any // pending forwarding events. FwdEventTicker ticker.Ticker @@ -3022,9 +3037,73 @@ func (s *Switch) handlePacketAdd(packet *htlcPacket, // channel. packet.outgoingChanID = destination.ShortChanID() + // Feed the (read-only) reputation manager this forward. This only + // observes the event to update internal reputation state; it never + // affects the forwarding decision (log-only). + if s.cfg.ReputationManager != nil { + // Use the fee the node ADVERTISED for this forward, not the + // (possibly larger) fee offered by the incoming HTLC. Scoring + // reputation/revenue on the offered fee would let a sender + // inflate or destroy reputation by over/under-paying; the + // advertised fee is what the node actually charges. + // + // The advertised fee is the same total CheckHtlcForward + // requires (outbound plus inbound component), attributed to the + // outgoing link. An inbound discount can push the total + // negative, in which case it is clamped at zero: reputation and + // revenue track fees actually earned, never owed. + totalFee := TotalForwardingFee( + packet.amount, destination.AdvertisedFee(packet.amount), + packet.inboundFee, + ) + if totalFee < 0 { + totalFee = 0 + } + advertisedFee := lnwire.MilliSatoshi(totalFee) + + // Derive the outgoing accountable bit exactly as the outgoing + // link does: only accountable if we received it accountable AND + // this node forwards the experimental accountability signal. A + // node running --protocol.no-experimental-accountability drops + // the bit, so it must not penalise a peer never told the HTLC + // was accountable. + outgoingAccountable := htlcAccountable(htlc) && + s.shouldFwdExpAccountability() + + s.cfg.ReputationManager.OnForward( + CircuitKey{ + ChanID: packet.incomingChanID, + HtlcID: packet.incomingHTLCID, + }, + packet.outgoingChanID, packet.incomingAmount, + packet.amount, advertisedFee, packet.incomingTimeout, + s.BestHeight(), outgoingAccountable, + ) + } + return destination.handleSwitchPacket(packet) } +// htlcAccountable extracts the experimental accountable signal from an +// incoming update_add_htlc's custom records (TLV 106823). +func htlcAccountable(htlc *lnwire.UpdateAddHTLC) bool { + key := uint64(lnwire.ExperimentalAccountableType) + rec, ok := htlc.CustomRecords[key] + + return ok && len(rec) > 0 && rec[0] == lnwire.ExperimentalAccountable +} + +// shouldFwdExpAccountability reports whether this node forwards the +// experimental accountability signal, defaulting to true when the closure is +// unset. +func (s *Switch) shouldFwdExpAccountability() bool { + if s.cfg.ShouldFwdExpAccountability == nil { + return true + } + + return s.cfg.ShouldFwdExpAccountability() +} + // handlePacketSettle handles forwarding a settle packet. func (s *Switch) handlePacketSettle(packet *htlcPacket) error { // If the source of this packet has not been set, use the circuit map @@ -3101,6 +3180,12 @@ func (s *Switch) handlePacketSettle(packet *htlcPacket) error { }, ) s.fwdEventMtx.Unlock() + + // Feed the read-only reputation manager this settle; + // log-only, never affects resolution. + if s.cfg.ReputationManager != nil { + s.cfg.ReputationManager.OnSettle(circuit.Incoming) + } } // Deliver this packet. @@ -3135,6 +3220,15 @@ func (s *Switch) handlePacketFail(packet *htlcPacket, return nil } + // Feed the read-only reputation manager this forwarded fail; + // log-only, never affects resolution. The HTLC is identified by its + // incoming circuit key alone: fail paths do not reliably know the + // outgoing channel (an add failed back through the outgoing link's + // mailbox never had a keystone set). + if s.cfg.ReputationManager != nil && circuit != nil { + s.cfg.ReputationManager.OnFail(circuit.Incoming) + } + // Exit early if this hasSource is true. This flag is only set via // mailbox's `FailAdd`. This method has two callsites, // - the packet has timed out after `MailboxDeliveryTimeout`, defaults diff --git a/lncfg/routing.go b/lncfg/routing.go index 89675788706..61107ce839c 100644 --- a/lncfg/routing.go +++ b/lncfg/routing.go @@ -10,6 +10,8 @@ type Routing struct { StrictZombiePruning bool `long:"strictgraphpruning" description:"If true, then the graph will be pruned more aggressively for zombies. In practice this means that edges with a single stale edge will be considered a zombie."` + NoReputation bool `long:"no-reputation" description:"EXPERIMENTAL: disable the read-only local reputation subsystem (channel jamming mitigation), which is enabled by default. The subsystem only observes HTLC forwarding to compute and log reputation; it does NOT currently affect routing in any way."` + BlindedPaths BlindedPaths `group:"blinding" namespace:"blinding"` } diff --git a/log.go b/log.go index 563ee3eb11d..ee25266b22a 100644 --- a/log.go +++ b/log.go @@ -52,6 +52,7 @@ import ( "github.com/lightningnetwork/lnd/peer" "github.com/lightningnetwork/lnd/peernotifier" "github.com/lightningnetwork/lnd/protofsm" + "github.com/lightningnetwork/lnd/reputation" "github.com/lightningnetwork/lnd/routing" "github.com/lightningnetwork/lnd/routing/blindedpath" "github.com/lightningnetwork/lnd/routing/localchans" @@ -216,6 +217,9 @@ func SetupLoggers(root *build.SubLoggerManager, interceptor signal.Interceptor) ) AddSubLogger(root, onionmessage.Subsystem, interceptor, onionmessage.UseLogger) + AddSubLogger( + root, reputation.Subsystem, interceptor, reputation.UseLogger, + ) } // AddSubLogger is a helper method to conveniently create and register the diff --git a/sample-lnd.conf b/sample-lnd.conf index f881c1174e9..5282d5416a5 100644 --- a/sample-lnd.conf +++ b/sample-lnd.conf @@ -1964,6 +1964,11 @@ [routing] +; EXPERIMENTAL: disable the read-only local reputation subsystem (channel jamming +; mitigation), which is enabled by default. It only observes HTLC forwarding to +; compute and log reputation; it does NOT currently affect routing in any way. +; routing.no-reputation=false + ; DEPRECATED: This is now turned on by default for Neutrino (use ; neutrino.validatechannels=true to turn off) and shouldn't be used for any ; other backend! diff --git a/server.go b/server.go index a0312bb0146..4d3031430fa 100644 --- a/server.go +++ b/server.go @@ -75,6 +75,7 @@ import ( "github.com/lightningnetwork/lnd/peernotifier" "github.com/lightningnetwork/lnd/pool" "github.com/lightningnetwork/lnd/queue" + "github.com/lightningnetwork/lnd/reputation" "github.com/lightningnetwork/lnd/routing" "github.com/lightningnetwork/lnd/routing/localchans" "github.com/lightningnetwork/lnd/routing/route" @@ -357,6 +358,11 @@ type server struct { htlcNotifier *htlcswitch.HtlcNotifier + // reputationMgr is the read-only local reputation subsystem. It is + // enabled by default and nil only when the experimental + // routing.no-reputation flag is set. + reputationMgr *reputation.Manager + witnessBeacon contractcourt.WitnessBeacon breachArbitrator *contractcourt.BreachArbitrator @@ -880,6 +886,30 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, return nil, err } + // Construct the read-only local reputation subsystem, which is enabled + // by default. It only observes HTLC forwarding to compute and log + // reputation; it never affects routing (log-only). When disabled via + // no-reputation, repMgrIface stays a nil interface so the switch skips + // the hooks entirely. Nothing is persisted, so reputation is re-accrued + // from live traffic on restart. + var repMgrIface htlcswitch.ReputationManager + if !cfg.Routing.NoReputation { + s.reputationMgr, err = reputation.NewManager( + reputation.DefaultConfig(), clock.NewDefaultClock(), + ) + if err != nil { + return nil, err + } + + // Wrap the manager in a panic boundary before handing it to + // the switch: the hooks run on the forwarding goroutine, so a + // bug in the (log-only) subsystem must never take down HTLC + // forwarding. + repMgrIface = htlcswitch.NewGuardedReputationManager( + s.reputationMgr, + ) + } + s.htlcSwitch, err = htlcswitch.New(htlcswitch.Config{ DB: dbs.ChanStateDB, FetchAllOpenChannels: s.chanStateDB.FetchAllOpenChannels, @@ -905,6 +935,10 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, FetchLastChannelUpdate: s.fetchLastChanUpdate(), Notifier: s.cc.ChainNotifier, HtlcNotifier: s.htlcNotifier, + ReputationManager: repMgrIface, + ShouldFwdExpAccountability: func() bool { + return !s.cfg.ProtocolOptions.NoExpAccountability() + }, FwdEventTicker: ticker.New(htlcswitch.DefaultFwdEventInterval), LogEventTicker: ticker.New(htlcswitch.DefaultLogInterval), AckEventTicker: ticker.New(htlcswitch.DefaultAckInterval), @@ -2355,6 +2389,14 @@ func (s *server) Start(ctx context.Context) error { return } + if s.reputationMgr != nil { + cleanup = cleanup.add(s.reputationMgr.Stop) + if err := s.reputationMgr.Start(); err != nil { + startErr = err + return + } + } + if s.towerClientMgr != nil { cleanup = cleanup.add(s.towerClientMgr.Stop) if err := s.towerClientMgr.Start(); err != nil { @@ -2841,6 +2883,12 @@ func (s *server) Stop() error { if err := s.htlcNotifier.Stop(); err != nil { srvrLog.Warnf("failed to stop htlcNotifier: %v", err) } + if s.reputationMgr != nil { + if err := s.reputationMgr.Stop(); err != nil { + srvrLog.Warnf("failed to stop reputationMgr: "+ + "%v", err) + } + } // Update channel.backup file. Make sure to do it before // stopping chanSubSwapper. From 1fe2ffa24ce0333f857566361ed9141964993968 Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Tue, 18 Aug 2026 13:40:34 +0000 Subject: [PATCH 5/6] itest: test the local reputation subsystem end-to-end Add an integration test asserting that a forwarding node running the log-only reputation subsystem forwards, fails and restarts exactly as it would without it, while emitting the expected reputation log lines. --- itest/list_on_test.go | 4 ++ itest/lnd_reputation_test.go | 121 +++++++++++++++++++++++++++++++++++ lntest/harness_assertion.go | 63 ++++++++++++++++++ 3 files changed, 188 insertions(+) create mode 100644 itest/lnd_reputation_test.go diff --git a/itest/list_on_test.go b/itest/list_on_test.go index f1e2eb48eca..ebe081fe370 100644 --- a/itest/list_on_test.go +++ b/itest/list_on_test.go @@ -10,6 +10,10 @@ import ( ) var allTestCases = []*lntest.TestCase{ + { + Name: "local reputation log only", + TestFunc: testLocalReputationLogOnly, + }, { Name: "update channel status", TestFunc: testUpdateChanStatus, diff --git a/itest/lnd_reputation_test.go b/itest/lnd_reputation_test.go new file mode 100644 index 00000000000..39ad5adb0e7 --- /dev/null +++ b/itest/lnd_reputation_test.go @@ -0,0 +1,121 @@ +package itest + +import ( + "github.com/btcsuite/btcd/btcutil/v2" + "github.com/lightningnetwork/lnd/lnrpc" + "github.com/lightningnetwork/lnd/lnrpc/routerrpc" + "github.com/lightningnetwork/lnd/lntest" +) + +// reputationChangeLog is the greppable prefix the reputation subsystem logs +// once per resolved HTLC. +const reputationChangeLog = "Reputation change: outgoing=" + +// testLocalReputationLogOnly verifies that enabling the experimental, +// read-only local reputation subsystem on a forwarding node does not affect +// routing. It exercises the log-only invariant across the paths the switch +// hooks observe (a successful forward, a failed forward, and a restart), +// asserting forwarding behaviour is unchanged in every case. +// +// Beyond non-interference, it also confirms the subsystem actually computes +// reputation by matching the greppable log lines it emits: on the forward Bob +// logs the per-HTLC reputation decision, and on resolution he logs the +// resulting reputation change. After a restart, which resets the in-memory +// state, a further forward must produce another change, proving the subsystem +// rebuilt its state from live traffic. +func testLocalReputationLogOnly(ht *lntest.HarnessTest) { + const chanAmt = btcutil.Amount(100_000) + const paymentAmt = 1000 + + // Alice -> Bob -> Carol. The read-only reputation subsystem is enabled + // by default, so Bob (the forwarding node) runs it without any extra + // flag. + alice := ht.NewNodeWithCoins("Alice", nil) + bob := ht.NewNodeWithCoins("Bob", nil) + carol := ht.NewNode("Carol", nil) + + ht.ConnectNodes(alice, bob) + ht.ConnectNodes(bob, carol) + + // Open Alice -> Bob and Bob -> Carol. + chanPointAB := ht.OpenChannel( + alice, bob, lntest.OpenChannelParams{Amt: chanAmt}, + ) + chanPointBC := ht.OpenChannel( + bob, carol, lntest.OpenChannelParams{Amt: chanAmt}, + ) + + // Make sure Alice has learned of the Bob -> Carol channel so she can + // route the multi-hop payment. + ht.AssertChannelInGraph(alice, chanPointBC) + + // 1. Successful forward. Carol invoices, Alice pays via Bob. With Bob's + // reputation subsystem in log-only mode this must succeed exactly as it + // would without it (Bob observes OnForward + OnSettle). + payReqs, _, _ := ht.CreatePayReqs(carol, paymentAmt, 1) + ht.CompletePaymentRequests(alice, payReqs) + + // On the forward, Bob logs the per-HTLC reputation decision ("if this + // HTLC were forwarded in isolation, would its outgoing channel have + // sufficient reputation to be protected?"). Its presence confirms the + // OnForward hook fired and the decision was computed (log-only). + ht.AssertNodeLogContains(bob, "reputation decision: chan=") + + // On resolution Bob logs the reputation change for the outgoing + // (Bob -> Carol) channel, confirming the subsystem observed both the + // OnForward and the OnSettle hook and computed an update. + ht.AssertNodeLogContains(bob, reputationChangeLog) + + // 2. Failed forward. A payment to Carol with an unknown payment hash is + // routed Alice -> Bob -> Carol and rejected at Carol, so Bob observes + // the forward and its downstream failure (OnFail). Bob must remain + // unaffected and the payment must fail cleanly. + failReq := &routerrpc.SendPaymentRequest{ + Dest: carol.PubKey[:], + Amt: paymentAmt, + PaymentHash: ht.Random32Bytes(), + FinalCltvDelta: finalCltvDelta, + FeeLimitMsat: noFeeLimitMsat, + } + ht.SendPaymentAssertFail( + alice, failReq, + lnrpc.PaymentFailureReason_FAILURE_REASON_INCORRECT_PAYMENT_DETAILS, //nolint:ll + ) + + // 3. Restart. This slice has no persistence, so restarting Bob resets + // the in-memory reputation state; it re-accrues from live traffic (the + // documented self-bootstrapping behaviour). Bob must come back and keep + // forwarding. + // + // Record how many reputation changes have been logged so far, so that + // the assertion after the restart can require a new one rather than + // re-matching a line from an earlier step. + changesBeforeRestart := ht.CountNodeLogOccurrences( + bob, reputationChangeLog, + ) + + ht.RestartNode(bob) + ht.EnsureConnected(alice, bob) + ht.EnsureConnected(bob, carol) + ht.AssertNodeNumChannels(bob, 2) + ht.AssertChannelActive(bob, chanPointAB) + ht.AssertChannelActive(bob, chanPointBC) + + // A subsequent payment must still forward successfully after the + // restart, confirming the subsystem does not interfere with forwarding + // once it has restarted with empty state. + payReqs2, _, _ := ht.CreatePayReqs(carol, paymentAmt, 1) + ht.CompletePaymentRequests(alice, payReqs2) + + // And reputation re-accrues from live traffic. Earlier steps already + // logged reputation changes of their own, so asserting the line is + // merely present would prove nothing here: we require the count to have + // grown, which can only come from this post-restart forward being + // observed and scored by the reset subsystem. + ht.AssertNodeLogCountAtLeast( + bob, reputationChangeLog, changesBeforeRestart+1, + ) + + ht.CloseChannel(alice, chanPointAB) + ht.CloseChannel(bob, chanPointBC) +} diff --git a/lntest/harness_assertion.go b/lntest/harness_assertion.go index 03c1819ff74..184ced2c8e9 100644 --- a/lntest/harness_assertion.go +++ b/lntest/harness_assertion.go @@ -8,6 +8,8 @@ import ( "encoding/json" "fmt" "math" + "os" + "path/filepath" "sort" "strings" "time" @@ -61,6 +63,67 @@ func (h *HarnessTest) WaitForBlockchainSync(hn *node.HarnessNode) { require.NoError(h, err, "timeout waiting for blockchain sync") } +// AssertNodeLogContains waits until the node's lnd.log contains the given +// substring, failing the test if it does not appear within DefaultTimeout. +// Logs flush asynchronously, so the file is polled. This is used to assert on +// log-only subsystem behaviour (e.g. the read-only reputation subsystem) that +// is not otherwise exposed over RPC. +func (h *HarnessTest) AssertNodeLogContains(hn *node.HarnessNode, + substr string) { + + h.AssertNodeLogCountAtLeast(hn, substr, 1) +} + +// AssertNodeLogCountAtLeast waits until the node's lnd.log contains at least +// count occurrences of the given substring, failing the test if that many do +// not appear within DefaultTimeout. This distinguishes a line logged again from +// one that was already present earlier in the test. +func (h *HarnessTest) AssertNodeLogCountAtLeast(hn *node.HarnessNode, + substr string, count int) { + + err := wait.NoError(func() error { + got := h.CountNodeLogOccurrences(hn, substr) + if got >= count { + return nil + } + + return fmt.Errorf("%s log contains %d occurrences of %q, "+ + "want at least %d", hn.Name(), got, substr, count) + }, DefaultTimeout) + + require.NoError(h, err, "timeout waiting for %d occurrences of log "+ + "substring %q", count, substr) +} + +// CountNodeLogOccurrences returns the total number of occurrences of substr +// across the node's lnd.log files. It is useful for asserting that a line was +// logged *again*, by comparing counts around an action. +func (h *HarnessTest) CountNodeLogOccurrences(hn *node.HarnessNode, + substr string) int { + + var total int + + _ = filepath.WalkDir(hn.Cfg.LogDir, func(path string, d os.DirEntry, + err error) error { + + if err != nil || d.IsDir() { + return nil + } + if filepath.Base(path) != "lnd.log" { + return nil + } + + data, readErr := os.ReadFile(path) + if readErr == nil { + total += strings.Count(string(data), substr) + } + + return nil + }) + + return total +} + // WaitForBlockchainSyncTo waits until the node is synced to bestBlock. func (h *HarnessTest) WaitForBlockchainSyncTo(hn *node.HarnessNode, bestBlock chainhash.Hash) { From a79ffec9ee8a399731852c454cf414a5bf18d1b0 Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Tue, 18 Aug 2026 13:40:34 +0000 Subject: [PATCH 6/6] docs: add release note for the local reputation subsystem --- docs/release-notes/release-notes-0.22.0.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/docs/release-notes/release-notes-0.22.0.md b/docs/release-notes/release-notes-0.22.0.md index 85626b68f41..9e528313723 100644 --- a/docs/release-notes/release-notes-0.22.0.md +++ b/docs/release-notes/release-notes-0.22.0.md @@ -59,6 +59,15 @@ ## Functional Enhancements +* A new experimental [local reputation + subsystem](https://github.com/lightningnetwork/lnd/pull/10919) tracks the + historical forwarding behaviour of peers, following the scoring recommended in + BOLT [#1280](https://github.com/lightning/bolts/pull/1280). It is enabled by + default but is purely observational: it watches forwarded HTLCs to compute and + log a per-HTLC reputation decision (whether the HTLC could stand on the + outgoing channel's reputation if forwarded in isolation) and does not currently + affect routing in any way. It can be disabled with `routing.no-reputation`. + ## RPC Additions * The `routerrpc.EstimateRouteFee` RPC now supports [restricting fee estimates