WIP

Author: ffranr

rfqmsg/accept.go

@@ -152,7 +152,9 @@ type Accept struct {
 // NewAcceptFromRequest creates a new instance of a quote accept message given
 // a quote request message.
 func NewAcceptFromRequest(request Request, askingPrice lnwire.MilliSatoshi,
-	expiry uint64, sig [64]byte) Accept {
+	expiry uint64) Accept {
+
+	var sig [64]byte
 
 	return Accept{
 		Peer:        request.Peer,

rfqservice/negotiator.go

@@ -15,9 +15,8 @@ type NegotiatorCfg struct {
 	PriceOracle PriceOracle
 }
 
-// Negotiator is a struct that handles the negotiation of quotes. It is a
-// subsystem of the RFQ system. It determines whether a quote is accepted or
-// rejected.
+// Negotiator is a struct that handles the negotiation of quotes. It is a RFQ
+// subsystem. It determines whether a quote is accepted or rejected.
 type Negotiator struct {
 	startOnce sync.Once
 	stopOnce  sync.Once
@@ -50,7 +49,7 @@ func NewNegotiator(cfg NegotiatorCfg,
 
 // queryPriceOracle queries the price oracle for the asking price.
 func (n *Negotiator) queryPriceOracle(
-	req rfqmsg.Request) (*PriceOracleSuggestedRate, error) {
+	req rfqmsg.Request) (*OracleAskingPriceResponse, error) {
 
 	ctx, cancel := n.WithCtxQuitNoTimeout()
 	defer cancel()
@@ -66,46 +65,47 @@ func (n *Negotiator) queryPriceOracle(
 	return res, nil
 }
 
-// handlePriceOracleResponse handles the response from the price oracle.
-func (n *Negotiator) handlePriceOracleResponse(
-	request rfqmsg.Request, oracleResponse *PriceOracleSuggestedRate) error {
+// handlePriceOracleResponse handles a response from the price oracle.
+func (n *Negotiator) handlePriceOracleResponse(request rfqmsg.Request,
+	oracleResponse *OracleAskingPriceResponse) error {
 
-	// If the suggested rate is nil, then we will return the error message
+	// If the asking price is nil, then we will return the error message
 	// supplied by the price oracle.
 	if oracleResponse.AskingPrice == nil {
 		rejectMsg := rfqmsg.NewRejectMsg(
-			request.Peer, request.ID, oracleResponse.Err,
+			request.Peer, request.ID, *oracleResponse.Err,
 		)
 		var msg rfqmsg.OutgoingMsg = &rejectMsg
 
 		sendSuccess := fn.SendOrQuit(n.outgoingMessages, msg, n.Quit)
 		if !sendSuccess {
-			return fmt.Errorf("negotiator failed to send reject " +
-				"message")
+			return fmt.Errorf("negotiator failed to add reject " +
+				"message to the outgoing messages channel")
 		}
 	}
 
-	// If the suggested rate is not nil, then we can proceed to respond
-	// with an accept message.
-	var sig [64]byte
+	// TODO(ffranr): Ensure that the expiry time is valid and sufficient.
+
+	// If the asking price is not nil, then we can proceed to respond with
+	// an accept message.
 	acceptMsg := rfqmsg.NewAcceptFromRequest(
-		request, *oracleResponse.AskingPrice, oracleResponse.Expiry, sig,
+		request, *oracleResponse.AskingPrice, oracleResponse.Expiry,
 	)
 	var msg rfqmsg.OutgoingMsg = &acceptMsg
 
 	sendSuccess := fn.SendOrQuit(n.outgoingMessages, msg, n.Quit)
 	if !sendSuccess {
-		return fmt.Errorf("negotiator failed to populate accept message " +
-			"channel")
+		return fmt.Errorf("negotiator failed to add accept message " +
+			"to the outgoing messages channel")
 	}
 
 	return nil
 }
 
 // HandleIncomingQuoteRequest handles an incoming quote request.
 func (n *Negotiator) HandleIncomingQuoteRequest(req rfqmsg.Request) error {
-	// If there is no price oracle available, then we cannot proceed with the
-	// negotiation. We will reject the quote request with an error.
+	// If there is no price oracle available, then we cannot proceed with
+	// the negotiation. We will reject the quote request with an error.
 	if n.cfg.PriceOracle == nil {
 		rejectMsg := rfqmsg.NewRejectMsg(
 			req.Peer, req.ID, rfqmsg.ErrPriceOracleUnavailable,
@@ -114,7 +114,8 @@ func (n *Negotiator) HandleIncomingQuoteRequest(req rfqmsg.Request) error {
 
 		sendSuccess := fn.SendOrQuit(n.outgoingMessages, msg, n.Quit)
 		if !sendSuccess {
-			return fmt.Errorf("negotiator failed to send reject message")
+			return fmt.Errorf("negotiator failed to send reject " +
+				"message")
 		}
 	}
 
@@ -174,6 +175,7 @@ func (n *Negotiator) Start() error {
 func (n *Negotiator) Stop() error {
 	log.Info("Stopping RFQ subsystem: quote negotiator")
 
+	// Stop the main event loop.
 	close(n.Quit)
 	return nil
 }

rfqservice/oracle.go

@@ -14,26 +14,17 @@ import (
 	"google.golang.org/grpc/credentials"
 )
 
-// PriceOracleSuggestedRate is a struct that holds the price oracle's suggested
-// exchange rate for an asset.
-type PriceOracleSuggestedRate struct {
-	// AssetID is the asset ID.
-	AssetID *asset.ID
-
-	// AssetGroupKey is the asset group key.
-	AssetGroupKey *btcec.PublicKey
-
-	// AssetAmount is the asset amount.
-	AssetAmount uint64
-
+// OracleAskingPriceResponse is a struct that holds the price oracle's suggested
+// asking price for an asset.
+type OracleAskingPriceResponse struct {
 	// AskingPrice is the asking price of the quote.
 	AskingPrice *lnwire.MilliSatoshi
 
 	// Expiry is the asking price expiry lifetime unix timestamp.
 	Expiry uint64
 
 	// Err is the error returned by the price oracle service.
-	Err rfqmsg.RejectErr
+	Err *rfqmsg.RejectErr
 }
 
 // PriceOracle is an interface that provides exchange rate information for
@@ -42,7 +33,7 @@ type PriceOracle interface {
 	// QueryAskingPrice returns the asking price for the given asset amount.
 	QueryAskingPrice(ctx context.Context, assetId *asset.ID,
 		assetGroupKey *btcec.PublicKey, assetAmount uint64,
-		bidPrice *lnwire.MilliSatoshi) (*PriceOracleSuggestedRate,
+		bidPrice *lnwire.MilliSatoshi) (*OracleAskingPriceResponse,
 		error)
 }
 
@@ -85,7 +76,7 @@ func NewRpcPriceOracle(addr url.URL) (*RpcPriceOracle, error) {
 // QueryAskingPrice returns the asking price for the given asset amount.
 func (r *RpcPriceOracle) QueryAskingPrice(ctx context.Context,
 	assetId *asset.ID, assetGroupKey *btcec.PublicKey, assetAmount uint64,
-	bidPrice *lnwire.MilliSatoshi) (*PriceOracleSuggestedRate, error) {
+	bidPrice *lnwire.MilliSatoshi) (*OracleAskingPriceResponse, error) {
 
 	//// Call the external oracle service to get the exchange rate.
 	//conn := getClientConn(ctx, false)
@@ -111,18 +102,15 @@ func NewMockPriceOracle(rateLifetime uint64) *MockPriceOracle {
 
 // QueryAskingPrice returns the asking price for the given asset amount.
 func (m *MockPriceOracle) QueryAskingPrice(_ context.Context,
-	assetId *asset.ID, assetGroupKey *btcec.PublicKey, assetAmt uint64,
-	bidPrice *lnwire.MilliSatoshi) (*PriceOracleSuggestedRate, error) {
+	_ *asset.ID, _ *btcec.PublicKey, _ uint64,
+	bidPrice *lnwire.MilliSatoshi) (*OracleAskingPriceResponse, error) {
 
 	// Calculate the rate expiry lifetime.
 	expiry := uint64(time.Now().Unix()) + m.rateLifetime
 
-	return &PriceOracleSuggestedRate{
-		AssetID:       assetId,
-		AssetGroupKey: assetGroupKey,
-		AssetAmount:   assetAmt,
-		AskingPrice:   bidPrice,
-		Expiry:        expiry,
+	return &OracleAskingPriceResponse{
+		AskingPrice: bidPrice,
+		Expiry:      expiry,
 	}, nil
 }
 

rfqservice/order.go

@@ -18,15 +18,15 @@ type SerialisedScid uint64
 // ChannelRemit is a struct that holds the terms which determine whether a
 // channel HTLC is accepted or rejected.
 type ChannelRemit struct {
-	// Scid is the serialised short channel ID (SCID) of the channel for
+	// Scid is the serialised short channel ID (SCID) of the channel to
 	// which the remit applies.
 	Scid SerialisedScid
 
 	// AssetAmount is the amount of the tap asset that is being requested.
 	AssetAmount uint64
 
-	// MinimumChannelPayment is the minimum number of millisatoshis that must be
-	// sent in the HTLC.
+	// MinimumChannelPayment is the minimum number of millisatoshis that
+	// must be sent in the HTLC.
 	MinimumChannelPayment lnwire.MilliSatoshi
 
 	// Expiry is the asking price expiry lifetime unix timestamp.
@@ -51,23 +51,25 @@ func NewChannelRemit(quoteAccept rfqmsg.Accept) (*ChannelRemit, error) {
 func (c *ChannelRemit) CheckHtlcCompliance(
 	htlc lndclient.InterceptedHtlc) error {
 
+	// Check that the channel SCID is as expected.
+	htlcScid := SerialisedScid(htlc.OutgoingChannelID.ToUint64())
+	if htlcScid != c.Scid {
+		return fmt.Errorf("htlc outgoing channel ID does not match "+
+			"remit's SCID (htlc_scid=%d, remit_scid=%d)", htlcScid,
+			c.Scid)
+	}
+
 	// Check that the HTLC amount is at least the minimum acceptable amount.
 	if htlc.AmountOutMsat <= c.MinimumChannelPayment {
 		return fmt.Errorf("htlc out amount is less than the remit's "+
 			"minimum (htlc_out_msat=%d, remit_min_msat=%d)",
 			htlc.AmountOutMsat, c.MinimumChannelPayment)
 	}
 
-	// Check that the channel SCID is as expected.
-	htlcScid := SerialisedScid(htlc.OutgoingChannelID.ToUint64())
-	if htlcScid != c.Scid {
-		return fmt.Errorf("htlc outgoing channel ID does not match "+
-			"remit's SCID (htlc_scid=%d, remit_scid=%d)", htlcScid, c.Scid)
-	}
-
 	// Lastly, check to ensure that the channel remit has not expired.
 	if time.Now().Unix() > int64(c.Expiry) {
-		return fmt.Errorf("channel remit has expired (expiry=%d)", c.Expiry)
+		return fmt.Errorf("channel remit has expired (expiry=%d)",
+			c.Expiry)
 	}
 
 	return nil
@@ -76,19 +78,18 @@ func (c *ChannelRemit) CheckHtlcCompliance(
 // OrderHandlerCfg is a struct that holds the configuration parameters for the
 // order handler service.
 type OrderHandlerCfg struct {
-	// CleanupInterval is the interval at which the order
-	// handler cleans up expired accepted quotes from its local cache.
+	// CleanupInterval is the interval at which the order handler cleans up
+	// expired accepted quotes from its local cache.
 	CleanupInterval time.Duration
 
 	// HtlcInterceptor is the HTLC interceptor. This component is used to
 	// intercept and accept/reject HTLCs.
 	HtlcInterceptor HtlcInterceptor
 }
 
-// OrderHandler orchestrates management of accepted RFQ (Request For Quote)
-// bundles. It monitors HTLCs (Hash Time Locked Contracts), determining
-// acceptance or rejection based on compliance with the terms of the associated
-// quote.
+// OrderHandler orchestrates management of accepted quote bundles. It monitors
+// HTLCs (Hash Time Locked Contracts), and determines acceptance/rejection based
+// on the terms of the associated accepted quote.
 type OrderHandler struct {
 	startOnce sync.Once
 	stopOnce  sync.Once
@@ -97,7 +98,7 @@ type OrderHandler struct {
 	cfg OrderHandlerCfg
 
 	// channelRemits is a map of serialised short channel IDs (SCIDs) to
-	// associated active channel quote remits.
+	// associated active channel remits.
 	channelRemits map[SerialisedScid]ChannelRemit
 
 	// channelRemitsMtx guards the channelRemits map.
@@ -241,7 +242,7 @@ func (h *OrderHandler) RegisterChannelRemit(quoteAccept rfqmsg.Accept) error {
 
 // FetchChannelRemit fetches a channel remit given a serialised SCID. If a
 // channel remit is not found, false is returned. Expired channel remits are
-// also not returned and are removed from the cache.
+// not returned and are removed from the cache.
 func (h *OrderHandler) FetchChannelRemit(scid SerialisedScid) (*ChannelRemit,
 	bool) {
 
@@ -278,15 +279,16 @@ func (h *OrderHandler) cleanupStaleChannelRemits() {
 	}
 
 	if staleCounter > 0 {
-		log.Tracef("Removed %d stale channel remits from the order "+
-			"handler", staleCounter)
+		log.Tracef("Removed stale channel remits from the order "+
+			"handler: (count=%d)", staleCounter)
 	}
 }
 
 // Stop stops the handler.
 func (h *OrderHandler) Stop() error {
 	log.Info("Stopping RFQ subsystem: order handler")
 
+	// Stop the main event loop.
 	close(h.Quit)
 	return nil
 }

rfqservice/rfq.go

@@ -183,6 +183,9 @@ func (m *Manager) Stop() error {
 	m.stopOnce.Do(func() {
 		log.Info("Stopping RFQ system")
 		stopErr = m.stopSubsystems()
+
+		// Stop the main event loop.
+		close(m.Quit)
 	})
 
 	return stopErr
@@ -221,8 +224,8 @@ func (m *Manager) handleIncomingMessage(incomingMsg rfqmsg.IncomingMsg) error {
 	case *rfqmsg.Request:
 		err := m.negotiator.HandleIncomingQuoteRequest(*msg)
 		if err != nil {
-			return fmt.Errorf("error handling incoming quote request: %w",
-				err)
+			return fmt.Errorf("error handling incoming quote "+
+				"request: %w", err)
 		}
 	}
 
@@ -234,11 +237,12 @@ func (m *Manager) handleOutgoingMessage(outgoingMsg rfqmsg.OutgoingMsg) error {
 	// Perform type specific handling of the outgoing message.
 	switch msg := outgoingMsg.(type) {
 	case *rfqmsg.Accept:
-		// Inform the HTLC order handler that we've accepted the quote request.
-		// TODO(ffranr): set the asset amount correctly
+		// Inform the HTLC order handler that we've accepted the quote
+		// request.
 		err := m.orderHandler.RegisterChannelRemit(*msg)
 		if err != nil {
-			return fmt.Errorf("error registering channel remit: %w", err)
+			return fmt.Errorf("error registering channel remit: %w",
+				err)
 		}
 	}
 
@@ -258,26 +262,29 @@ func (m *Manager) mainEventLoop() {
 		select {
 		// Handle incoming message.
 		case incomingMsg := <-m.incomingMessages:
-			log.Debugf("RFQ manager has received an incoming message")
+			log.Debugf("RFQ manager has received an incoming " +
+				"message")
 
 			err := m.handleIncomingMessage(incomingMsg)
 			if err != nil {
-				log.Warnf("Error handling incoming quote "+
-					"request: %v", err)
+				log.Warnf("Error handling incoming message: %v",
+					err)
 			}
 
 		// Handle outgoing message.
 		case outgoingMsg := <-m.outgoingMessages:
-			log.Debugf("RFQ manager has received an outgoing message.")
+			log.Debugf("RFQ manager has received an outgoing " +
+				"message.")
 
 			err := m.handleOutgoingMessage(outgoingMsg)
 			if err != nil {
-				log.Warnf("Error handling outgoing message: %v", err)
+				log.Warnf("Error handling outgoing message: %v",
+					err)
 			}
 
 		case <-m.Quit:
-			log.Debug("RFQ manager main event loop has received the " +
-				"shutdown signal")
+			log.Debug("RFQ manager main event loop has received " +
+				"the shutdown signal")
 			return
 		}
 	}