htlcswitch: add final htlc event to notifier

contractcourt/chain_arbitrator.go

@@ -192,6 +192,9 @@ type ChainArbitratorConfig struct {
 	// database.
 	PutFinalHtlcOutcome func(chanId lnwire.ShortChannelID,
 		htlcId uint64, settled bool) error
+
+	// HtlcNotifier is an interface that htlc events are sent to.
+	HtlcNotifier HtlcNotifier
 }
 
 // ChainArbitrator is a sub-system that oversees the on-chain resolution of all

contractcourt/channel_arbitrator.go

@@ -2243,6 +2243,15 @@ func (c *ChannelArbitrator) prepContractResolutions(
 				if err != nil {
 					return nil, nil, err
 				}
+
+				// Send notification.
+				chainArbCfg.HtlcNotifier.NotifyFinalHtlcEvent(
+					key,
+					channeldb.FinalHtlcInfo{
+						Settled:  false,
+						Offchain: false,
+					},
+				)
 			}
 
 		// Finally, if this is an outgoing HTLC we've sent, then we'll

contractcourt/channel_arbitrator_test.go

@@ -361,8 +361,9 @@ func createTestChannelArbitrator(t *testing.T, log ArbitratorLog,
 			chanArbCtx.breachSubscribed <- struct{}{}
 			return false, nil
 		},
-		Clock:   clock.NewDefaultClock(),
-		Sweeper: mockSweeper,
+		Clock:        clock.NewDefaultClock(),
+		Sweeper:      mockSweeper,
+		HtlcNotifier: &mockHTLCNotifier{},
 		PutFinalHtlcOutcome: func(chanId lnwire.ShortChannelID,
 			htlcId uint64, settled bool) error {
 

contractcourt/htlc_incoming_contest_resolver.go

@@ -59,6 +59,18 @@ func (h *htlcIncomingContestResolver) processFinalHtlcFail() error {
 		return err
 	}
 
+	// Send notification.
+	h.ChainArbitratorConfig.HtlcNotifier.NotifyFinalHtlcEvent(
+		channeldb.CircuitKey{
+			ChanID: h.ShortChanID,
+			HtlcID: h.htlc.HtlcIndex,
+		},
+		channeldb.FinalHtlcInfo{
+			Settled:  false,
+			Offchain: false,
+		},
+	)
+
 	return nil
 }
 

contractcourt/htlc_incoming_resolver_test.go

@@ -333,6 +333,8 @@ func newIncomingResolverTestContext(t *testing.T, isExit bool) *incomingResolver
 		t:              t,
 	}
 
+	htlcNotifier := &mockHTLCNotifier{}
+
 	chainCfg := ChannelArbitratorConfig{
 		ChainArbitratorConfig: ChainArbitratorConfig{
 			Notifier:       notifier,
@@ -346,6 +348,7 @@ func newIncomingResolverTestContext(t *testing.T, isExit bool) *incomingResolver
 
 				return nil
 			},
+			HtlcNotifier: htlcNotifier,
 		},
 		PutResolverReport: func(_ kvdb.RwTx,
 			_ *channeldb.ResolverReport) error {

contractcourt/htlc_success_resolver.go

@@ -477,6 +477,18 @@ func (h *htlcSuccessResolver) checkpointClaim(spendTx *chainhash.Hash,
 		return err
 	}
 
+	// Send notification.
+	h.ChainArbitratorConfig.HtlcNotifier.NotifyFinalHtlcEvent(
+		channeldb.CircuitKey{
+			ChanID: h.ShortChanID,
+			HtlcID: h.htlc.HtlcIndex,
+		},
+		channeldb.FinalHtlcInfo{
+			Settled:  true,
+			Offchain: false,
+		},
+	)
+
 	// Create a resolver report for claiming of the htlc itself.
 	amt := btcutil.Amount(h.htlcResolution.SweepSignDesc.Output.Value)
 	reports := []*channeldb.ResolverReport{

contractcourt/htlc_success_resolver_test.go

@@ -53,6 +53,8 @@ func newHtlcResolverTestContext(t *testing.T,
 		t:              t,
 	}
 
+	htlcNotifier := &mockHTLCNotifier{}
+
 	witnessBeacon := newMockWitnessBeacon()
 	chainCfg := ChannelArbitratorConfig{
 		ChainArbitratorConfig: ChainArbitratorConfig{
@@ -84,6 +86,7 @@ func newHtlcResolverTestContext(t *testing.T,
 
 				return nil
 			},
+			HtlcNotifier: htlcNotifier,
 		},
 		PutResolverReport: func(_ kvdb.RwTx,
 			report *channeldb.ResolverReport) error {

contractcourt/interfaces.go

@@ -65,3 +65,11 @@ type UtxoSweeper interface {
 	UpdateParams(input wire.OutPoint, params sweep.ParamsUpdate) (
 		chan sweep.Result, error)
 }
+
+// HtlcNotifier defines the notification functions that contract court requires.
+type HtlcNotifier interface {
+	// NotifyFinalHtlcEvent notifies the HtlcNotifier that the final outcome
+	// for an htlc has been determined.
+	NotifyFinalHtlcEvent(key channeldb.CircuitKey,
+		info channeldb.FinalHtlcInfo)
+}

contractcourt/mock_htlcnotifier_test.go

@@ -0,0 +1,11 @@
+package contractcourt
+
+import "github.com/lightningnetwork/lnd/channeldb"
+
+type mockHTLCNotifier struct {
+	HtlcNotifier
+}
+
+func (m *mockHTLCNotifier) NotifyFinalHtlcEvent(key channeldb.CircuitKey,
+	info channeldb.FinalHtlcInfo) { // nolint:whitespace
+}

htlcswitch/htlcnotifier.go

@@ -294,6 +294,18 @@ type SettleEvent struct {
 	Timestamp time.Time
 }
 
+type FinalHtlcEvent struct {
+	CircuitKey
+
+	Settled bool
+
+	// Offchain is indicating whether the htlc was resolved off-chain.
+	Offchain bool
+
+	// Timestamp is the time when this htlc was settled.
+	Timestamp time.Time
+}
+
 // NotifyForwardingEvent notifies the HtlcNotifier than a htlc has been
 // forwarded.
 //
@@ -382,6 +394,27 @@ func (h *HtlcNotifier) NotifySettleEvent(key HtlcKey,
 	}
 }
 
+// NotifyFinalHtlcEvent notifies the HtlcNotifier that the final outcome for an
+// htlc has been determined.
+//
+// Note this is part of the htlcNotifier interface.
+func (h *HtlcNotifier) NotifyFinalHtlcEvent(key channeldb.CircuitKey,
+	info channeldb.FinalHtlcInfo) {
+
+	event := &FinalHtlcEvent{
+		CircuitKey: key,
+		Settled:    info.Settled,
+		Offchain:   info.Offchain,
+		Timestamp:  h.now(),
+	}
+
+	log.Tracef("Notifying final settle event: %v", 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

htlcswitch/interfaces.go

@@ -370,4 +370,9 @@ type htlcNotifier interface {
 	// settled.
 	NotifySettleEvent(key HtlcKey, preimage lntypes.Preimage,
 		eventType HtlcEventType)
+
+	// NotifyFinalHtlcEvent notifies the HtlcNotifier that the final outcome
+	// for an htlc has been determined.
+	NotifyFinalHtlcEvent(key channeldb.CircuitKey,
+		info channeldb.FinalHtlcInfo)
 }

htlcswitch/link.go

@@ -1920,7 +1920,7 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 		// As we've just accepted a new state, we'll now
 		// immediately send the remote peer a revocation for our prior
 		// state.
-		nextRevocation, currentHtlcs, _, err :=
+		nextRevocation, currentHtlcs, finalHTLCs, err :=
 			l.channel.RevokeCurrentCommitment()
 		if err != nil {
 			l.log.Errorf("unable to revoke commitment: %v", err)
@@ -1928,6 +1928,21 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 		}
 		l.cfg.Peer.SendMessage(false, nextRevocation)
 
+		// Notify the incoming htlcs of which the resolutions were
+		// locked in.
+		for id, settled := range finalHTLCs {
+			l.cfg.HtlcNotifier.NotifyFinalHtlcEvent(
+				channeldb.CircuitKey{
+					ChanID: l.shortChanID,
+					HtlcID: id,
+				},
+				channeldb.FinalHtlcInfo{
+					Settled:  settled,
+					Offchain: true,
+				},
+			)
+		}
+
 		// Since we just revoked our commitment, we may have a new set
 		// of HTLC's on our commitment, so we'll send them using our
 		// function closure NotifyContractUpdate.

htlcswitch/mock.go

@@ -1077,7 +1077,9 @@ func (m *mockOnionErrorDecryptor) DecryptError(encryptedData []byte) (
 
 var _ htlcNotifier = (*mockHTLCNotifier)(nil)
 
-type mockHTLCNotifier struct{}
+type mockHTLCNotifier struct {
+	htlcNotifier
+}
 
 func (h *mockHTLCNotifier) NotifyForwardingEvent(key HtlcKey, info HtlcInfo,
 	eventType HtlcEventType) { // nolint:whitespace
@@ -1095,3 +1097,7 @@ func (h *mockHTLCNotifier) NotifyForwardingFailEvent(key HtlcKey,
 func (h *mockHTLCNotifier) NotifySettleEvent(key HtlcKey,
 	preimage lntypes.Preimage, eventType HtlcEventType) { // nolint:whitespace
 }
+
+func (h *mockHTLCNotifier) NotifyFinalHtlcEvent(key channeldb.CircuitKey,
+	info channeldb.FinalHtlcInfo) { // nolint:whitespace
+}

htlcswitch/switch_test.go

@@ -3535,6 +3535,13 @@ func checkHtlcEvents(t *testing.T, events <-chan interface{},
 			t.Fatalf("expected event: %v", expected)
 		}
 	}
+
+	// Check that there are no unexpected events following.
+	select {
+	case event := <-events:
+		t.Fatalf("unexpected event: %v", event)
+	default:
+	}
 }
 
 // sendThreeHopPayment is a helper function which sends a payment over
@@ -3638,6 +3645,12 @@ func getThreeHopEvents(channels *clusterChannels, htlcID uint64,
 				Incoming:      false,
 				Timestamp:     ts,
 			},
+			&FinalHtlcEvent{
+				CircuitKey: bobKey.IncomingCircuit,
+				Settled:    false,
+				Offchain:   true,
+				Timestamp:  ts,
+			},
 		}
 
 		return aliceEvents, bobEvents, nil
@@ -3669,6 +3682,12 @@ func getThreeHopEvents(channels *clusterChannels, htlcID uint64,
 			HtlcEventType: HtlcEventTypeForward,
 			Timestamp:     ts,
 		},
+		&FinalHtlcEvent{
+			CircuitKey: bobKey.IncomingCircuit,
+			Settled:    true,
+			Offchain:   true,
+			Timestamp:  ts,
+		},
 	}
 
 	carolEvents := []interface{}{
@@ -3683,6 +3702,14 @@ func getThreeHopEvents(channels *clusterChannels, htlcID uint64,
 			Preimage:      *preimage,
 			HtlcEventType: HtlcEventTypeReceive,
 			Timestamp:     ts,
+		}, &FinalHtlcEvent{
+			CircuitKey: channeldb.CircuitKey{
+				ChanID: channels.carolToBob.ShortChanID(),
+				HtlcID: htlcID,
+			},
+			Settled:   true,
+			Offchain:  true,
+			Timestamp: ts,
 		},
 	}
 

lnrpc/routerrpc/router.proto

@@ -619,6 +619,7 @@ message HtlcEvent {
         SettleEvent settle_event = 9;
         LinkFailEvent link_fail_event = 10;
         SubscribedEvent subscribed_event = 11;
+        FinalHtlcEvent final_htlc_event = 12;
     }
 }
 
@@ -649,6 +650,11 @@ message SettleEvent {
     bytes preimage = 1;
 }
 
+message FinalHtlcEvent {
+    bool settled = 1;
+    bool offchain = 2;
+}
+
 message SubscribedEvent {
 }
 

lnrpc/routerrpc/router.swagger.json

@@ -1168,6 +1168,17 @@
       ],
       "default": "UNKNOWN"
     },
+    "routerrpcFinalHtlcEvent": {
+      "type": "object",
+      "properties": {
+        "settled": {
+          "type": "boolean"
+        },
+        "offchain": {
+          "type": "boolean"
+        }
+      }
+    },
     "routerrpcForwardEvent": {
       "type": "object",
       "properties": {
@@ -1320,6 +1331,9 @@
         },
         "subscribed_event": {
           "$ref": "#/definitions/routerrpcSubscribedEvent"
+        },
+        "final_htlc_event": {
+          "$ref": "#/definitions/routerrpcFinalHtlcEvent"
         }
       },
       "title": "HtlcEvent contains the htlc event that was processed. These are served on a\nbest-effort basis; events are not persisted, delivery is not guaranteed\n(in the event of a crash in the switch, forward events may be lost) and\nsome events may be replayed upon restart. Events consumed from this package\nshould be de-duplicated by the htlc's unique combination of incoming and\noutgoing channel id and htlc id. [EXPERIMENTAL]"

lnrpc/routerrpc/subscribe_events.go

@@ -14,7 +14,7 @@ func rpcHtlcEvent(htlcEvent interface{}) (*HtlcEvent, error) {
 	var (
 		key       htlcswitch.HtlcKey
 		timestamp time.Time
-		eventType htlcswitch.HtlcEventType
+		eventType *htlcswitch.HtlcEventType
 		event     isHtlcEvent_Event
 	)
 
@@ -27,7 +27,7 @@ func rpcHtlcEvent(htlcEvent interface{}) (*HtlcEvent, error) {
 		}
 
 		key = e.HtlcKey
-		eventType = e.HtlcEventType
+		eventType = &e.HtlcEventType
 		timestamp = e.Timestamp
 
 	case *htlcswitch.ForwardingFailEvent:
@@ -36,7 +36,7 @@ func rpcHtlcEvent(htlcEvent interface{}) (*HtlcEvent, error) {
 		}
 
 		key = e.HtlcKey
-		eventType = e.HtlcEventType
+		eventType = &e.HtlcEventType
 		timestamp = e.Timestamp
 
 	case *htlcswitch.LinkFailEvent:
@@ -57,7 +57,7 @@ func rpcHtlcEvent(htlcEvent interface{}) (*HtlcEvent, error) {
 		}
 
 		key = e.HtlcKey
-		eventType = e.HtlcEventType
+		eventType = &e.HtlcEventType
 		timestamp = e.Timestamp
 
 	case *htlcswitch.SettleEvent:
@@ -68,7 +68,20 @@ func rpcHtlcEvent(htlcEvent interface{}) (*HtlcEvent, error) {
 		}
 
 		key = e.HtlcKey
-		eventType = e.HtlcEventType
+		eventType = &e.HtlcEventType
+		timestamp = e.Timestamp
+
+	case *htlcswitch.FinalHtlcEvent:
+		event = &HtlcEvent_FinalHtlcEvent{
+			FinalHtlcEvent: &FinalHtlcEvent{
+				Settled:  e.Settled,
+				Offchain: e.Offchain,
+			},
+		}
+
+		key = htlcswitch.HtlcKey{
+			IncomingCircuit: e.CircuitKey,
+		}
 		timestamp = e.Timestamp
 
 	default:
@@ -85,18 +98,21 @@ func rpcHtlcEvent(htlcEvent interface{}) (*HtlcEvent, error) {
 	}
 
 	// Convert the htlc event type to a rpc event.
-	switch eventType {
-	case htlcswitch.HtlcEventTypeSend:
-		rpcEvent.EventType = HtlcEvent_SEND
+	if eventType != nil {
+		switch *eventType {
+		case htlcswitch.HtlcEventTypeSend:
+			rpcEvent.EventType = HtlcEvent_SEND
 
-	case htlcswitch.HtlcEventTypeReceive:
-		rpcEvent.EventType = HtlcEvent_RECEIVE
+		case htlcswitch.HtlcEventTypeReceive:
+			rpcEvent.EventType = HtlcEvent_RECEIVE
 
-	case htlcswitch.HtlcEventTypeForward:
-		rpcEvent.EventType = HtlcEvent_FORWARD
+		case htlcswitch.HtlcEventTypeForward:
+			rpcEvent.EventType = HtlcEvent_FORWARD
 
-	default:
-		return nil, fmt.Errorf("unknown event type: %v", eventType)
+		default:
+			return nil, fmt.Errorf("unknown event type: %v",
+				eventType)
+		}
 	}
 
 	return rpcEvent, nil

lntest/itest/lnd_multi-hop-error-propagation_test.go

@@ -424,4 +424,14 @@ func assertLinkFailure(t *harnessTest,
 		t.Fatalf("expected: %v, got: %v", failureDetail,
 			linkFail.LinkFailEvent.FailureDetail)
 	}
+
+	event = assertEventAndType(t, routerrpc.HtlcEvent_UNKNOWN, client)
+	finalHtlc, ok := event.Event.(*routerrpc.HtlcEvent_FinalHtlcEvent)
+	if !ok {
+		t.Fatalf("expected final htlc, got: %T", event.Event)
+	}
+
+	if finalHtlc.FinalHtlcEvent.Settled {
+		t.Fatalf("expected final fail")
+	}
 }

lntest/itest/lnd_multi-hop-payments_test.go

@@ -348,13 +348,26 @@ func assertHtlcEvents(t *harnessTest, fwdCount, fwdFailCount, settleCount int,
 	userType routerrpc.HtlcEvent_EventType,
 	client routerrpc.Router_SubscribeHtlcEventsClient) {
 
-	var forwards, forwardFails, settles int
+	var forwards, forwardFails, settles, finalSettles, finalFails int
+
+	var finalFailCount, finalSettleCount int
+	if userType != routerrpc.HtlcEvent_SEND {
+		finalFailCount = fwdFailCount
+		finalSettleCount = settleCount
+	}
+
+	numEvents := fwdCount + fwdFailCount + settleCount +
+		finalFailCount + finalSettleCount
 
-	numEvents := fwdCount + fwdFailCount + settleCount
 	for i := 0; i < numEvents; i++ {
-		event := assertEventAndType(t, userType, client)
+		event, err := client.Recv()
+		if err != nil {
+			t.Fatalf("could not get event")
+		}
 
-		switch event.Event.(type) {
+		expectedEventType := userType
+
+		switch e := event.Event.(type) {
 		case *routerrpc.HtlcEvent_ForwardEvent:
 			forwards++
 
@@ -364,9 +377,23 @@ func assertHtlcEvents(t *harnessTest, fwdCount, fwdFailCount, settleCount int,
 		case *routerrpc.HtlcEvent_SettleEvent:
 			settles++
 
+		case *routerrpc.HtlcEvent_FinalHtlcEvent:
+			if e.FinalHtlcEvent.Settled {
+				finalSettles++
+			} else {
+				finalFails++
+			}
+
+			expectedEventType = routerrpc.HtlcEvent_UNKNOWN
+
 		default:
 			t.Fatalf("unexpected event: %T", event.Event)
 		}
+
+		if event.EventType != expectedEventType {
+			t.Fatalf("expected: %v, got: %v", expectedEventType,
+				event.EventType)
+		}
 	}
 
 	if forwards != fwdCount {
@@ -378,9 +405,19 @@ func assertHtlcEvents(t *harnessTest, fwdCount, fwdFailCount, settleCount int,
 			forwardFails)
 	}
 
+	if finalFails != finalFailCount {
+		t.Fatalf("expected: %v final fails, got: %v", finalFailCount,
+			finalFails)
+	}
+
 	if settles != settleCount {
 		t.Fatalf("expected: %v settles, got: %v", settleCount, settles)
 	}
+
+	if finalSettles != finalSettleCount {
+		t.Fatalf("expected: %v settles, got: %v", finalSettleCount,
+			finalSettles)
+	}
 }
 
 // assertEventAndType reads an event from the stream provided and ensures that

server.go

@@ -1166,6 +1166,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 		Clock:                         clock.NewDefaultClock(),
 		SubscribeBreachComplete:       s.breachArbiter.SubscribeBreachComplete,
 		PutFinalHtlcOutcome:           s.chanStateDB.PutOnchainFinalHtlcOutcome, // nolint: lll
+		HtlcNotifier:                  s.htlcNotifier,
 	}, dbs.ChanStateDB)
 
 	// Select the configuration and furnding parameters for Bitcoin or