WIP: add RFQ system HTLC accepted event notification

Author: ffranr

itest/rfq_test.go

@@ -2,6 +2,8 @@ package itest
 
 import (
 	"context"
+	"fmt"
+	"math"
 	"time"
 
 	"github.com/btcsuite/btcd/btcutil"
@@ -10,6 +12,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/taprpc/rfqrpc"
 	"github.com/lightningnetwork/lnd/chainreg"
 	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/lightningnetwork/lnd/lnrpc/routerrpc"
 	"github.com/lightningnetwork/lnd/lntest"
 	"github.com/lightningnetwork/lnd/lntest/node"
 	"github.com/lightningnetwork/lnd/lntest/wait"
@@ -96,9 +99,8 @@ func testRfqHtlcIntercept(t *harnessTest) {
 		event, err := carolEventNtfns.Recv()
 		require.NoError(t.t, err)
 
-		if _, ok := event.Event.(*rfqrpc.RfqEvent_IncomingAcceptQuote); ok {
-			return nil
-		}
+		_, ok := event.Event.(*rfqrpc.RfqEvent_IncomingAcceptQuote)
+		require.True(t.t, ok, "unexpected event: %v", event)
 
 		return nil
 	}, defaultWaitTimeout)
@@ -159,13 +161,53 @@ func testRfqHtlcIntercept(t *harnessTest) {
 	})
 	invoice := ts.CarolLnd.RPC.LookupInvoice(addInvoiceResp.RHash)
 
-	// Alice pays the invoice.
-	t.lndHarness.CompletePaymentRequests(
-		ts.AliceLnd, []string{invoice.PaymentRequest},
+	// Register to receive RFQ events from Bob's tapd node. We'll use this
+	// to wait for Bob to receive the HTLC with the asset transfer specific
+	// scid.
+	bobEventNtfns, err := ts.BobTapd.SubscribeRfqEventNtfns(
+		ctxb, &rfqrpc.SubscribeRfqEventNtfnsRequest{},
 	)
+	require.NoError(t.t, err)
+
+	// Alice pays the invoice.
+	t.Log("Alice paying invoice")
+	req := &routerrpc.SendPaymentRequest{
+		PaymentRequest: invoice.PaymentRequest,
+		TimeoutSeconds: int32(wait.PaymentTimeout.Seconds()),
+		FeeLimitMsat:   math.MaxInt64,
+	}
+	ts.AliceLnd.RPC.SendPayment(req)
+	t.Log("Alice payment sent")
 
 	// At this point Bob should have received a HTLC with the asset transfer
-	// specific scid.
+	// specific scid. We'll wait for Bob to publish an accept HTLC event and
+	// then validate it against the accepted quote.
+	waitErr = wait.NoError(func() error {
+		t.Log("Waiting for Bob to receive HTLC")
+
+		event, err := bobEventNtfns.Recv()
+		require.NoError(t.t, err)
+
+		acceptHtlc, ok := event.Event.(*rfqrpc.RfqEvent_AcceptHtlc)
+		if ok {
+			require.Equal(
+				t.t, acceptedQuote.Scid,
+				acceptHtlc.AcceptHtlc.Scid,
+			)
+			t.Log("Bob has accepted the HTLC")
+			return nil
+		}
+
+		return fmt.Errorf("unexpected event: %v", event)
+	}, defaultWaitTimeout)
+	require.NoError(t.t, waitErr)
+
+	// Close event streams.
+	err = carolEventNtfns.CloseSend()
+	require.NoError(t.t, err)
+
+	err = bobEventNtfns.CloseSend()
+	require.NoError(t.t, err)
 }
 
 // newLndNode creates a new lnd node with the given name and funds its wallet

rfq/manager.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/btcec/v2"
+	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/taproot-assets/asset"
 	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/rfqmsg"
@@ -77,6 +78,10 @@ type Manager struct {
 	// messages.
 	outgoingMessages chan rfqmsg.OutgoingMsg
 
+	// acceptHtlcEvents is a channel which is populated with accept HTLCs
+	// events.
+	acceptHtlcEvents chan AcceptHtlcEvent
+
 	// peerAcceptedQuotes is a map of serialised short channel IDs (SCIDs)
 	// to associated accepted quotes. These quotes have been accepted by
 	// peer nodes and are therefore available for use in buying assets.
@@ -105,6 +110,7 @@ func NewManager(cfg ManagerCfg) (Manager, error) {
 		incomingMessages: make(chan rfqmsg.IncomingMsg),
 		outgoingMessages: make(chan rfqmsg.OutgoingMsg),
 
+		acceptHtlcEvents:   make(chan AcceptHtlcEvent),
 		peerAcceptedQuotes: make(map[SerialisedScid]rfqmsg.Accept),
 
 		subscribers: make(
@@ -124,8 +130,9 @@ func (m *Manager) startSubsystems(ctx context.Context) error {
 
 	// Initialise and start the order handler.
 	m.orderHandler, err = NewOrderHandler(OrderHandlerCfg{
-		CleanupInterval: CacheCleanupInterval,
-		HtlcInterceptor: m.cfg.HtlcInterceptor,
+		CleanupInterval:  CacheCleanupInterval,
+		HtlcInterceptor:  m.cfg.HtlcInterceptor,
+		AcceptHtlcEvents: m.acceptHtlcEvents,
 	})
 	if err != nil {
 		return fmt.Errorf("error initializing RFQ order handler: %w",
@@ -339,6 +346,10 @@ func (m *Manager) mainEventLoop() {
 					err)
 			}
 
+		case acceptHtlcEvent := <-m.acceptHtlcEvents:
+			// Handle a HTLC accept event. Notify any subscribers.
+			m.publishSubscriberEvent(&acceptHtlcEvent)
+
 		// Handle errors from the negotiator.
 		case err := <-m.negotiator.ErrChan:
 			log.Warnf("Negotiator has encountered an error: %v",
@@ -487,6 +498,7 @@ func (m *Manager) publishSubscriberEvent(event fn.Event) {
 // IncomingAcceptQuoteEvent is an event that is broadcast when the RFQ manager
 // receives an accept quote message from a peer.
 type IncomingAcceptQuoteEvent struct {
+	// timestamp is the event creation UTC timestamp.
 	timestamp time.Time
 
 	// Accept is the accepted quote.
@@ -511,3 +523,35 @@ func (q *IncomingAcceptQuoteEvent) Timestamp() time.Time {
 // Ensure that the IncomingAcceptQuoteEvent struct implements the Event
 // interface.
 var _ fn.Event = (*IncomingAcceptQuoteEvent)(nil)
+
+// AcceptHtlcEvent is an event that is sent to the accept HTLCs channel when
+// an HTLC is accepted.
+type AcceptHtlcEvent struct {
+	// Timestamp is the unix timestamp at which the HTLC was accepted.
+	timestamp uint64
+
+	// Htlc is the intercepted HTLC.
+	Htlc lndclient.InterceptedHtlc
+
+	// ChannelRemit is the channel remit that the HTLC complies with.
+	ChannelRemit ChannelRemit
+}
+
+// NewAcceptHtlcEvent creates a new AcceptedHtlcEvent.
+func NewAcceptHtlcEvent(htlc lndclient.InterceptedHtlc,
+	channelRemit ChannelRemit) AcceptHtlcEvent {
+
+	return AcceptHtlcEvent{
+		timestamp:    uint64(time.Now().UTC().Unix()),
+		Htlc:         htlc,
+		ChannelRemit: channelRemit,
+	}
+}
+
+// Timestamp returns the event creation UTC timestamp.
+func (q *AcceptHtlcEvent) Timestamp() time.Time {
+	return time.Unix(int64(q.timestamp), 0).UTC()
+}
+
+// Ensure that the AcceptedHtlcEvent struct implements the Event interface.
+var _ fn.Event = (*AcceptHtlcEvent)(nil)

rfq/order.go

@@ -85,6 +85,9 @@ type OrderHandlerCfg struct {
 	// HtlcInterceptor is the HTLC interceptor. This component is used to
 	// intercept and accept/reject HTLCs.
 	HtlcInterceptor HtlcInterceptor
+
+	// AcceptHtlcEvents is a channel that receives accepted HTLCs.
+	AcceptHtlcEvents chan<- AcceptHtlcEvent
 }
 
 // OrderHandler orchestrates management of accepted quote bundles. It monitors
@@ -157,7 +160,9 @@ func (h *OrderHandler) handleIncomingHtlc(_ context.Context,
 		}, nil
 	}
 
-	log.Info("HTLC complies with channel remit. Accepting HTLC")
+	log.Debug("HTLC complies with channel remit.")
+	acceptHtlcEvent := NewAcceptHtlcEvent(htlc, *channelRemit)
+	h.cfg.AcceptHtlcEvents <- acceptHtlcEvent
 
 	return &lndclient.InterceptedHtlcResponse{
 		Action: lndclient.InterceptorActionResume,

rfq/stream.go

@@ -3,7 +3,6 @@ package rfq
 import (
 	"context"
 	"fmt"
-	"math/rand"
 	"sync"
 
 	"github.com/lightninglabs/lndclient"
@@ -43,8 +42,6 @@ type StreamHandler struct {
 	// stream handler.
 	cfg StreamHandlerCfg
 
-	randId int
-
 	// recvRawMessages is a channel that receives incoming raw peer
 	// messages.
 	recvRawMessages <-chan lndclient.CustomMessage
@@ -81,15 +78,9 @@ func NewStreamHandler(ctx context.Context, cfg StreamHandlerCfg,
 			"messages via peer message porter: %w", err)
 	}
 
-	minRand := 100
-	maxRand := 1000
-	randId := rand.Intn(maxRand-minRand+1) + minRand
-
 	return &StreamHandler{
 		cfg: cfg,
 
-		randId: randId,
-
 		recvRawMessages:    msgChan,
 		errRecvRawMessages: peerMsgErrChan,
 
@@ -132,9 +123,10 @@ func (h *StreamHandler) handleIncomingQuoteRequest(
 
 	h.seenMessagesMtx.Unlock()
 
-	log.Debugf("(%d) Stream handling incoming message (msg_type=%T, msg_id=%s, "+
-		"origin_peer=%s, self=%s)", h.randId, quoteRequest, quoteRequest.ID.String(),
-		quoteRequest.MsgPeer(), h.cfg.LightningSelfId)
+	log.Debugf("Stream handling incoming message (msg_type=%T, "+
+		"msg_id=%s, origin_peer=%s, self=%s)", quoteRequest,
+		quoteRequest.ID.String(), quoteRequest.MsgPeer(),
+		h.cfg.LightningSelfId)
 
 	// Send the quote request to the RFQ manager.
 	var msg rfqmsg.IncomingMsg = quoteRequest

rpcserver.go

@@ -4771,6 +4771,17 @@ func marshallRfqEvent(eventInterface fn.Event) (*rfqrpc.RfqEvent, error) {
 		return &rfqrpc.RfqEvent{
 			Event: eventRpc,
 		}, nil
+
+	case *rfq.AcceptHtlcEvent:
+		eventRpc := &rfqrpc.RfqEvent_AcceptHtlc{
+			AcceptHtlc: &rfqrpc.AcceptHtlcEvent{
+				Timestamp: uint64(timestamp),
+				Scid:      uint64(event.ChannelRemit.Scid),
+			},
+		}
+		return &rfqrpc.RfqEvent{
+			Event: eventRpc,
+		}, nil
 	}
 
 	return nil, fmt.Errorf("unknown RFQ event type: %T", eventInterface)

taprpc/rfqrpc/rfq.proto

@@ -105,6 +105,9 @@ message QueryRfqAcceptedQuotesResponse {
     repeated AcceptedQuote accepted_quotes = 1;
 }
 
+message SubscribeRfqEventNtfnsRequest {
+}
+
 message IncomingAcceptQuoteEvent {
     // Unix timestamp.
     uint64 timestamp = 1;
@@ -113,13 +116,23 @@ message IncomingAcceptQuoteEvent {
     AcceptedQuote accepted_quote = 2;
 }
 
-message SubscribeRfqEventNtfnsRequest {
+message AcceptHtlcEvent {
+    // Unix timestamp.
+    uint64 timestamp = 1;
+
+    // scid is the short channel ID of the channel over which the payment for
+    // the quote is made.
+    uint64 scid = 2;
 }
 
 message RfqEvent {
     oneof event {
         // incoming_accept_quote is an event that is sent when an incoming
         // accept quote message is received.
         IncomingAcceptQuoteEvent incoming_accept_quote = 1;
+
+        // accept_htlc is an event that is sent when a HTLC is accepted by the
+        // RFQ service.
+        AcceptHtlcEvent accept_htlc = 2;
     }
 }