From 6d4cedd519c8085856346375f795d98136945557 Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Mon, 27 Jul 2026 16:11:07 +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 | 69 +++++++++++++ reputation/decaying_average.go | 74 +++++++++++++ reputation/decaying_average_test.go | 154 ++++++++++++++++++++++++++++ reputation/revenue.go | 85 +++++++++++++++ reputation/saturated.go | 69 +++++++++++++ 5 files changed, 451 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 diff --git a/reputation/README.md b/reputation/README.md new file mode 100644 index 0000000000..0d1998fe98 --- /dev/null +++ b/reputation/README.md @@ -0,0 +1,69 @@ +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 + +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. + +## 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 0000000000..f569ff234a --- /dev/null +++ b/reputation/decaying_average.go @@ -0,0 +1,74 @@ +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. It backs both outgoing-channel reputation and (via +// aggregatedWindowAverage) the incoming-revenue threshold. The running value +// saturates rather than wrapping (see saturatedI64). +type decayingAverage struct { + value saturatedI64 + lastUpdated uint64 // unix seconds + windowSecs float64 + decayRate float64 +} + +// newDecayingAverage creates a decaying average that starts at zero as of the +// provided start timestamp, decaying over the given window. +func newDecayingAverage(start uint64, window time.Duration) *decayingAverage { + windowSecs := window.Seconds() + + return &decayingAverage{ + value: 0, + lastUpdated: start, + windowSecs: windowSecs, + decayRate: decayRateForWindow(windowSecs), + } +} + +// decayRateForWindow computes the per-second decay rate for a window expressed +// in seconds. 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 +// over a full window. +func decayRateForWindow(windowSecs float64) float64 { + return math.Pow(0.5, 1.0/(math.Ln2*windowSecs)) +} + +// valueAt decays the stored value forward to the given timestamp, updates the +// internal state, and returns the decayed value. It errors if the timestamp is +// before the last update. +func (d *decayingAverage) valueAt(ts uint64) (int64, error) { + if ts < d.lastUpdated { + return 0, errBackwardsTime + } + + elapsed := float64(ts - d.lastUpdated) + d.value = satFromFloat( + math.Round(float64(d.value) * math.Pow(d.decayRate, elapsed)), + ) + d.lastUpdated = ts + + return d.value.Int64(), nil +} + +// add decays the value to the given timestamp and then adds the provided +// (possibly negative) value. +func (d *decayingAverage) add(value int64, ts uint64) (int64, error) { + if _, err := d.valueAt(ts); err != nil { + return 0, err + } + + d.value = d.value.Add(saturatedI64(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 0000000000..1ded944265 --- /dev/null +++ b/reputation/decaying_average_test.go @@ -0,0 +1,154 @@ +package reputation + +import ( + "errors" + "math" + "testing" + "time" +) + +// TestDecayingAverageDecay verifies the decay e^(-elapsed/window): the value +// decays to 1000/e^0.5 at half a window and to 1000/e at a full window. +func TestDecayingAverageDecay(t *testing.T) { + t.Parallel() + + const window = 100 * time.Second + d := newDecayingAverage(0, window) + + if _, err := d.add(1000, 0); err != nil { + t.Fatalf("add: %v", err) + } + + // At half a window (50s): 1000 * e^(-0.5) = 606.5 -> 607. + got, err := d.valueAt(50) + if err != nil { + t.Fatalf("valueAt: %v", err) + } + if got != 607 { + t.Fatalf("half window: got %d, want 607", got) + } + + // At a full window (another 50s): decays by e^(-0.5) again -> 368. + got, err = d.valueAt(100) + if err != nil { + t.Fatalf("valueAt: %v", err) + } + if got != 368 { + t.Fatalf("full window: got %d, want 368", got) + } +} + +// TestDecayingAverageAddAndDecay checks add-then-decay sequencing. +func TestDecayingAverageAddAndDecay(t *testing.T) { + t.Parallel() + + const window = 100 * time.Second + d := newDecayingAverage(0, window) + + if _, err := d.add(1000, 0); err != nil { + t.Fatalf("add: %v", err) + } + // Decay to 50s -> 607, then add 1000 -> 1607. + v, err := d.add(1000, 50) + if err != nil { + t.Fatalf("add: %v", err) + } + if v != 1607 { + t.Fatalf("got %d, want 1607", v) + } +} + +// TestDecayingAverageBackwardsTime ensures a backwards timestamp errors, since +// the decay assumes monotonic time. +func TestDecayingAverageBackwardsTime(t *testing.T) { + t.Parallel() + + d := newDecayingAverage(100, time.Hour) + if _, err := d.valueAt(50); !errors.Is(err, errBackwardsTime) { + t.Fatalf("expected errBackwardsTime, got %v", err) + } +} + +// TestSaturatedI64 checks the saturating arithmetic clamps rather than wraps. +func TestSaturatedI64(t *testing.T) { + t.Parallel() + + if got := saturatedI64(math.MaxInt64).Add(1); got != math.MaxInt64 { + t.Fatalf("overflow: got %d", got) + } + if got := saturatedI64(math.MinInt64).Add(-1); got != math.MinInt64 { + t.Fatalf("underflow: got %d", got) + } + if got := saturatedI64(5).Add(-3); got != 2 { + t.Fatalf("normal add: got %d", got) + } + if got := saturatedI64(5).Sub(3); got != 2 { + t.Fatalf("normal sub: got %d", got) + } + if got := satFromUint(math.MaxUint64); got != math.MaxInt64 { + t.Fatalf("satFromUint overflow: got %d", got) + } +} + +// 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(0, window) + + // Saturate the running value to MaxInt64. + if _, err := d.add(math.MaxInt64, 0); err != nil { + t.Fatalf("add: %v", err) + } + if d.value != math.MaxInt64 { + t.Fatalf("setup: value not saturated: %d", d.value) + } + + // Evaluating at the same timestamp (no decay) round-trips the value + // through float64; without the clamp this overflows to MinInt64. + got, err := d.valueAt(0) + if err != nil { + t.Fatalf("valueAt: %v", err) + } + if got != math.MaxInt64 { + t.Fatalf("overflow clamp: got %d, want MaxInt64 (negative "+ + "means the float->int64 cast overflowed)", got) + } +} + +// 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, 0) + + // 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). + if _, err := a.add(600, 0); err != nil { + t.Fatalf("add: %v", err) + } + got, err := a.valueAt(0) + if err != nil { + t.Fatalf("valueAt: %v", err) + } + if got != 600 { + t.Fatalf("warmup t=0: got %d, want 600", got) + } + + // 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(300) + if err != nil { + t.Fatalf("valueAt: %v", err) + } + if got != 154 { + t.Fatalf("warmup t=300: got %d, want 154", got) + } +} diff --git a/reputation/revenue.go b/reputation/revenue.go new file mode 100644 index 0000000000..544d79a025 --- /dev/null +++ b/reputation/revenue.go @@ -0,0 +1,85 @@ +package reputation + +import ( + "math" + "time" +) + +// aggregatedWindowAverage tracks an average value over multiple rolling +// windows to smooth out volatility, used for the incoming-revenue threshold. +// Aggregating over several windows (rather than reading a single window) makes +// the threshold harder for a peer to move quickly by manipulating its recent +// forwarding. +// +// 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 uint64 // unix seconds + 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 uint64) *aggregatedWindowAverage { + + return &aggregatedWindowAverage{ + start: start, + windowCount: windowCount, + windowDuration: window, + inner: newDecayingAverage( + start, window*time.Duration(windowCount), + ), + } +} + +// add records a value at the given timestamp. +func (a *aggregatedWindowAverage) add(value int64, ts uint64) (int64, error) { + return a.inner.add(value, ts) +} + +// windowsTracked returns the (fractional) number of windows (periods) elapsed +// since start. +func (a *aggregatedWindowAverage) windowsTracked(ts uint64) float64 { + elapsed := float64(ts - a.start) + + return elapsed / a.windowDuration.Seconds() +} + +// 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 timestamp. +func (a *aggregatedWindowAverage) valueAt(ts uint64) (int64, error) { + if ts < a.start { + return 0, errBackwardsTime + } + + warmup := a.warmupFactor(a.windowsTracked(ts)) + + 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 0000000000..cdc1f89e95 --- /dev/null +++ b/reputation/saturated.go @@ -0,0 +1,69 @@ +package reputation + +import "math" + +// saturatedI64 is an int64 that clamps to the int64 range on overflow instead +// of wrapping. Reputation and revenue accumulate fees over long windows, so the +// arithmetic must never silently flip sign when a value grows past the int64 +// bound; routing every operation through this type makes that guarantee +// explicit at the call sites. +type saturatedI64 int64 + +// satFromUint converts an unsigned value to a saturatedI64, clamping to the +// maximum when it does not fit. +func satFromUint(v uint64) saturatedI64 { + if v > math.MaxInt64 { + return math.MaxInt64 + } + + return saturatedI64(v) +} + +// satFromFloat converts a float to a saturatedI64, 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 math.MaxInt64 + + case f <= float64(math.MinInt64): + return math.MinInt64 + + default: + return saturatedI64(f) + } +} + +// Add returns the saturating sum of the two values. +func (s saturatedI64) Add(o saturatedI64) saturatedI64 { + sum := s + o + switch { + case s > 0 && o > 0 && sum < 0: + return math.MaxInt64 + + case s < 0 && o < 0 && sum >= 0: + return math.MinInt64 + + default: + return sum + } +} + +// Sub returns the saturating difference of the two values. +func (s saturatedI64) Sub(o saturatedI64) saturatedI64 { + // Negating MinInt64 overflows, so fall back to float clamping for that + // single edge (unreachable in practice: all subtrahends here are + // non-negative opportunity costs). + if o == math.MinInt64 { + return satFromFloat(float64(s) - float64(o)) + } + + return s.Add(-o) +} + +// Int64 returns the underlying int64. +func (s saturatedI64) Int64() int64 { + return int64(s) +} From a22c0fe2ec9fd1ec61e09ee31deadc026ca5a951 Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Mon, 27 Jul 2026 16:11:07 +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 - htlc_risk >= revenue_threshold. --- reputation/channel.go | 54 +++++++++++++++++++ reputation/config.go | 97 +++++++++++++++++++++++++++++++++ reputation/config_test.go | 70 ++++++++++++++++++++++++ reputation/decision.go | 37 +++++++++++++ reputation/htlc.go | 110 ++++++++++++++++++++++++++++++++++++++ reputation/htlc_test.go | 98 +++++++++++++++++++++++++++++++++ 6 files changed, 466 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 0000000000..92b7ac5b81 --- /dev/null +++ b/reputation/channel.go @@ -0,0 +1,54 @@ +package reputation + +// 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 uint64) *channelReputation { + return &channelReputation{ + outgoingReputation: newDecayingAverage( + start, cfg.reputationWindow(), + ), + incomingRevenue: newAggregatedWindowAverage( + cfg.RevenueWindow, cfg.RevenueWindowCount, start, + ), + pendingHTLCs: make(map[htlcRef]*pendingHTLC), + } +} + +// sufficientReputation evaluates the reputation inequality +// +// outgoing_reputation - htlc_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(htlcRisk uint64, + outgoingReputation int64, at uint64) (bool, int64, error) { + + threshold, err := c.incomingRevenue.valueAt(at) + if err != nil { + return false, 0, err + } + + net := saturatedI64(outgoingReputation).Sub(satFromUint(htlcRisk)) + + return net.Int64() >= threshold, threshold, nil +} diff --git a/reputation/config.go b/reputation/config.go new file mode 100644 index 0000000000..74500b7f02 --- /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 seconds per block (10 minutes) used to + // convert cltv deltas to durations. + blockInterval = 10 * 60 +) + +// 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 0000000000..795918a3c4 --- /dev/null +++ b/reputation/config_test.go @@ -0,0 +1,70 @@ +package reputation + +import ( + "testing" + "time" +) + +// 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) { + err := tc.cfg.Validate() + if tc.wantErr && err == nil { + t.Fatalf("expected error, got nil") + } + if !tc.wantErr && err != nil { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} diff --git a/reputation/decision.go b/reputation/decision.go new file mode 100644 index 0000000000..d58673a806 --- /dev/null +++ b/reputation/decision.go @@ -0,0 +1,37 @@ +package reputation + +import "fmt" + +// isolationDecision is the result of the per-HTLC isolation verdict: if this +// HTLC were forwarded in isolation, would its outgoing channel have sufficient +// reputation to be protected? +// +// sufficient = outgoingReputation - htlcRisk >= revenueThreshold +// +// It is log-only: the verdict never affects forwarding. In a later enforcement +// step an insufficient verdict is what would deny an HTLC the protected +// resources. +type isolationDecision struct { + // sufficient reports whether the outgoing channel's reputation would be + // sufficient for this HTLC to be protected in isolation. + sufficient bool + + // outgoingReputation is the outgoing channel's reputation at decision + // time. + outgoingReputation int64 + + // htlcRisk is the in-flight risk of this HTLC. + htlcRisk uint64 + + // threshold is the incoming channel's revenue threshold the reputation + // was compared against. + threshold int64 +} + +// String returns a human readable description of the isolation decision for +// logging. +func (d isolationDecision) String() string { + return fmt.Sprintf("sufficient=%v (outgoing_reputation=%d - "+ + "htlc_risk=%d vs threshold=%d)", d.sufficient, + d.outgoingReputation, d.htlcRisk, d.threshold) +} diff --git a/reputation/htlc.go b/reputation/htlc.go new file mode 100644 index 0000000000..e32e8892d8 --- /dev/null +++ b/reputation/htlc.go @@ -0,0 +1,110 @@ +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). Everything derivable from the incoming cltv (in-flight +// risk, worst-case hold) is computed at forward time and not retained. +type pendingHTLC struct { + // fee is the forwarding fee advertised for this HTLC in millisatoshis. + fee uint64 + + // accountable is the accountable signal observed for this HTLC. + accountable bool + + // addedAt is the unix-seconds timestamp at which the HTLC was + // forwarded. + addedAt uint64 + + // maxHoldSeconds is the worst-case time the HTLC can be held, used by + // the stale-pending garbage collector. + maxHoldSeconds 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() + } + + // oc is a non-negative opportunity cost that fits in an int64, + // so negating it cannot overflow. + return -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 +} + +// maxHoldSeconds returns the worst-case number of seconds an HTLC may be held, +// derived from how far its incoming cltv expiry is from the height it was added +// at (assuming 10-minute blocks). A non-positive delta yields zero; callers +// validate that the incoming expiry is in the future before adding an HTLC. +func maxHoldSeconds(incomingCltv, heightAdded uint32) uint64 { + var delta uint32 + if incomingCltv > heightAdded { + delta = incomingCltv - heightAdded + } + + return uint64(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 { + + hold := maxHoldSeconds(incomingCltv, heightAdded) + + return c.opportunityCost(time.Duration(hold)*time.Second, feeMsat) +} diff --git a/reputation/htlc_test.go b/reputation/htlc_test.go new file mode 100644 index 0000000000..90dcf33c8f --- /dev/null +++ b/reputation/htlc_test.go @@ -0,0 +1,98 @@ +package reputation + +import ( + "testing" + "time" +) + +// 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) + if got != tc.want { + t.Fatalf("opportunityCost(%v): got %d, want %d", + tc.resolution, got, tc.want) + } + } +} + +// 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) { + got := cfg.effectiveFee( + fee, tc.resolution, tc.accountable, tc.settled, + ) + if got != tc.want { + t.Fatalf("got %d, want %d", got, tc.want) + } + }) + } +} + +// 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. + got := cfg.inFlightRisk(100, 101, 100) + if got != 567 { + t.Fatalf("inFlightRisk: got %d, want 567", got) + } + + // No delta -> zero hold -> zero risk. + if got := cfg.inFlightRisk(100, 100, 100); got != 0 { + t.Fatalf("inFlightRisk zero-delta: got %d, want 0", got) + } +} From a024bf84e7f250a03652c3299370073d7048d81a Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Mon, 27 Jul 2026 16:11:07 +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, 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. A periodic garbage collector warns about and evicts pending HTLCs whose resolution was never observed. Includes unit tests and benchmarks for the per-forward hook cost. --- reputation/bench_test.go | 73 +++++++ reputation/log.go | 30 +++ reputation/manager.go | 384 ++++++++++++++++++++++++++++++++++++ reputation/manager_test.go | 284 ++++++++++++++++++++++++++ reputation/testutil_test.go | 22 +++ 5 files changed, 793 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 0000000000..ab05e9b9f7 --- /dev/null +++ b/reputation/bench_test.go @@ -0,0 +1,73 @@ +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 := circuit(2, 0) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + m.OnForward(in, out, 2000, 1000, 1000, 200, 100, false) + m.OnSettle(in, out) + } +} + +// 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 := circuit(2, 0) + + 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, out) + 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(0, DefaultConfig().reputationWindow()) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := d.add(1000, uint64(i)); err != nil { + b.Fatalf("add: %v", err) + } + } +} diff --git a/reputation/log.go b/reputation/log.go new file mode 100644 index 0000000000..48ef1bdac7 --- /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 0000000000..60098ed9de --- /dev/null +++ b/reputation/manager.go @@ -0,0 +1,384 @@ +package reputation + +import ( + "fmt" + "sync" + "time" + + "github.com/lightningnetwork/lnd/clock" + "github.com/lightningnetwork/lnd/graph/db/models" + "github.com/lightningnetwork/lnd/lnwire" +) + +// gcInterval is how often the stale-pending garbage collector runs. It is a +// safety net for pending HTLCs whose resolution was never observed; such an +// eviction is logged as a warning because it indicates the manager's view of +// in-flight HTLCs has diverged from the switch's. +const gcInterval = 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 + + 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), + quit: make(chan struct{}), + }, nil +} + +// Start launches the periodic stale-pending garbage collector. 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.gcLoop() + }) + + 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 +} + +// now returns the current time in unix seconds. +func (m *Manager) now() uint64 { + return unixSeconds(m.clock.Now()) +} + +// OnForward observes a forwarded HTLC at the point the switch commits to +// forwarding it. advertisedFee is the fee the node advertised on the outgoing +// link for this forward, height is the current best block height, and +// accountable is the outgoing accountable signal. +func (m *Manager) OnForward(incoming, outgoing models.CircuitKey, + incomingAmt, outgoingAmt, advertisedFee lnwire.MilliSatoshi, + incomingCltv, height uint32, accountable bool) { + + at := m.now() + + m.mu.Lock() + defer m.mu.Unlock() + + d, err := m.addHTLC( + incoming, outgoing, incomingAmt, outgoingAmt, 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: if this HTLC were forwarded in + // isolation, would its outgoing channel have sufficient reputation to + // be protected? This is log-only and never affects forwarding. + log.Infof("reputation decision: chan=%v htlc=%v sufficient=%v (would "+ + "be protected in isolation)", outgoing.ChanID.ToUint64(), + incoming, d.sufficient) + + 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. +func (m *Manager) OnSettle(incoming, outgoing models.CircuitKey) { + m.resolve(incoming, outgoing, true) +} + +// OnFail observes the failed resolution of a forwarded HTLC. +func (m *Manager) OnFail(incoming, outgoing models.CircuitKey) { + m.resolve(incoming, outgoing, false) +} + +// resolve applies an HTLC resolution under the lock. +func (m *Manager) resolve(incoming, outgoing models.CircuitKey, settled bool) { + at := m.now() + + m.mu.Lock() + defer m.mu.Unlock() + + if err := m.resolveHTLC(incoming, outgoing, settled, at); err != nil { + log.Warnf("Reputation resolve(in=%v out=%v settled=%v) error: "+ + "%v", incoming, outgoing, 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 uint64) *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, outgoing models.CircuitKey, + incomingAmt, outgoingAmt, advertisedFee lnwire.MilliSatoshi, + incomingCltv, height uint32, accountable bool, + at uint64) (isolationDecision, error) { + + if outgoingAmt > incomingAmt { + return isolationDecision{}, fmt.Errorf("outgoing amount %v "+ + "exceeds incoming %v", outgoingAmt, incomingAmt) + } + + // 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 isolationDecision{}, fmt.Errorf("incoming cltv %d not "+ + "beyond current height %d", incomingCltv, height) + } + + outScid := outgoing.ChanID.ToUint64() + inScid := incoming.ChanID.ToUint64() + + // Initialise any lazily-created channel state as of the event timestamp + // (a fresh m.now() 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) + + ref := incoming + if _, ok := outChan.pendingHTLCs[ref]; ok { + return isolationDecision{}, fmt.Errorf("duplicate htlc %v", ref) + } + + // 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) + htlcRisk := m.cfg.inFlightRisk(fee, incomingCltv, height) + + outReputation, err := outChan.outgoingReputation.valueAt(at) + if err != nil { + return isolationDecision{}, err + } + + // Score the HTLC in isolation: pass only this HTLC's risk, not the + // summed risk of other pending HTLCs on the outgoing channel. + sufficient, threshold, err := inChan.sufficientReputation( + htlcRisk, outReputation, at, + ) + if err != nil { + return isolationDecision{}, err + } + + outChan.pendingHTLCs[ref] = &pendingHTLC{ + fee: fee, + accountable: accountable, + addedAt: at, + maxHoldSeconds: maxHoldSeconds(incomingCltv, height), + } + + return isolationDecision{ + sufficient: sufficient, + outgoingReputation: outReputation, + htlcRisk: htlcRisk, + threshold: threshold, + }, nil +} + +// resolveHTLC applies an HTLC resolution to reputation and revenue. Caller must +// hold mu. +func (m *Manager) resolveHTLC(incoming, outgoing models.CircuitKey, + settled bool, at uint64) error { + + inScid := incoming.ChanID.ToUint64() + outScid := outgoing.ChanID.ToUint64() + + outChan, ok := m.channels[outScid] + if !ok { + // Tolerate: we never saw the forward (e.g. enabled mid-flight). + log.Debugf("Reputation resolve for unknown outgoing channel "+ + "%d (htlc %v); ignoring", outScid, incoming) + + return nil + } + + ref := incoming + pending, ok := outChan.pendingHTLCs[ref] + if !ok { + log.Debugf("Reputation resolve for unmatched htlc %v; ignoring", + incoming) + + return nil + } + + // Validate the timestamp before mutating any state: if the resolution + // predates the add we leave the pending entry and the averages + // untouched rather than deleting the pending and then failing, which + // would leave a lost pending that can never resolve. + if at < pending.addedAt { + return errBackwardsTime + } + resolutionTime := time.Duration(at-pending.addedAt) * time.Second + + effFee := m.cfg.effectiveFee( + pending.fee, resolutionTime, pending.accountable, settled, + ) + + newRep, err := outChan.outgoingReputation.add(effFee, at) + if err != nil { + return err + } + + if settled { + inChan := m.getOrCreateChannel(inScid, at) + fee := satFromUint(pending.fee).Int64() + if _, err := inChan.incomingRevenue.add(fee, at); err != nil { + return err + } + } + + // All updates succeeded; now it is safe to drop the pending entry. + delete(outChan.pendingHTLCs, ref) + + // Log a greppable summary when a resolution moves reputation. The + // phrasing is stable: integration tests match on it. + switch { + case effFee > 0: + log.Infof("Reputation gained: outgoing=%v eff_fee=%d "+ + "new_outgoing_reputation=%d", outScid, effFee, newRep) + + case effFee < 0: + log.Infof("Reputation lost: outgoing=%v eff_fee=%d "+ + "new_outgoing_reputation=%d", outScid, effFee, newRep) + + default: + log.Debugf("Reputation resolve in=%v out=%v settled=%v "+ + "eff_fee=0 (neutral)", incoming, outgoing, settled) + } + + return nil +} + +// gcLoop runs the periodic stale-pending garbage collector until the manager +// is stopped. +func (m *Manager) gcLoop() { + defer m.wg.Done() + + ticker := time.NewTicker(gcInterval) + defer ticker.Stop() + + for { + select { + case <-m.quit: + return + + case <-ticker.C: + m.gcStalePendings() + } + } +} + +// gcStalePendings evicts pending HTLCs that have outlived their worst-case hold +// time. Such an eviction should never happen in normal operation (every +// forwarded HTLC is resolved by the switch), so it is logged as a warning: it +// signals that the manager's in-flight view has diverged from the switch's. +func (m *Manager) gcStalePendings() { + at := m.now() + + m.mu.Lock() + defer m.mu.Unlock() + + var evicted int + for scid, ch := range m.channels { + for ref, p := range ch.pendingHTLCs { + if at <= p.addedAt+p.maxHoldSeconds { + continue + } + + delete(ch.pendingHTLCs, ref) + evicted++ + + log.Warnf("Reputation GC evicted stale pending htlc "+ + "%v on outgoing channel %d (added_at=%d, "+ + "max_hold=%ds): its resolution was never "+ + "observed", ref, scid, p.addedAt, + p.maxHoldSeconds) + } + } + + if evicted > 0 { + log.Warnf("Reputation GC evicted %d stale pending HTLC(s); "+ + "the manager's in-flight view may have diverged from "+ + "the switch", evicted) + } +} + +// unixSeconds expresses a time as whole unix seconds, the granularity used +// throughout the reputation math. +func unixSeconds(t time.Time) uint64 { + s := t.Unix() + if s < 0 { + return 0 + } + + return uint64(s) +} diff --git a/reputation/manager_test.go b/reputation/manager_test.go new file mode 100644 index 0000000000..2c36747fa4 --- /dev/null +++ b/reputation/manager_test.go @@ -0,0 +1,284 @@ +package reputation + +import ( + "testing" + "time" + + "github.com/lightningnetwork/lnd/clock" +) + +// 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) + if err != nil { + t.Fatalf("NewManager: %v", err) + } + if err := m.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + 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)), + ) + if err != nil { + t.Fatalf("NewManager: %v", err) + } + + if err := m.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + + // Hooks on an empty manager must be safe no-ops (other than lazy + // channel creation) — they must never panic. + m.OnSettle(circuit(1, 0), circuit(2, 0)) + m.OnFail(circuit(1, 0), circuit(2, 0)) + + if err := m.Stop(); err != nil { + t.Fatalf("Stop: %v", 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 := circuit(2, 0) + + // advertised fee = 1000 (equals in-out here). cltv 200 > height 100. + m.OnForward(in, out, 2000, 1000, 1000, 200, testHeight, false) + + outChan := m.channels[2] + if len(outChan.pendingHTLCs) != 1 { + t.Fatalf("expected 1 pending, got %d", + len(outChan.pendingHTLCs)) + } + + // Settle 30s later (within resolution period). + advance(clk, 30*time.Second) + m.OnSettle(in, out) + + if len(outChan.pendingHTLCs) != 0 { + t.Fatalf("pending not cleared: %d", len(outChan.pendingHTLCs)) + } + + rep, err := outChan.outgoingReputation.valueAt(m.now()) + if err != nil { + t.Fatalf("valueAt: %v", err) + } + if rep != 1000 { + t.Fatalf("reputation: got %d, want 1000", rep) + } + + // Incoming channel earned the fee as revenue. + inChan := m.channels[1] + rev, err := inChan.incomingRevenue.valueAt(m.now()) + if err != nil { + t.Fatalf("valueAt: %v", err) + } + if rev <= 0 { + t.Fatalf("revenue: got %d, want > 0", rev) + } +} + +// 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), circuit(2, 0) + + m.OnForward(in, out, 2000, 1000, 1000, 200, testHeight, false) + advance(clk, 30*time.Second) + m.OnFail(in, out) + + rep, _ := m.channels[2].outgoingReputation.valueAt(m.now()) + if rep != 0 { + t.Fatalf("reputation after fail: got %d, want 0", rep) + } + rev, _ := m.channels[1].incomingRevenue.valueAt(m.now()) + if rev != 0 { + t.Fatalf("revenue after fail: got %d, want 0", rev) + } +} + +// 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), circuit(2, 99)) + m.OnFail(circuit(1, 88), circuit(2, 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), circuit(2, 0) + + // 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 && len(c.pendingHTLCs) != 0 { + t.Fatalf("expired-cltv forward must not create a pending htlc") + } +} + +// TestGCStalePending verifies stale pendings are evicted past their max hold. +func TestGCStalePending(t *testing.T) { + t.Parallel() + + const start = 1_000_000 + m, clk := buildManager(t, start) + + // cltv 101 at height 100 => max hold 600s. + if _, err := m.addHTLC( + circuit(1, 0), circuit(2, 0), 2000, 1000, 1000, 101, + testHeight, false, start, + ); err != nil { + t.Fatalf("addHTLC: %v", err) + } + + // Advance past the max hold and run GC. + advance(clk, 700*time.Second) + m.gcStalePendings() + + if len(m.channels[2].pendingHTLCs) != 0 { + t.Fatalf("stale pending not evicted") + } +} + +// TestSufficiencyBoundary unit-tests the core reputation inequality at its +// boundary. +func TestSufficiencyBoundary(t *testing.T) { + t.Parallel() + + cfg := DefaultConfig() + const start = 1_000_000 + c := newChannelReputation(cfg, start) + + // Seed an incoming-revenue threshold of 1000 and read it back so the + // aggregated average's warmup divisor is settled. + if _, err := c.incomingRevenue.add(1000, start); err != nil { + t.Fatalf("add: %v", err) + } + threshold, err := c.incomingRevenue.valueAt(start) + if err != nil { + t.Fatalf("valueAt: %v", err) + } + + // Reputation exactly at threshold, no in-flight risk => sufficient. + ok, _, err := c.sufficientReputation(0, threshold, start) + if err != nil { + t.Fatalf("sufficientReputation: %v", err) + } + if !ok { + t.Fatalf("expected sufficient at threshold") + } + + // One msat below threshold => insufficient. + ok, _, err = c.sufficientReputation(0, threshold-1, start) + if err != nil { + t.Fatalf("sufficientReputation: %v", err) + } + if ok { + t.Fatalf("expected insufficient below threshold") + } + + // At threshold but with in-flight risk => insufficient. + ok, _, err = c.sufficientReputation(1, threshold, start) + if err != nil { + t.Fatalf("sufficientReputation: %v", err) + } + if ok { + t.Fatalf("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 + + addHTLC := func(m *Manager) (isolationDecision, error) { + return m.addHTLC( + circuit(1, 0), circuit(2, 0), 2000, 1000, 1000, 200, + testHeight, true, start, + ) + } + + t.Run("zero reputation insufficient", func(t *testing.T) { + m, _ := buildManager(t, start) + + // Give the incoming channel a positive revenue threshold so the + // (zero) outgoing reputation is insufficient. + inChan := m.getOrCreateChannel(1, uint64(start)) + if _, err := inChan.incomingRevenue.add( + 1_000_000, uint64(start), + ); err != nil { + t.Fatalf("seed revenue: %v", err) + } + + d, err := addHTLC(m) + if err != nil { + t.Fatalf("addHTLC: %v", err) + } + if d.sufficient { + t.Fatalf("expected insufficient, got %s", d) + } + }) + + t.Run("ample reputation sufficient", func(t *testing.T) { + m, _ := buildManager(t, start) + + // Give the outgoing channel ample reputation. + outChan := m.getOrCreateChannel(2, uint64(start)) + if _, err := outChan.outgoingReputation.add( + 10_000_000, uint64(start), + ); err != nil { + t.Fatalf("seed reputation: %v", err) + } + + d, err := addHTLC(m) + if err != nil { + t.Fatalf("addHTLC: %v", err) + } + if !d.sufficient { + t.Fatalf("expected sufficient, got %s", d) + } + }) +} diff --git a/reputation/testutil_test.go b/reputation/testutil_test.go new file mode 100644 index 0000000000..c713736b88 --- /dev/null +++ b/reputation/testutil_test.go @@ -0,0 +1,22 @@ +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(scid, htlcID uint64) models.CircuitKey { + return models.CircuitKey{ + ChanID: lnwire.NewShortChanIDFromInt(scid), + HtlcID: htlcID, + } +} + +// advance moves a test clock forward by d. +func advance(c *clock.TestClock, d time.Duration) { + c.SetTime(c.Now().Add(d)) +} From 87820542e2cb4e3c92db20219900cc7fe1d6e91a Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Mon, 27 Jul 2026 16:11:07 +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. 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 circuit keys, a nil manager is a no-op, local sends are skipped, and a hook panic is absorbed by the guard. --- htlcswitch/interfaces.go | 30 ++ htlcswitch/link.go | 14 + htlcswitch/mock.go | 10 + htlcswitch/reputation_guard.go | 82 ++++ htlcswitch/reputation_hooks_test.go | 641 ++++++++++++++++++++++++++++ htlcswitch/switch.go | 95 +++++ lncfg/routing.go | 2 + log.go | 4 + reputation_adapter.go | 27 ++ sample-lnd.conf | 5 + server.go | 48 +++ 11 files changed, 958 insertions(+) create mode 100644 htlcswitch/reputation_guard.go create mode 100644 htlcswitch/reputation_hooks_test.go create mode 100644 reputation_adapter.go diff --git a/htlcswitch/interfaces.go b/htlcswitch/interfaces.go index ef62eb71d9..8d2d0c14a3 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,29 @@ 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 fee the node advertised on the outgoing link for this forward + // (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. + OnForward(incoming, outgoing CircuitKey, incomingAmt, + outgoingAmt, advertisedFee lnwire.MilliSatoshi, + incomingCltv, height uint32, accountable bool) + + // OnSettle observes the successful resolution of a forwarded HTLC. + OnSettle(incoming, outgoing CircuitKey) + + // OnFail observes the failed resolution of a forwarded HTLC. + OnFail(incoming, outgoing 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 9e3adf0bbc..74a8be1a29 100644 --- a/htlcswitch/link.go +++ b/htlcswitch/link.go @@ -2482,6 +2482,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 diff --git a/htlcswitch/mock.go b/htlcswitch/mock.go index a3079e6962..ac433552a9 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 0000000000..7e8a4d2503 --- /dev/null +++ b/htlcswitch/reputation_guard.go @@ -0,0 +1,82 @@ +package htlcswitch + +import ( + "sync/atomic" + + "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, permanently disable the subsystem (fail open), +// and continue forwarding unaffected. +// +// The hooks run synchronously on the switch's forwarding goroutine, so this +// boundary keeps a subsystem bug — a nil deref, an arithmetic panic — from +// taking down the node's HTLC forwarding. +type guardedReputationManager struct { + inner ReputationManager + disabled atomic.Bool +} + +// 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, outgoing CircuitKey, + incomingAmt, outgoingAmt, advertisedFee lnwire.MilliSatoshi, + incomingCltv, height uint32, accountable bool) { + + if g.disabled.Load() { + return + } + 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, outgoing CircuitKey) { + if g.disabled.Load() { + return + } + defer g.recoverHook("OnSettle") + + g.inner.OnSettle(incoming, outgoing) +} + +// OnFail forwards the observation to the wrapped manager behind a panic +// boundary. +func (g *guardedReputationManager) OnFail(incoming, outgoing CircuitKey) { + if g.disabled.Load() { + return + } + defer g.recoverHook("OnFail") + + g.inner.OnFail(incoming, outgoing) +} + +// recoverHook recovers from a panic in a reputation hook, logging it and +// permanently disabling the subsystem so a deterministic bug cannot panic on +// every forwarded HTLC. Forwarding is never affected. +func (g *guardedReputationManager) recoverHook(method string) { + if r := recover(); r != nil { + log.Errorf("Reputation %s hook panicked; disabling reputation "+ + "subsystem (forwarding is unaffected): %v", method, r) + g.disabled.Store(true) + } +} diff --git a/htlcswitch/reputation_hooks_test.go b/htlcswitch/reputation_hooks_test.go new file mode 100644 index 0000000000..026114a380 --- /dev/null +++ b/htlcswitch/reputation_hooks_test.go @@ -0,0 +1,641 @@ +package htlcswitch + +import ( + "crypto/sha256" + "sync" + "testing" + "time" + + "github.com/lightningnetwork/lnd/lnwire" +) + +// 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, out CircuitKey + inAmt, outAmt lnwire.MilliSatoshi + advertisedFee lnwire.MilliSatoshi + cltv uint32 + height uint32 + accountable bool +} + +type repResolve struct { + in, out CircuitKey +} + +func (r *mockReputationManager) OnForward(in, out CircuitKey, 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, out CircuitKey) { + r.mu.Lock() + defer r.mu.Unlock() + r.settles = append(r.settles, repResolve{in: in, out: out}) +} + +func (r *mockReputationManager) OnFail(in, out CircuitKey) { + r.mu.Lock() + defer r.mu.Unlock() + r.fails = append(r.fails, repResolve{in: in, out: out}) +} + +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.ChanID != 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() + if len(settles) != 1 { + t.Fatalf("expected 1 OnSettle, got %d", len(settles)) + } + if settles[0].in.ChanID != aliceLink.ShortChanID() { + t.Fatalf("OnSettle wrong incoming key: %v", settles[0].in) + } + if len(fails) != 0 { + t.Fatalf("expected 0 OnFail, got %d", len(fails)) + } +} + +// 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() + if len(forwards) != 1 { + t.Fatalf("expected 1 OnForward, got %d", len(forwards)) + } + if len(fails) != 1 { + t.Fatalf("expected 1 OnFail, got %d", len(fails)) + } + if fails[0].in.ChanID != aliceLink.ShortChanID() { + t.Fatalf("OnFail wrong incoming key: %v", fails[0].in) + } + if len(settles) != 0 { + t.Fatalf("expected 0 OnSettle, got %d", len(settles)) + } +} + +// 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 we can assert the guard self-disables. +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.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 and +// permanently disables the subsystem (fail open) — so 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 := CircuitKey{ChanID: lnwire.NewShortChanIDFromInt(2)} + + // The first call panics internally; the guard must recover so the + // caller's goroutine is unaffected. + mustNotPanic(t, func() { + guard.OnForward(in, out, 1, 1, 0, 100, 90, false) + }) + if inner.callCount() != 1 { + t.Fatalf("inner should have been called once, got %d", + inner.callCount()) + } + + // After a panic the subsystem is disabled: subsequent calls short- + // circuit without ever reaching the (panicking) inner manager. + mustNotPanic(t, func() { + guard.OnForward(in, out, 1, 1, 0, 100, 90, false) + guard.OnSettle(in, out) + guard.OnFail(in, out) + }) + if inner.callCount() != 1 { + t.Fatalf("inner must not be called again once disabled, got %d", + 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 — 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 outgoing +// link's ADVERTISED fee (not the offered in-out delta) to OnForward. +func TestSwitchReputationAdvertisedFee(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. + const wantFee = lnwire.MilliSatoshi(4242) + bobLink.advertisedFee = wantFee + + 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("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 != wantFee { + t.Fatalf("advertised fee: got %d, want %d", + forwards[0].advertisedFee, 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") + } + }) +} diff --git a/htlcswitch/switch.go b/htlcswitch/switch.go index 2c0bbddb62..eb0548a57e 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,65 @@ 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 on the outgoing link for this + // forward (base fee + proportional over the outgoing amount), + // 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. + advertisedFee := destination.AdvertisedFee(packet.amount) + + // 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, + }, + CircuitKey{ + ChanID: packet.outgoingChanID, + HtlcID: packet.outgoingHTLCID, + }, + 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 +3172,14 @@ 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, *circuit.Outgoing, + ) + } } // Deliver this packet. @@ -3135,6 +3214,22 @@ func (s *Switch) handlePacketFail(packet *htlcPacket, return nil } + // Feed the read-only reputation manager this forwarded fail; + // log-only, never affects resolution. We resolve against the + // incoming circuit (the key the reputation manager tracks); the + // outgoing leg is taken from the circuit keystone when available, + // else from the packet. + if s.cfg.ReputationManager != nil && circuit != nil { + outgoing := CircuitKey{ + ChanID: packet.outgoingChanID, + HtlcID: packet.outgoingHTLCID, + } + if circuit.Outgoing != nil { + outgoing = *circuit.Outgoing + } + s.cfg.ReputationManager.OnFail(circuit.Incoming, outgoing) + } + // 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 8967578870..61107ce839 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 563ee3eb11..ee25266b22 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/reputation_adapter.go b/reputation_adapter.go new file mode 100644 index 0000000000..33975c835b --- /dev/null +++ b/reputation_adapter.go @@ -0,0 +1,27 @@ +package lnd + +import ( + "github.com/lightningnetwork/lnd/htlcswitch" + "github.com/lightningnetwork/lnd/reputation" +) + +// reputationManagerAdapter bridges the reputation.Manager to the switch's +// read-only htlcswitch.ReputationManager seam. The manager's hook signatures +// already match the switch interface (both use models.CircuitKey and +// lnwire.MilliSatoshi), so this is a thin, explicit bridge that keeps the +// switch from importing the reputation package directly. +type reputationManagerAdapter struct { + *reputation.Manager +} + +// Compile-time assertion that the adapter satisfies the switch's read-only +// reputation seam. +var _ htlcswitch.ReputationManager = (*reputationManagerAdapter)(nil) + +// newReputationManagerAdapter wraps a reputation.Manager as an +// htlcswitch.ReputationManager. +func newReputationManagerAdapter( + m *reputation.Manager) htlcswitch.ReputationManager { + + return &reputationManagerAdapter{Manager: m} +} diff --git a/sample-lnd.conf b/sample-lnd.conf index f881c1174e..5282d5416a 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 a0312bb014..6e3a9fbca8 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,10 @@ type server struct { htlcNotifier *htlcswitch.HtlcNotifier + // reputationMgr is the optional read-only local reputation subsystem, + // non-nil only when the experimental reputation flag is set. + reputationMgr *reputation.Manager + witnessBeacon contractcourt.WitnessBeacon breachArbitrator *contractcourt.BreachArbitrator @@ -880,6 +885,31 @@ 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. On a hook panic the guard logs, disables the + // subsystem, and forwarding continues unaffected. + repMgrIface = htlcswitch.NewGuardedReputationManager( + newReputationManagerAdapter(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 e57ff3a882d787819f7b73d7de7926257415d82d Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Mon, 27 Jul 2026 16:11:07 +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 | 108 +++++++++++++++++++++++++++++++++++ lntest/harness_assertion.go | 44 ++++++++++++++ 3 files changed, 156 insertions(+) create mode 100644 itest/lnd_reputation_test.go diff --git a/itest/list_on_test.go b/itest/list_on_test.go index f1e2eb48ec..ebe081fe37 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 0000000000..021fecf516 --- /dev/null +++ b/itest/lnd_reputation_test.go @@ -0,0 +1,108 @@ +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" +) + +// 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 after the forward settles his log +// must contain the "Reputation gained" summary (emitted only when a resolution +// yields a positive effective fee). After a restart — which resets the +// in-memory state — a further forward must re-accrue reputation, 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=") + + // The forward earned Bob a fee, which his reputation subsystem records + // as a positive reputation gain for the outgoing (Bob -> Carol) + // channel. The "Reputation gained" summary is logged only when the + // effective fee is positive, so its presence confirms the subsystem + // observed the forward+settle and computed a real, positive update. + ht.AssertNodeLogContains(bob, "Reputation gained: outgoing=") + + // 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. + 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: after this second + // forward+settle Bob again logs a positive reputation gain, proving the + // reset subsystem rebuilt its state from scratch. + ht.AssertNodeLogContains(bob, "Reputation gained: outgoing=") + + ht.CloseChannel(alice, chanPointAB) + ht.CloseChannel(bob, chanPointBC) +} diff --git a/lntest/harness_assertion.go b/lntest/harness_assertion.go index 03c1819ff7..0567e08dd5 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,48 @@ 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) { + + err := wait.NoError(func() error { + var found bool + _ = filepath.WalkDir(hn.Cfg.LogDir, func(path string, + d os.DirEntry, err error) error { + + if err != nil || d.IsDir() || found { + return nil + } + if filepath.Base(path) != "lnd.log" { + return nil + } + + data, readErr := os.ReadFile(path) + if readErr == nil && strings.Contains( + string(data), substr, + ) { + + found = true + } + + return nil + }) + + if found { + return nil + } + + return fmt.Errorf("%s log does not contain %q", hn.Name(), + substr) + }, DefaultTimeout) + + require.NoError(h, err, "timeout waiting for log substring %q", substr) +} + // WaitForBlockchainSyncTo waits until the node is synced to bestBlock. func (h *HarnessTest) WaitForBlockchainSyncTo(hn *node.HarnessNode, bestBlock chainhash.Hash) { From c6480fe1b0fd90503bdbb15ff7049ae5046d02a7 Mon Sep 17 00:00:00 2001 From: George Tsagkarelis Date: Mon, 27 Jul 2026 16:11:07 +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 85626b68f4..9e52831372 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