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