multi: add htlcNotifier interface to switch and link

In this commit, a htlcNotifier interface is added to allow for easy
unit testing. Instances of the HtlcNotifier are added to the server,
switch and link.

htlcswitch/htlcnotifier.go

@@ -289,6 +289,8 @@ type SettleEvent struct {
 
 // NotifyForwardingEvent notifies the HtlcNotifier than a htlc has been
 // forwarded.
+//
+// Note this is part of the htlcNotifier interface.
 func (h *HtlcNotifier) NotifyForwardingEvent(key HtlcKey, info HtlcInfo,
 	eventType HtlcEventType) {
 
@@ -309,6 +311,8 @@ func (h *HtlcNotifier) NotifyForwardingEvent(key HtlcKey, info HtlcInfo,
 
 // NotifyLinkFailEvent notifies that a htlc has failed on our incoming
 // or outgoing link.
+//
+// Note this is part of the htlcNotifier interface.
 func (h *HtlcNotifier) NotifyLinkFailEvent(key HtlcKey, info HtlcInfo,
 	eventType HtlcEventType, linkErr *LinkError, incoming bool) {
 
@@ -331,6 +335,8 @@ func (h *HtlcNotifier) NotifyLinkFailEvent(key HtlcKey, info HtlcInfo,
 
 // NotifyForwardingFailEvent notifies the HtlcNotifier that a htlc we
 // forwarded has failed down the line.
+//
+// Note this is part of the htlcNotifier interface.
 func (h *HtlcNotifier) NotifyForwardingFailEvent(key HtlcKey,
 	eventType HtlcEventType) {
 
@@ -350,6 +356,8 @@ func (h *HtlcNotifier) NotifyForwardingFailEvent(key HtlcKey,
 
 // NotifySettleEvent notifies the HtlcNotifier that a htlc that we committed
 // to as part of a forward or a receive to our node has been settled.
+//
+// Note this is part of the htlcNotifier interface.
 func (h *HtlcNotifier) NotifySettleEvent(key HtlcKey, eventType HtlcEventType) {
 	event := &SettleEvent{
 		HtlcKey:       key,

htlcswitch/interfaces.go

@@ -180,3 +180,29 @@ type TowerClient interface {
 	// isTweakless should be true.
 	BackupState(*lnwire.ChannelID, *lnwallet.BreachRetribution, bool) error
 }
+
+// htlcNotifier is an interface which represents the input side of the
+// HtlcNotifier which htlc events are piped through. This interface is intended
+// to allow for mocking of the htlcNotifier in tests, so is unexported because
+// it is not needed outside of the htlcSwitch package.
+type htlcNotifier interface {
+	// NotifyForwardingEvent notifies the HtlcNotifier than a htlc has been
+	// forwarded.
+	NotifyForwardingEvent(key HtlcKey, info HtlcInfo,
+		eventType HtlcEventType)
+
+	// NotifyIncomingLinkFailEvent notifies that a htlc has failed on our
+	// incoming link. It takes an isReceive bool to differentiate between
+	// our node's receives and forwards.
+	NotifyLinkFailEvent(key HtlcKey, info HtlcInfo,
+		eventType HtlcEventType, linkErr *LinkError, incoming bool)
+
+	// NotifyForwardingFailEvent notifies the HtlcNotifier that a htlc we
+	// forwarded has failed down the line.
+	NotifyForwardingFailEvent(key HtlcKey, eventType HtlcEventType)
+
+	// NotifySettleEvent notifies the HtlcNotifier that a htlc that we
+	// committed to as part of a forward or a receive to our node has been
+	// settled.
+	NotifySettleEvent(key HtlcKey, eventType HtlcEventType)
+}

htlcswitch/link.go

@@ -272,6 +272,10 @@ type ChannelLinkConfig struct {
 	// NotifyInactiveChannel allows the switch to tell the ChannelNotifier
 	// when channels become inactive.
 	NotifyInactiveChannel func(wire.OutPoint)
+
+	// HtlcNotifier is an instance of a htlcNotifier which we will pipe htlc
+	// events through.
+	HtlcNotifier htlcNotifier
 }
 
 // channelLink is the service which drives a channel's commitment update

htlcswitch/link_test.go

@@ -1750,6 +1750,7 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		MaxFeeAllocation:      DefaultMaxLinkFeeAllocation,
 		NotifyActiveChannel:   func(wire.OutPoint) {},
 		NotifyInactiveChannel: func(wire.OutPoint) {},
+		HtlcNotifier:          aliceSwitch.cfg.HtlcNotifier,
 	}
 
 	aliceLink := NewChannelLink(aliceCfg, aliceLc.channel)
@@ -4313,6 +4314,7 @@ func (h *persistentLinkHarness) restartLink(
 		MaxFeeAllocation:      DefaultMaxLinkFeeAllocation,
 		NotifyActiveChannel:   func(wire.OutPoint) {},
 		NotifyInactiveChannel: func(wire.OutPoint) {},
+		HtlcNotifier:          aliceSwitch.cfg.HtlcNotifier,
 	}
 
 	aliceLink := NewChannelLink(aliceCfg, aliceChannel)
@@ -5523,6 +5525,7 @@ func TestCheckHtlcForward(t *testing.T) {
 			},
 			FetchLastChannelUpdate: fetchLastChannelUpdate,
 			MaxOutgoingCltvExpiry:  DefaultMaxOutgoingCltvExpiry,
+			HtlcNotifier:           &mockHTLCNotifier{},
 		},
 		log:           log,
 		channel:       testChannel.channel,

htlcswitch/mock.go

@@ -176,6 +176,7 @@ func initSwitchWithDB(startingHeight uint32, db *channeldb.DB) (*Switch, error)
 		FwdEventTicker: ticker.NewForce(DefaultFwdEventInterval),
 		LogEventTicker: ticker.NewForce(DefaultLogInterval),
 		AckEventTicker: ticker.NewForce(DefaultAckInterval),
+		HtlcNotifier:   &mockHTLCNotifier{},
 	}
 
 	return New(cfg, startingHeight)
@@ -1009,3 +1010,22 @@ func (m *mockOnionErrorDecryptor) DecryptError(encryptedData []byte) (
 		Message:   m.message,
 	}, m.err
 }
+
+var _ htlcNotifier = (*mockHTLCNotifier)(nil)
+
+type mockHTLCNotifier struct{}
+
+func (h *mockHTLCNotifier) NotifyForwardingEvent(key HtlcKey, info HtlcInfo,
+	eventType HtlcEventType) {
+}
+
+func (h *mockHTLCNotifier) NotifyLinkFailEvent(key HtlcKey, info HtlcInfo,
+	eventType HtlcEventType, linkErr *LinkError, incoming bool) {
+}
+
+func (h *mockHTLCNotifier) NotifyForwardingFailEvent(key HtlcKey,
+	eventType HtlcEventType) {
+}
+
+func (h *mockHTLCNotifier) NotifySettleEvent(key HtlcKey, eventType HtlcEventType) {
+}

htlcswitch/switch.go

@@ -150,6 +150,10 @@ type Config struct {
 	// the switch when a new block has arrived.
 	Notifier chainntnfs.ChainNotifier
 
+	// HtlcNotifier is an instance of a htlcNotifier which we will pipe htlc
+	// events through.
+	HtlcNotifier htlcNotifier
+
 	// FwdEventTicker is a signal that instructs the htlcswitch to flush any
 	// pending forwarding events.
 	FwdEventTicker ticker.Ticker

htlcswitch/test_utils.go

@@ -1139,6 +1139,7 @@ func (h *hopNetwork) createChannelLink(server, peer *mockServer,
 			MaxFeeAllocation:        DefaultMaxLinkFeeAllocation,
 			NotifyActiveChannel:     func(wire.OutPoint) {},
 			NotifyInactiveChannel:   func(wire.OutPoint) {},
+			HtlcNotifier:            server.htlcSwitch.cfg.HtlcNotifier,
 		},
 		channel,
 	)

peer.go

@@ -636,6 +636,7 @@ func (p *peer) addLink(chanPoint *wire.OutPoint,
 		MaxFeeAllocation:        cfg.MaxChannelFeeAllocation,
 		NotifyActiveChannel:     p.server.channelNotifier.NotifyActiveChannelEvent,
 		NotifyInactiveChannel:   p.server.channelNotifier.NotifyInactiveChannelEvent,
+		HtlcNotifier:            p.server.htlcNotifier,
 	}
 
 	link := htlcswitch.NewChannelLink(linkCfg, lnChan)

server.go

@@ -203,6 +203,8 @@ type server struct {
 
 	peerNotifier *peernotifier.PeerNotifier
 
+	htlcNotifier *htlcswitch.HtlcNotifier
+
 	witnessBeacon contractcourt.WitnessBeacon
 
 	breachArbiter *breachArbiter
@@ -438,6 +440,8 @@ func newServer(listenAddrs []net.Addr, chanDB *channeldb.DB,
 		return nil, err
 	}
 
+	s.htlcNotifier = htlcswitch.NewHtlcNotifier(time.Now)
+
 	s.htlcSwitch, err = htlcswitch.New(htlcswitch.Config{
 		DB: chanDB,
 		LocalChannelClose: func(pubKey []byte,
@@ -467,6 +471,7 @@ func newServer(listenAddrs []net.Addr, chanDB *channeldb.DB,
 		ExtractErrorEncrypter:  s.sphinx.ExtractErrorEncrypter,
 		FetchLastChannelUpdate: s.fetchLastChanUpdate(),
 		Notifier:               s.cc.chainNotifier,
+		HtlcNotifier:           s.htlcNotifier,
 		FwdEventTicker:         ticker.New(htlcswitch.DefaultFwdEventInterval),
 		LogEventTicker:         ticker.New(htlcswitch.DefaultLogInterval),
 		AckEventTicker:         ticker.New(htlcswitch.DefaultAckInterval),
@@ -1265,6 +1270,10 @@ func (s *server) Start() error {
 			startErr = err
 			return
 		}
+		if err := s.htlcNotifier.Start(); err != nil {
+			startErr = err
+			return
+		}
 		if err := s.sphinx.Start(); err != nil {
 			startErr = err
 			return
@@ -1429,6 +1438,7 @@ func (s *server) Stop() error {
 		s.sweeper.Stop()
 		s.channelNotifier.Stop()
 		s.peerNotifier.Stop()
+		s.htlcNotifier.Stop()
 		s.cc.wallet.Shutdown()
 		s.cc.chainView.Stop()
 		s.connMgr.Stop()