[1/?] Local reputation: subsystem core, read only

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
+```

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
+}

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)
+	}
+}

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
+}