htlcswitch: add notifications for forwards

This commit adds notifications for htlcs which are forwarded through
our node. Forwards are notified when the htlc is added on our ougoing
link, settles when we send a settle message to the downstream peer.
If a failure occurs, we check whether it occurred at our node, then
notify a link or forwarding failure accordingly.

Note that this change also adds forward event notifications for sends
which are initiated by our node because the handling code for adding
a htlc which originates from our node is the same as that for handling
forwards. Htlcs for our locally initiated sends have our internal pid
set in the incoming htlcs id field, so we extract this value and notify
with a zero htlc id to be consistent with receives (which have zero
outgoing circuits). Subsequent settles or failures are not noitfied
for local sends in this commit, and will be handled in a follow up.

htlcswitch/htlcnotifier.go

@@ -371,3 +371,59 @@ func (h *HtlcNotifier) NotifySettleEvent(key HtlcKey, eventType HtlcEventType) {
 		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/link.go

@@ -1418,6 +1418,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
@@ -1476,6 +1488,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
@@ -1529,10 +1547,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
-		// initially created the HTLC.
+		// We send the HTLC message to the peer which initially created
+		// 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