routerrpc: add htlc notifier subscribed event

lnrpc/routerrpc/router.proto

@@ -618,6 +618,7 @@ message HtlcEvent {
         ForwardFailEvent forward_fail_event = 8;
         SettleEvent settle_event = 9;
         LinkFailEvent link_fail_event = 10;
+        SubscribedEvent subscribed_event = 11;
     }
 }
 
@@ -648,6 +649,9 @@ message SettleEvent {
     bytes preimage = 1;
 }
 
+message SubscribedEvent {
+}
+
 message LinkFailEvent {
     // Info contains details about the htlc that we failed.
     HtlcInfo info = 1;

lnrpc/routerrpc/router.swagger.json

@@ -1317,6 +1317,9 @@
         },
         "link_fail_event": {
           "$ref": "#/definitions/routerrpcLinkFailEvent"
+        },
+        "subscribed_event": {
+          "$ref": "#/definitions/routerrpcSubscribedEvent"
         }
       },
       "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]"
@@ -1748,6 +1751,9 @@
         }
       }
     },
+    "routerrpcSubscribedEvent": {
+      "type": "object"
+    },
     "routerrpcUpdateChanStatusRequest": {
       "type": "object",
       "properties": {

lnrpc/routerrpc/router_server.go

@@ -900,6 +900,16 @@ func (s *Server) SubscribeHtlcEvents(req *SubscribeHtlcEventsRequest,
 	}
 	defer htlcClient.Cancel()
 
+	// Send out an initial subscribed event so that the caller knows the
+	// point from which new events will be transmitted.
+	if err := stream.Send(&HtlcEvent{
+		Event: &HtlcEvent_SubscribedEvent{
+			SubscribedEvent: &SubscribedEvent{},
+		},
+	}); err != nil {
+		return err
+	}
+
 	for {
 		select {
 		case event := <-htlcClient.Updates():

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

@@ -168,6 +168,7 @@ out:
 	if err != nil {
 		t.Fatalf("could not subscribe events: %v", err)
 	}
+	assertSubscribed(t, aliceEvents)
 
 	bobEvents, err := net.Bob.RouterClient.SubscribeHtlcEvents(
 		ctxt, &routerrpc.SubscribeHtlcEventsRequest{},
@@ -175,6 +176,7 @@ out:
 	if err != nil {
 		t.Fatalf("could not subscribe events: %v", err)
 	}
+	assertSubscribed(t, bobEvents)
 
 	carolEvents, err := carol.RouterClient.SubscribeHtlcEvents(
 		ctxt, &routerrpc.SubscribeHtlcEventsRequest{},
@@ -182,6 +184,7 @@ out:
 	if err != nil {
 		t.Fatalf("could not subscribe events: %v", err)
 	}
+	assertSubscribed(t, carolEvents)
 
 	// For the first scenario, we'll test the cancellation of an HTLC with
 	// an unknown payment hash.

lntest/itest/lnd_multi-hop-payments_test.go

@@ -172,6 +172,7 @@ func testMultiHopPayments(net *lntest.NetworkHarness, t *harnessTest) {
 	if err != nil {
 		t.Fatalf("could not subscribe events: %v", err)
 	}
+	assertSubscribed(t, aliceEvents)
 
 	bobEvents, err := net.Bob.RouterClient.SubscribeHtlcEvents(
 		ctxt, &routerrpc.SubscribeHtlcEventsRequest{},
@@ -179,6 +180,7 @@ func testMultiHopPayments(net *lntest.NetworkHarness, t *harnessTest) {
 	if err != nil {
 		t.Fatalf("could not subscribe events: %v", err)
 	}
+	assertSubscribed(t, bobEvents)
 
 	carolEvents, err := carol.RouterClient.SubscribeHtlcEvents(
 		ctxt, &routerrpc.SubscribeHtlcEventsRequest{},
@@ -186,6 +188,7 @@ func testMultiHopPayments(net *lntest.NetworkHarness, t *harnessTest) {
 	if err != nil {
 		t.Fatalf("could not subscribe events: %v", err)
 	}
+	assertSubscribed(t, carolEvents)
 
 	daveEvents, err := dave.RouterClient.SubscribeHtlcEvents(
 		ctxt, &routerrpc.SubscribeHtlcEventsRequest{},
@@ -193,6 +196,7 @@ func testMultiHopPayments(net *lntest.NetworkHarness, t *harnessTest) {
 	if err != nil {
 		t.Fatalf("could not subscribe events: %v", err)
 	}
+	assertSubscribed(t, daveEvents)
 
 	// Using Carol as the source, pay to the 5 invoices from Bob created
 	// above.
@@ -399,6 +403,14 @@ func assertEventAndType(t *harnessTest, eventType routerrpc.HtlcEvent_EventType,
 	return event
 }
 
+func assertSubscribed(t *harnessTest,
+	client routerrpc.Router_SubscribeHtlcEventsClient) {
+
+	event, err := client.Recv()
+	require.NoError(t.t, err)
+	require.NotNil(t.t, event.GetSubscribedEvent())
+}
+
 // updateChannelPolicy updates the channel policy of node to the
 // given fees and timelock delta. This function blocks until
 // listenerNode has received the policy update.