Merge pull request #3781 from carlaKC/3420-htlcForwarding

htlcswitch: add htlcNotifier

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(
-			res.Preimage, htlc.pd.HtlcIndex,
-			htlc.pd.SourceRef,
-		)
+		return l.settleHTLC(res.Preimage, htlc.pd)
 
 	// 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.obfuscator, htlc.pd.SourceRef,
+			htlc.pd, failure, htlc.obfuscator, true,
 		)
 		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
-		// 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
@@ -2656,9 +2692,7 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg,
 			// later date
 			failure := lnwire.NewInvalidOnionPayload(failedType, 0)
 			l.sendHTLCError(
-				pd.HtlcIndex,
-				NewLinkError(failure),
-				obfuscator, pd.SourceRef,
+				pd, NewLinkError(failure), obfuscator, false,
 			)
 
 			l.log.Errorf("unable to decode forwarding "+
@@ -2771,10 +2805,7 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg,
 				)
 
 				l.sendHTLCError(
-					pd.HtlcIndex,
-					NewLinkError(failure),
-					obfuscator,
-					pd.SourceRef,
+					pd, NewLinkError(failure), obfuscator, false,
 				)
 				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,
-	sourceRef *channeldb.AddRef) error {
+func (l *channelLink) settleHTLC(preimage lntypes.Preimage,
+	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,26 +3033,52 @@ 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,
-	e hop.ErrorEncrypter, sourceRef *channeldb.AddRef) {
+func (l *channelLink) sendHTLCError(pd *lnwallet.PaymentDescriptor,
+	failure *LinkError, e hop.ErrorEncrypter, isReceive bool) {
 
 	reason, err := e.EncryptFirstHop(failure.WireMessage())
 	if err != nil {
 		l.log.Errorf("unable to obfuscate error: %v", err)
 		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
 	}
 
 	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) {
+}