Merge pull request #3781 from carlaKC/3420-htlcForwarding

htlcswitch: add htlcNotifier

htlcswitch/htlcnotifier.go

@@ -0,0 +1,429 @@
+package htlcswitch
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/htlcswitch/hop"
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/subscribe"
+)
+
+// HtlcNotifier notifies clients of htlc forwards, failures and settles for
+// htlcs that the switch handles. It takes subscriptions for its events and
+// notifies them when htlc events occur. These are served on a best-effort
+// basis; events are not persisted, delivery is not guaranteed (in the event
+// of a crash in the switch, forward events may be lost) and some events may
+// be replayed upon restart. Events consumed from this package should be
+// de-duplicated by the htlc's unique combination of incoming+outgoing circuit
+// and not relied upon for critical operations.
+//
+// The htlc notifier sends the following kinds of events:
+// Forwarding Event:
+// - Represents a htlc which is forwarded onward from our node.
+// - Present for htlc forwards through our node and local sends.
+//
+// Link Failure Event:
+// - Indicates that a htlc has failed on our incoming or outgoing link,
+//   with an incoming boolean which indicates where the failure occurred.
+// - Incoming link failures are present for failed attempts to pay one of
+//   our invoices (insufficient amount or mpp timeout, for example) and for
+//   forwards that we cannot decode to forward onwards.
+// - Outgoing link failures are present for forwards or local payments that
+//   do not meet our outgoing link's policy (insufficient fees, for example)
+//   and when we fail to forward the payment on (insufficient outgoing
+//   capacity, or an unknown outgoing link).
+//
+// Forwarding Failure Event:
+// - Forwarding failures indicate that a htlc we forwarded has failed at
+//   another node down the route.
+// - Present for local sends and htlc forwards which fail after they left
+//   our node.
+//
+// Settle event:
+// - Settle events are present when a htlc which we added is settled through
+//   the release of a preimage.
+// - Present for local receives, and successful local sends or forwards.
+//
+// Each htlc is identified by its incoming and outgoing circuit key. Htlcs,
+// and their subsequent settles or fails, can be identified by the combination
+// of incoming and outgoing circuits. Note that receives to our node will
+// have a zero outgoing circuit key because the htlc terminates at our
+// node, and sends from our node will have a zero incoming circuit key because
+// the send originates at our node.
+type HtlcNotifier struct {
+	started sync.Once
+	stopped sync.Once
+
+	// now returns the current time, it is set in the htlcnotifier to allow
+	// for timestamp mocking in tests.
+	now func() time.Time
+
+	ntfnServer *subscribe.Server
+}
+
+// NewHtlcNotifier creates a new HtlcNotifier which gets htlc forwarded,
+// failed and settled events from links our node has established with peers
+// and sends notifications to subscribing clients.
+func NewHtlcNotifier(now func() time.Time) *HtlcNotifier {
+	return &HtlcNotifier{
+		now:        now,
+		ntfnServer: subscribe.NewServer(),
+	}
+}
+
+// Start starts the HtlcNotifier and all goroutines it needs to consume
+// events and provide subscriptions to clients.
+func (h *HtlcNotifier) Start() error {
+	var err error
+	h.started.Do(func() {
+		log.Trace("HtlcNotifier starting")
+		err = h.ntfnServer.Start()
+	})
+	return err
+}
+
+// Stop signals the notifier for a graceful shutdown.
+func (h *HtlcNotifier) Stop() {
+	h.stopped.Do(func() {
+		if err := h.ntfnServer.Stop(); err != nil {
+			log.Warnf("error stopping htlc notifier: %v", err)
+		}
+	})
+}
+
+// SubscribeHtlcEvents returns a subscribe.Client that will receive updates
+// any time the server is made aware of a new event.
+func (h *HtlcNotifier) SubscribeHtlcEvents() (*subscribe.Client, error) {
+	return h.ntfnServer.Subscribe()
+}
+
+// HtlcKey uniquely identifies the htlc.
+type HtlcKey struct {
+	// IncomingCircuit is the channel an htlc id of the incoming htlc.
+	IncomingCircuit channeldb.CircuitKey
+
+	// OutgoingCircuit is the channel and htlc id of the outgoing htlc.
+	OutgoingCircuit channeldb.CircuitKey
+}
+
+// String returns a string representation of a htlc key.
+func (k HtlcKey) String() string {
+	switch {
+	case k.IncomingCircuit.ChanID == hop.Source:
+		return k.OutgoingCircuit.String()
+
+	case k.OutgoingCircuit.ChanID == hop.Exit:
+		return k.IncomingCircuit.String()
+
+	default:
+		return fmt.Sprintf("%v -> %v", k.IncomingCircuit,
+			k.OutgoingCircuit)
+	}
+}
+
+// HtlcInfo provides the details of a htlc that our node has processed. For
+// forwards, incoming and outgoing values are set, whereas sends and receives
+// will only have outgoing or incoming details set.
+type HtlcInfo struct {
+	// IncomingTimelock is the time lock of the htlc on our incoming
+	// channel.
+	IncomingTimeLock uint32
+
+	// OutgoingTimelock is the time lock the htlc on our outgoing channel.
+	OutgoingTimeLock uint32
+
+	// IncomingAmt is the amount of the htlc on our incoming channel.
+	IncomingAmt lnwire.MilliSatoshi
+
+	// OutgoingAmt is the amount of the htlc on our outgoing channel.
+	OutgoingAmt lnwire.MilliSatoshi
+}
+
+// String returns a string representation of a htlc.
+func (h HtlcInfo) String() string {
+	var details []string
+
+	// If the incoming information is not zero, as is the case for a send,
+	// we include the incoming amount and timelock.
+	if h.IncomingAmt != 0 || h.IncomingTimeLock != 0 {
+		str := fmt.Sprintf("incoming amount: %v, "+
+			"incoming timelock: %v", h.IncomingAmt,
+			h.IncomingTimeLock)
+
+		details = append(details, str)
+	}
+
+	// If the outgoing information is not zero, as is the case for a
+	// receive, we include the outgoing amount and timelock.
+	if h.OutgoingAmt != 0 || h.OutgoingTimeLock != 0 {
+		str := fmt.Sprintf("outgoing amount: %v, "+
+			"outgoing timelock: %v", h.OutgoingAmt,
+			h.OutgoingTimeLock)
+
+		details = append(details, str)
+	}
+
+	return strings.Join(details, ", ")
+}
+
+// HtlcEventType represents the type of event that a htlc was part of.
+type HtlcEventType int
+
+const (
+	// HtlcEventTypeSend represents a htlc that was part of a send from
+	// our node.
+	HtlcEventTypeSend HtlcEventType = iota
+
+	// HtlcEventTypeReceive represents a htlc that was part of a receive
+	// to our node.
+	HtlcEventTypeReceive
+
+	// HtlcEventTypeForward represents a htlc that was forwarded through
+	// our node.
+	HtlcEventTypeForward
+)
+
+// String returns a string representation of a htlc event type.
+func (h HtlcEventType) String() string {
+	switch h {
+	case HtlcEventTypeSend:
+		return "send"
+
+	case HtlcEventTypeReceive:
+		return "receive"
+
+	case HtlcEventTypeForward:
+		return "forward"
+
+	default:
+		return "unknown"
+	}
+}
+
+// ForwardingEvent represents a htlc that was forwarded onwards from our node.
+// Sends which originate from our node will report forward events with zero
+// incoming circuits in their htlc key.
+type ForwardingEvent struct {
+	// HtlcKey uniquely identifies the htlc, and can be used to match the
+	// forwarding event with subsequent settle/fail events.
+	HtlcKey
+
+	// HtlcInfo contains details about the htlc.
+	HtlcInfo
+
+	// HtlcEventType classifies the event as part of a local send or
+	// receive, or as part of a forward.
+	HtlcEventType
+
+	// Timestamp is the time when this htlc was forwarded.
+	Timestamp time.Time
+}
+
+// LinkFailEvent describes a htlc that failed on our incoming or outgoing
+// link. The incoming bool is true for failures on incoming links, and false
+// for failures on outgoing links. The failure reason is provided by a lnwire
+// failure message which is enriched with a failure detail in the cases where
+// the wire failure message does not contain full information about the
+// failure.
+type LinkFailEvent struct {
+	// HtlcKey uniquely identifies the htlc.
+	HtlcKey
+
+	// HtlcInfo contains details about the htlc.
+	HtlcInfo
+
+	// HtlcEventType classifies the event as part of a local send or
+	// receive, or as part of a forward.
+	HtlcEventType
+
+	// LinkError is the reason that we failed the htlc.
+	LinkError *LinkError
+
+	// Incoming is true if the htlc was failed on an incoming link.
+	// If it failed on the outgoing link, it is false.
+	Incoming bool
+
+	// Timestamp is the time when the link failure occurred.
+	Timestamp time.Time
+}
+
+// ForwardingFailEvent represents a htlc failure which occurred down the line
+// after we forwarded a htlc onwards. An error is not included in this event
+// because errors returned down the route are encrypted. HtlcInfo is not
+// reliably available for forwarding failures, so it is omitted. These events
+// should be matched with their corresponding forward event to obtain this
+// information.
+type ForwardingFailEvent struct {
+	// HtlcKey uniquely identifies the htlc, and can be used to match the
+	// htlc with its corresponding forwarding event.
+	HtlcKey
+
+	// HtlcEventType classifies the event as part of a local send or
+	// receive, or as part of a forward.
+	HtlcEventType
+
+	// Timestamp is the time when the forwarding failure was received.
+	Timestamp time.Time
+}
+
+// SettleEvent represents a htlc that was settled. HtlcInfo is not reliably
+// available for forwarding failures, so it is omitted. These events should
+// be matched with corresponding forward events or invoices (for receives)
+// to obtain additional information about the htlc.
+type SettleEvent struct {
+	// HtlcKey uniquely identifies the htlc, and can be used to match
+	// forwards with their corresponding forwarding event.
+	HtlcKey
+
+	// HtlcEventType classifies the event as part of a local send or
+	// receive, or as part of a forward.
+	HtlcEventType
+
+	// Timestamp is the time when this htlc was settled.
+	Timestamp time.Time
+}
+
+// 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) {
+
+	event := &ForwardingEvent{
+		HtlcKey:       key,
+		HtlcInfo:      info,
+		HtlcEventType: eventType,
+		Timestamp:     h.now(),
+	}
+
+	log.Tracef("Notifying forward event: %v over %v, %v", eventType, key,
+		info)
+
+	if err := h.ntfnServer.SendUpdate(event); err != nil {
+		log.Warnf("Unable to send forwarding event: %v", err)
+	}
+}
+
+// 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) {
+
+	event := &LinkFailEvent{
+		HtlcKey:       key,
+		HtlcInfo:      info,
+		HtlcEventType: eventType,
+		LinkError:     linkErr,
+		Incoming:      incoming,
+		Timestamp:     h.now(),
+	}
+
+	log.Tracef("Notifying link failure event: %v over %v, %v", eventType,
+		key, info)
+
+	if err := h.ntfnServer.SendUpdate(event); err != nil {
+		log.Warnf("Unable to send link fail event: %v", err)
+	}
+}
+
+// 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) {
+
+	event := &ForwardingFailEvent{
+		HtlcKey:       key,
+		HtlcEventType: eventType,
+		Timestamp:     h.now(),
+	}
+
+	log.Tracef("Notifying forwarding failure event: %v over %v", eventType,
+		key)
+
+	if err := h.ntfnServer.SendUpdate(event); err != nil {
+		log.Warnf("Unable to send forwarding fail event: %v", err)
+	}
+}
+
+// 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,
+		HtlcEventType: eventType,
+		Timestamp:     h.now(),
+	}
+
+	log.Tracef("Notifying settle event: %v over %v", eventType, key)
+
+	if err := h.ntfnServer.SendUpdate(event); err != nil {
+		log.Warnf("Unable to send settle event: %v", err)
+	}
+}
+
+// newHtlc key returns a htlc key for the packet provided. If the packet
+// has a zero incoming channel ID, the packet is for one of our own sends,
+// which has the payment id stashed in the incoming htlc id. If this is the
+// case, we replace the incoming htlc id with zero so that the notifier
+// consistently reports zero circuit keys for events that terminate or
+// originate at our node.
+func newHtlcKey(pkt *htlcPacket) HtlcKey {
+	htlcKey := HtlcKey{
+		IncomingCircuit: channeldb.CircuitKey{
+			ChanID: pkt.incomingChanID,
+			HtlcID: pkt.incomingHTLCID,
+		},
+		OutgoingCircuit: CircuitKey{
+			ChanID: pkt.outgoingChanID,
+			HtlcID: pkt.outgoingHTLCID,
+		},
+	}
+
+	// If the packet has a zero incoming channel ID, it is a send that was
+	// initiated at our node. If this is the case, our internal pid is in
+	// the incoming htlc ID, so we overwrite it with 0 for notification
+	// purposes.
+	if pkt.incomingChanID == hop.Source {
+		htlcKey.IncomingCircuit.HtlcID = 0
+	}
+
+	return htlcKey
+}
+
+// newHtlcInfo returns HtlcInfo for the packet provided.
+func newHtlcInfo(pkt *htlcPacket) HtlcInfo {
+	return HtlcInfo{
+		IncomingTimeLock: pkt.incomingTimeout,
+		OutgoingTimeLock: pkt.outgoingTimeout,
+		IncomingAmt:      pkt.incomingAmount,
+		OutgoingAmt:      pkt.amount,
+	}
+}
+
+// getEventType returns the htlc type based on the fields set in the htlc
+// packet. Sends that originate at our node have the source (zero) incoming
+// channel ID. Receives to our node have the exit (zero) outgoing channel ID
+// and forwards have both fields set.
+func getEventType(pkt *htlcPacket) HtlcEventType {
+	switch {
+	case pkt.incomingChanID == hop.Source:
+		return HtlcEventTypeSend
+
+	case pkt.outgoingChanID == hop.Exit:
+		return HtlcEventTypeReceive
+
+	default:
+		return HtlcEventTypeForward
+	}
+}

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
@@ -1204,10 +1208,7 @@ func (l *channelLink) processHtlcResolution(resolution invoices.HtlcResolution,
 		l.log.Debugf("received settle resolution for %v"+
 			"with outcome: %v", circuitKey, res.Outcome)
 
-		return l.settleHTLC(
+		return l.settleHTLC(res.Preimage, htlc.pd)
-			res.Preimage, htlc.pd.HtlcIndex,
-			htlc.pd.SourceRef,
-		)
 
 	// For htlc failures, we get the relevant failure message based
 	// on the failure resolution and then fail the htlc.
@@ -1220,8 +1221,7 @@ func (l *channelLink) processHtlcResolution(resolution invoices.HtlcResolution,
 		failure := getResolutionFailure(res, htlc.pd.Amount)
 
 		l.sendHTLCError(
-			htlc.pd.HtlcIndex, failure,
+			htlc.pd, failure, htlc.obfuscator, true,
-			htlc.obfuscator, htlc.pd.SourceRef,
 		)
 		return nil
 
@@ -1414,6 +1414,18 @@ func (l *channelLink) handleDownStreamPkt(pkt *htlcPacket, isReProcess bool) {
 
 		l.cfg.Peer.SendMessage(false, htlc)
 
+		// Send a forward event notification to htlcNotifier.
+		l.cfg.HtlcNotifier.NotifyForwardingEvent(
+			newHtlcKey(pkt),
+			HtlcInfo{
+				IncomingTimeLock: pkt.incomingTimeout,
+				IncomingAmt:      pkt.incomingAmount,
+				OutgoingTimeLock: htlc.Expiry,
+				OutgoingAmt:      htlc.Amount,
+			},
+			getEventType(pkt),
+		)
+
 	case *lnwire.UpdateFulfillHTLC:
 		// If hodl.SettleOutgoing mode is active, we exit early to
 		// simulate arbitrary delays between the switch adding the
@@ -1472,6 +1484,12 @@ func (l *channelLink) handleDownStreamPkt(pkt *htlcPacket, isReProcess bool) {
 		l.cfg.Peer.SendMessage(false, htlc)
 		isSettle = true
 
+		// Send a settle event notification to htlcNotifier.
+		l.cfg.HtlcNotifier.NotifySettleEvent(
+			newHtlcKey(pkt),
+			getEventType(pkt),
+		)
+
 	case *lnwire.UpdateFailHTLC:
 		// If hodl.FailOutgoing mode is active, we exit early to
 		// simulate arbitrary delays between the switch adding a FAIL to
@@ -1525,10 +1543,28 @@ func (l *channelLink) handleDownStreamPkt(pkt *htlcPacket, isReProcess bool) {
 		htlc.ChanID = l.ChanID()
 		htlc.ID = pkt.incomingHTLCID
 
-		// Finally, we send the HTLC message to the peer which
+		// We send the HTLC message to the peer which initially created
-		// initially created the HTLC.
+		// the HTLC.
 		l.cfg.Peer.SendMessage(false, htlc)
 		isSettle = true
+
+		// If the packet does not have a link failure set, it failed
+		// further down the route so we notify a forwarding failure.
+		// Otherwise, we notify a link failure because it failed at our
+		// node.
+		if pkt.linkFailure != nil {
+			l.cfg.HtlcNotifier.NotifyLinkFailEvent(
+				newHtlcKey(pkt),
+				newHtlcInfo(pkt),
+				getEventType(pkt),
+				pkt.linkFailure,
+				false,
+			)
+		} else {
+			l.cfg.HtlcNotifier.NotifyForwardingFailEvent(
+				newHtlcKey(pkt), getEventType(pkt),
+			)
+		}
 	}
 
 	// If this newly added update exceeds the min batch size for adds, or
@@ -2656,9 +2692,7 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg,
 			// later date
 			failure := lnwire.NewInvalidOnionPayload(failedType, 0)
 			l.sendHTLCError(
-				pd.HtlcIndex,
+				pd, NewLinkError(failure), obfuscator, false,
-				NewLinkError(failure),
-				obfuscator, pd.SourceRef,
 			)
 
 			l.log.Errorf("unable to decode forwarding "+
@@ -2771,10 +2805,7 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg,
 				)
 
 				l.sendHTLCError(
-					pd.HtlcIndex,
+					pd, NewLinkError(failure), obfuscator, false,
-					NewLinkError(failure),
-					obfuscator,
-					pd.SourceRef,
 				)
 				continue
 			}
@@ -2860,7 +2891,7 @@ func (l *channelLink) processExitHop(pd *lnwallet.PaymentDescriptor,
 		failure := NewLinkError(
 			lnwire.NewFinalIncorrectHtlcAmount(pd.Amount),
 		)
-		l.sendHTLCError(pd.HtlcIndex, failure, obfuscator, pd.SourceRef)
+		l.sendHTLCError(pd, failure, obfuscator, true)
 
 		return nil
 	}
@@ -2875,7 +2906,7 @@ func (l *channelLink) processExitHop(pd *lnwallet.PaymentDescriptor,
 		failure := NewLinkError(
 			lnwire.NewFinalIncorrectCltvExpiry(pd.Timeout),
 		)
-		l.sendHTLCError(pd.HtlcIndex, failure, obfuscator, pd.SourceRef)
+		l.sendHTLCError(pd, failure, obfuscator, true)
 
 		return nil
 	}
@@ -2916,15 +2947,15 @@ func (l *channelLink) processExitHop(pd *lnwallet.PaymentDescriptor,
 }
 
 // settleHTLC settles the HTLC on the channel.
-func (l *channelLink) settleHTLC(preimage lntypes.Preimage, htlcIndex uint64,
+func (l *channelLink) settleHTLC(preimage lntypes.Preimage,
-	sourceRef *channeldb.AddRef) error {
+	pd *lnwallet.PaymentDescriptor) error {
 
 	hash := preimage.Hash()
 
 	l.log.Infof("settling htlc %v as exit hop", hash)
 
 	err := l.channel.SettleHTLC(
-		preimage, htlcIndex, sourceRef, nil, nil,
+		preimage, pd.HtlcIndex, pd.SourceRef, nil, nil,
 	)
 	if err != nil {
 		return fmt.Errorf("unable to settle htlc: %v", err)
@@ -2942,10 +2973,21 @@ func (l *channelLink) settleHTLC(preimage lntypes.Preimage, htlcIndex uint64,
 	// remote peer.
 	l.cfg.Peer.SendMessage(false, &lnwire.UpdateFulfillHTLC{
 		ChanID:          l.ChanID(),
-		ID:              htlcIndex,
+		ID:              pd.HtlcIndex,
 		PaymentPreimage: preimage,
 	})
 
+	// Once we have successfully settled the htlc, notify a settle event.
+	l.cfg.HtlcNotifier.NotifySettleEvent(
+		HtlcKey{
+			IncomingCircuit: channeldb.CircuitKey{
+				ChanID: l.ShortChanID(),
+				HtlcID: pd.HtlcIndex,
+			},
+		},
+		HtlcEventTypeReceive,
+	)
+
 	return nil
 }
 
@@ -2991,8 +3033,8 @@ func (l *channelLink) handleBatchFwdErrs(errChan chan error) {
 
 // sendHTLCError functions cancels HTLC and send cancel message back to the
 // peer from which HTLC was received.
-func (l *channelLink) sendHTLCError(htlcIndex uint64, failure *LinkError,
+func (l *channelLink) sendHTLCError(pd *lnwallet.PaymentDescriptor,
-	e hop.ErrorEncrypter, sourceRef *channeldb.AddRef) {
+	failure *LinkError, e hop.ErrorEncrypter, isReceive bool) {
 
 	reason, err := e.EncryptFirstHop(failure.WireMessage())
 	if err != nil {
@@ -3000,7 +3042,7 @@ func (l *channelLink) sendHTLCError(htlcIndex uint64, failure *LinkError,
 		return
 	}
 
-	err = l.channel.FailHTLC(htlcIndex, reason, sourceRef, nil, nil)
+	err = l.channel.FailHTLC(pd.HtlcIndex, reason, pd.SourceRef, nil, nil)
 	if err != nil {
 		l.log.Errorf("unable cancel htlc: %v", err)
 		return
@@ -3008,9 +3050,35 @@ func (l *channelLink) sendHTLCError(htlcIndex uint64, failure *LinkError,
 
 	l.cfg.Peer.SendMessage(false, &lnwire.UpdateFailHTLC{
 		ChanID: l.ChanID(),
-		ID:     htlcIndex,
+		ID:     pd.HtlcIndex,
 		Reason: reason,
 	})
+
+	// Notify a link failure on our incoming link. Outgoing htlc information
+	// is not available at this point, because we have not decrypted the
+	// onion, so it is excluded.
+	var eventType HtlcEventType
+	if isReceive {
+		eventType = HtlcEventTypeReceive
+	} else {
+		eventType = HtlcEventTypeForward
+	}
+
+	l.cfg.HtlcNotifier.NotifyLinkFailEvent(
+		HtlcKey{
+			IncomingCircuit: channeldb.CircuitKey{
+				ChanID: l.ShortChanID(),
+				HtlcID: pd.HtlcIndex,
+			},
+		},
+		HtlcInfo{
+			IncomingTimeLock: pd.Timeout,
+			IncomingAmt:      pd.Amount,
+		},
+		eventType,
+		failure,
+		true,
+	)
 }
 
 // sendMalformedHTLCError helper function which sends the malformed HTLC 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
@@ -764,38 +768,23 @@ func (s *Switch) handleLocalDispatch(pkt *htlcPacket) error {
 	// User have created the htlc update therefore we should find the
 	// appropriate channel link and send the payment over this link.
 	if htlc, ok := pkt.htlc.(*lnwire.UpdateAddHTLC); ok {
-		// Try to find links by node destination.
+		link, err := s.handleLocalAddHTLC(pkt, htlc)
-		s.indexMtx.RLock()
-		link, err := s.getLinkByShortID(pkt.outgoingChanID)
-		s.indexMtx.RUnlock()
 		if err != nil {
-			log.Errorf("Link %v not found", pkt.outgoingChanID)
+			// Notify the htlc notifier of a link failure on our
-			return NewLinkError(&lnwire.FailUnknownNextPeer{})
+			// outgoing link. Incoming timelock/amount values are
-		}
+			// not set because they are not present for local sends.
-
+			s.cfg.HtlcNotifier.NotifyLinkFailEvent(
-		if !link.EligibleToForward() {
+				newHtlcKey(pkt),
-			log.Errorf("Link %v is not available to forward",
+				HtlcInfo{
-				pkt.outgoingChanID)
+					OutgoingTimeLock: htlc.Expiry,
-
+					OutgoingAmt:      htlc.Amount,
-			// The update does not need to be populated as the error
+				},
-			// will be returned back to the router.
+				HtlcEventTypeSend,
-			return NewDetailedLinkError(
+				err,
-				lnwire.NewTemporaryChannelFailure(nil),
+				false,
-				OutgoingFailureLinkNotEligible,
 			)
-		}
 
-		// Ensure that the htlc satisfies the outgoing channel policy.
+			return err
-		currentHeight := atomic.LoadUint32(&s.bestHeight)
-		htlcErr := link.CheckHtlcTransit(
-			htlc.PaymentHash,
-			htlc.Amount,
-			htlc.Expiry, currentHeight,
-		)
-		if htlcErr != nil {
-			log.Errorf("Link %v policy for local forward not "+
-				"satisfied", pkt.outgoingChanID)
-			return htlcErr
 		}
 
 		return link.HandleSwitchPacket(pkt)
@@ -807,6 +796,47 @@ func (s *Switch) handleLocalDispatch(pkt *htlcPacket) error {
 	return nil
 }
 
+// handleLocalAddHTLC handles the addition of a htlc for a send that
+// originates from our node. It returns the link that the htlc should
+// be forwarded outwards on, and a link error if the htlc cannot be
+// forwarded.
+func (s *Switch) handleLocalAddHTLC(pkt *htlcPacket,
+	htlc *lnwire.UpdateAddHTLC) (ChannelLink, *LinkError) {
+
+	// Try to find links by node destination.
+	s.indexMtx.RLock()
+	link, err := s.getLinkByShortID(pkt.outgoingChanID)
+	s.indexMtx.RUnlock()
+	if err != nil {
+		log.Errorf("Link %v not found", pkt.outgoingChanID)
+		return nil, NewLinkError(&lnwire.FailUnknownNextPeer{})
+	}
+
+	if !link.EligibleToForward() {
+		log.Errorf("Link %v is not available to forward",
+			pkt.outgoingChanID)
+
+		// The update does not need to be populated as the error
+		// will be returned back to the router.
+		return nil, NewDetailedLinkError(
+			lnwire.NewTemporaryChannelFailure(nil),
+			OutgoingFailureLinkNotEligible,
+		)
+	}
+
+	// Ensure that the htlc satisfies the outgoing channel policy.
+	currentHeight := atomic.LoadUint32(&s.bestHeight)
+	htlcErr := link.CheckHtlcTransit(
+		htlc.PaymentHash, htlc.Amount, htlc.Expiry, currentHeight,
+	)
+	if htlcErr != nil {
+		log.Errorf("Link %v policy for local forward not "+
+			"satisfied", pkt.outgoingChanID)
+		return nil, htlcErr
+	}
+	return link, nil
+}
+
 // handleLocalResponse processes a Settle or Fail responding to a
 // locally-initiated payment. This is handled asynchronously to avoid blocking
 // the main event loop within the switch, as these operations can require
@@ -868,6 +898,18 @@ func (s *Switch) handleLocalResponse(pkt *htlcPacket) {
 			pkt.inKey(), err)
 		return
 	}
+
+	// Finally, notify on the htlc failure or success that has been handled.
+	key := newHtlcKey(pkt)
+	eventType := getEventType(pkt)
+
+	switch pkt.htlc.(type) {
+	case *lnwire.UpdateFulfillHTLC:
+		s.cfg.HtlcNotifier.NotifySettleEvent(key, eventType)
+
+	case *lnwire.UpdateFailHTLC:
+		s.cfg.HtlcNotifier.NotifyForwardingFailEvent(key, eventType)
+	}
 }
 
 // extractResult uses the given deobfuscator to extract the payment result from
@@ -1252,10 +1294,19 @@ func (s *Switch) failAddPacket(packet *htlcPacket, failure *LinkError) error {
 
 	log.Error(failure.Error())
 
+	// Create a failure packet for this htlc. The the full set of
+	// information about the htlc failure is included so that they can
+	// be included in link failure notifications.
 	failPkt := &htlcPacket{
 		sourceRef:       packet.sourceRef,
 		incomingChanID:  packet.incomingChanID,
 		incomingHTLCID:  packet.incomingHTLCID,
+		outgoingChanID:  packet.outgoingChanID,
+		outgoingHTLCID:  packet.outgoingHTLCID,
+		incomingAmount:  packet.incomingAmount,
+		amount:          packet.amount,
+		incomingTimeout: packet.incomingTimeout,
+		outgoingTimeout: packet.outgoingTimeout,
 		circuit:         packet.circuit,
 		linkFailure:     failure,
 		htlc: &lnwire.UpdateFailHTLC{

htlcswitch/switch_test.go

@@ -14,11 +14,14 @@ import (
 	"github.com/btcsuite/fastsha256"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/htlcswitch/hop"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/ticker"
 )
 
+var zeroCircuit = channeldb.CircuitKey{}
+
 func genPreimage() ([32]byte, error) {
 	var preimage [32]byte
 	if _, err := io.ReadFull(rand.Reader, preimage[:]); err != nil {
@@ -2697,3 +2700,361 @@ func TestInvalidFailure(t *testing.T) {
 		t.Fatal("err wasn't received")
 	}
 }
+
+// htlcNotifierEvents is a function that generates a set of expected htlc
+// notifier evetns for each node in a three hop network with the dynamic
+// values provided. These functions take dynamic values so that changes to
+// external systems (such as our default timelock delta) do not break
+// these tests.
+type htlcNotifierEvents func(channels *clusterChannels, htlcID uint64,
+	ts time.Time, htlc *lnwire.UpdateAddHTLC,
+	hops []*hop.Payload) ([]interface{}, []interface{}, []interface{})
+
+// TestHtlcNotifier tests the notifying of htlc events that are routed over a
+// three hop network. It sets up an Alice -> Bob -> Carol network and routes
+// payments from Alice -> Carol to test events from the perspective of a
+// sending (Alice), forwarding (Bob) and receiving (Carol) node. Test cases
+// are present for saduccessful and failed payments.
+func TestHtlcNotifier(t *testing.T) {
+	tests := []struct {
+		name string
+
+		// Options is a set of options to apply to the three hop
+		// network's servers.
+		options []serverOption
+
+		// expectedEvents is a function which returns an expected set
+		// of events for the test.
+		expectedEvents htlcNotifierEvents
+
+		// iterations is the number of times we will send a payment,
+		// this is used to send more than one payment to force non-
+		// zero htlc indexes to make sure we aren't just checking
+		// default values.
+		iterations int
+	}{
+		{
+			name:    "successful three hop payment",
+			options: nil,
+			expectedEvents: func(channels *clusterChannels,
+				htlcID uint64, ts time.Time,
+				htlc *lnwire.UpdateAddHTLC,
+				hops []*hop.Payload) ([]interface{},
+				[]interface{}, []interface{}) {
+
+				return getThreeHopEvents(
+					channels, htlcID, ts, htlc, hops, nil,
+				)
+			},
+			iterations: 2,
+		},
+		{
+			name: "failed at forwarding link",
+			// Set a functional option which disables bob as a
+			// forwarding node to force a payment error.
+			options: []serverOption{
+				serverOptionRejectHtlc(false, true, false),
+			},
+			expectedEvents: func(channels *clusterChannels,
+				htlcID uint64, ts time.Time,
+				htlc *lnwire.UpdateAddHTLC,
+				hops []*hop.Payload) ([]interface{},
+				[]interface{}, []interface{}) {
+
+				return getThreeHopEvents(
+					channels, htlcID, ts, htlc, hops,
+					&LinkError{
+						msg:           &lnwire.FailChannelDisabled{},
+						FailureDetail: OutgoingFailureForwardsDisabled,
+					},
+				)
+			},
+			iterations: 1,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			testHtcNotifier(
+				t, test.options, test.iterations,
+				test.expectedEvents,
+			)
+		})
+	}
+}
+
+// testHtcNotifier runs a htlc notifier test.
+func testHtcNotifier(t *testing.T, testOpts []serverOption, iterations int,
+	getEvents htlcNotifierEvents) {
+
+	t.Parallel()
+
+	// First, we'll create our traditional three hop
+	// network.
+	channels, cleanUp, _, err := createClusterChannels(
+		btcutil.SatoshiPerBitcoin*3,
+		btcutil.SatoshiPerBitcoin*5)
+	if err != nil {
+		t.Fatalf("unable to create channel: %v", err)
+	}
+	defer cleanUp()
+
+	// Mock time so that all events are reported with a static timestamp.
+	now := time.Now()
+	mockTime := func() time.Time {
+		return now
+	}
+
+	// Create htlc notifiers for each server in the three hop network and
+	// start them.
+	aliceNotifier := NewHtlcNotifier(mockTime)
+	if err := aliceNotifier.Start(); err != nil {
+		t.Fatalf("could not start alice notifier")
+	}
+	defer aliceNotifier.Stop()
+
+	bobNotifier := NewHtlcNotifier(mockTime)
+	if err := bobNotifier.Start(); err != nil {
+		t.Fatalf("could not start bob notifier")
+	}
+	defer bobNotifier.Stop()
+
+	carolNotifier := NewHtlcNotifier(mockTime)
+	if err := carolNotifier.Start(); err != nil {
+		t.Fatalf("could not start carol notifier")
+	}
+	defer carolNotifier.Stop()
+
+	// Create a notifier server option which will set our htlc notifiers
+	// for the three hop network.
+	notifierOption := serverOptionWithHtlcNotifier(
+		aliceNotifier, bobNotifier, carolNotifier,
+	)
+
+	// Add the htlcNotifier option to any other options
+	// set in the test.
+	options := append(testOpts, notifierOption)
+
+	n := newThreeHopNetwork(
+		t, channels.aliceToBob,
+		channels.bobToAlice, channels.bobToCarol,
+		channels.carolToBob, testStartingHeight,
+		options...,
+	)
+	if err := n.start(); err != nil {
+		t.Fatalf("unable to start three hop "+
+			"network: %v", err)
+	}
+	defer n.stop()
+
+	// Before we forward anything, subscribe to htlc events
+	// from each notifier.
+	aliceEvents, err := aliceNotifier.SubscribeHtlcEvents()
+	if err != nil {
+		t.Fatalf("could not subscribe to alice's"+
+			" events: %v", err)
+	}
+	defer aliceEvents.Cancel()
+
+	bobEvents, err := bobNotifier.SubscribeHtlcEvents()
+	if err != nil {
+		t.Fatalf("could not subscribe to bob's"+
+			" events: %v", err)
+	}
+	defer bobEvents.Cancel()
+
+	carolEvents, err := carolNotifier.SubscribeHtlcEvents()
+	if err != nil {
+		t.Fatalf("could not subscribe to carol's"+
+			" events: %v", err)
+	}
+	defer carolEvents.Cancel()
+
+	// Send multiple payments, as specified by the test to test incrementing
+	// of htlc ids.
+	for i := 0; i < iterations; i++ {
+		// We'll start off by making a payment from
+		// Alice -> Bob -> Carol.
+		htlc, hops := n.sendThreeHopPayment(t)
+
+		alice, bob, carol := getEvents(
+			channels, uint64(i), now, htlc, hops,
+		)
+
+		checkHtlcEvents(t, aliceEvents.Updates(), alice)
+		checkHtlcEvents(t, bobEvents.Updates(), bob)
+		checkHtlcEvents(t, carolEvents.Updates(), carol)
+
+	}
+}
+
+// checkHtlcEvents checks that a subscription has the set of htlc events
+// we expect it to have.
+func checkHtlcEvents(t *testing.T, events <-chan interface{},
+	expectedEvents []interface{}) {
+
+	for _, expected := range expectedEvents {
+		select {
+		case event := <-events:
+			if !reflect.DeepEqual(event, expected) {
+				t.Fatalf("expected %v, got: %v", expected,
+					event)
+			}
+
+		case <-time.After(time.Second):
+			t.Fatalf("expected event: %v", expected)
+		}
+	}
+}
+
+// sendThreeHopPayment is a helper function which sends a payment over
+// Alice -> Bob -> Carol in a three hop network and returns Alice's first htlc
+// and the remainder of the hops.
+func (n *threeHopNetwork) sendThreeHopPayment(t *testing.T) (*lnwire.UpdateAddHTLC,
+	[]*hop.Payload) {
+
+	amount := lnwire.NewMSatFromSatoshis(btcutil.SatoshiPerBitcoin)
+
+	htlcAmt, totalTimelock, hops := generateHops(amount, testStartingHeight,
+		n.firstBobChannelLink, n.carolChannelLink)
+	blob, err := generateRoute(hops...)
+	if err != nil {
+		t.Fatal(err)
+	}
+	invoice, htlc, pid, err := generatePayment(
+		amount, htlcAmt, totalTimelock, blob,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = n.carolServer.registry.AddInvoice(*invoice, htlc.PaymentHash)
+	if err != nil {
+		t.Fatalf("unable to add invoice in carol registry: %v", err)
+	}
+
+	if err := n.aliceServer.htlcSwitch.SendHTLC(
+		n.firstBobChannelLink.ShortChanID(), pid, htlc,
+	); err != nil {
+		t.Fatalf("could not send htlc")
+	}
+
+	return htlc, hops
+}
+
+// getThreeHopEvents gets the set of htlc events that we expect for a payment
+// from Alice -> Bob -> Carol. If a non-nil link error is provided, the set
+// of events will fail on Bob's outgoing link.
+func getThreeHopEvents(channels *clusterChannels, htlcID uint64,
+	ts time.Time, htlc *lnwire.UpdateAddHTLC, hops []*hop.Payload,
+	linkError *LinkError) ([]interface{}, []interface{}, []interface{}) {
+
+	aliceKey := HtlcKey{
+		IncomingCircuit: zeroCircuit,
+		OutgoingCircuit: channeldb.CircuitKey{
+			ChanID: channels.aliceToBob.ShortChanID(),
+			HtlcID: htlcID,
+		},
+	}
+
+	// Alice always needs a forwarding event because she initiates the
+	// send.
+	aliceEvents := []interface{}{
+		&ForwardingEvent{
+			HtlcKey: aliceKey,
+			HtlcInfo: HtlcInfo{
+				OutgoingTimeLock: htlc.Expiry,
+				OutgoingAmt:      htlc.Amount,
+			},
+			HtlcEventType: HtlcEventTypeSend,
+			Timestamp:     ts,
+		},
+	}
+
+	bobKey := HtlcKey{
+		IncomingCircuit: channeldb.CircuitKey{
+			ChanID: channels.bobToAlice.ShortChanID(),
+			HtlcID: htlcID,
+		},
+		OutgoingCircuit: channeldb.CircuitKey{
+			ChanID: channels.bobToCarol.ShortChanID(),
+			HtlcID: htlcID,
+		},
+	}
+
+	bobInfo := HtlcInfo{
+		IncomingTimeLock: htlc.Expiry,
+		IncomingAmt:      htlc.Amount,
+		OutgoingTimeLock: hops[1].FwdInfo.OutgoingCTLV,
+		OutgoingAmt:      hops[1].FwdInfo.AmountToForward,
+	}
+
+	// If we expect the payment to fail, we add failures for alice and
+	// bob, and no events for carol because the payment never reaches her.
+	if linkError != nil {
+		aliceEvents = append(aliceEvents,
+			&ForwardingFailEvent{
+				HtlcKey:       aliceKey,
+				HtlcEventType: HtlcEventTypeSend,
+				Timestamp:     ts,
+			},
+		)
+
+		bobEvents := []interface{}{
+			&LinkFailEvent{
+				HtlcKey:       bobKey,
+				HtlcInfo:      bobInfo,
+				HtlcEventType: HtlcEventTypeForward,
+				LinkError:     linkError,
+				Incoming:      false,
+				Timestamp:     ts,
+			},
+		}
+
+		return aliceEvents, bobEvents, nil
+	}
+
+	// If we want to get events for a successful payment, we add a settle
+	// for alice, a forward and settle for bob and a receive settle for
+	// carol.
+	aliceEvents = append(
+		aliceEvents,
+		&SettleEvent{
+			HtlcKey:       aliceKey,
+			HtlcEventType: HtlcEventTypeSend,
+			Timestamp:     ts,
+		},
+	)
+
+	bobEvents := []interface{}{
+		&ForwardingEvent{
+			HtlcKey:       bobKey,
+			HtlcInfo:      bobInfo,
+			HtlcEventType: HtlcEventTypeForward,
+			Timestamp:     ts,
+		},
+		&SettleEvent{
+			HtlcKey:       bobKey,
+			HtlcEventType: HtlcEventTypeForward,
+			Timestamp:     ts,
+		},
+	}
+
+	carolEvents := []interface{}{
+		&SettleEvent{
+			HtlcKey: HtlcKey{
+				IncomingCircuit: channeldb.CircuitKey{
+					ChanID: channels.carolToBob.ShortChanID(),
+					HtlcID: htlcID,
+				},
+				OutgoingCircuit: zeroCircuit,
+			},
+			HtlcEventType: HtlcEventTypeReceive,
+			Timestamp:     ts,
+		},
+	}
+
+	return aliceEvents, bobEvents, carolEvents
+}

htlcswitch/test_utils.go

@@ -966,9 +966,11 @@ func createClusterChannels(aliceToBob, bobToCarol btcutil.Amount) (
 // alice                   first bob    second bob              carol
 // channel link	    	  channel link   channel link		channel link
 //
+// This function takes server options which can be used to apply custom
+// settings to alice, bob and carol.
 func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 	secondBobChannel, carolChannel *lnwallet.LightningChannel,
-	startingHeight uint32) *threeHopNetwork {
+	startingHeight uint32, opts ...serverOption) *threeHopNetwork {
 
 	aliceDb := aliceChannel.State().Db
 	bobDb := firstBobChannel.State().Db
@@ -996,6 +998,12 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 		t.Fatalf("unable to create carol server: %v", err)
 	}
 
+	// Apply all additional functional options to the servers before
+	// creating any links.
+	for _, option := range opts {
+		option(aliceServer, bobServer, carolServer)
+	}
+
 	// Create mock decoder instead of sphinx one in order to mock the route
 	// which htlc should follow.
 	aliceDecoder := newMockIteratorDecoder()
@@ -1045,6 +1053,34 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 	}
 }
 
+// serverOption is a function which alters the three servers created for
+// a three hop network to allow custom settings on each server.
+type serverOption func(aliceServer, bobServer, carolServer *mockServer)
+
+// serverOptionWithHtlcNotifier is a functional option for the creation of
+// three hop network servers which allows setting of htlc notifiers.
+// Note that these notifiers should be started and stopped by the calling
+// function.
+func serverOptionWithHtlcNotifier(alice, bob,
+	carol *HtlcNotifier) serverOption {
+
+	return func(aliceServer, bobServer, carolServer *mockServer) {
+		aliceServer.htlcSwitch.cfg.HtlcNotifier = alice
+		bobServer.htlcSwitch.cfg.HtlcNotifier = bob
+		carolServer.htlcSwitch.cfg.HtlcNotifier = carol
+	}
+}
+
+// serverOptionRejectHtlc is the functional option for setting the reject
+// htlc config option in each server's switch.
+func serverOptionRejectHtlc(alice, bob, carol bool) serverOption {
+	return func(aliceServer, bobServer, carolServer *mockServer) {
+		aliceServer.htlcSwitch.cfg.RejectHTLC = alice
+		bobServer.htlcSwitch.cfg.RejectHTLC = bob
+		carolServer.htlcSwitch.cfg.RejectHTLC = carol
+	}
+}
+
 // createTwoClusterChannels creates lightning channels which are needed for
 // a 2 hop network cluster to be initialized.
 func createTwoClusterChannels(aliceToBob, bobToCarol btcutil.Amount) (
@@ -1139,6 +1175,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()