rfq+rpcserver: match event to order

Fixes an issue reported on Slack:

2025-05-01 11:41:02.817 [ERR] RPCS: [/tapchannelrpc.TaprootAssetChannels/SendPayment]: error adding sell order: error marshalling sell order response: unknown AddAssetSellOrder event type: *rfq.PeerAcceptedBuyQuoteEvent

This happens when two orders of different types are added at more or
less the same time, then we receive the wrong one on the subscriber
channel.
So we need to filter by type and also by matching the order to the
returned quote event.

Author: guggero

rfq/manager.go

@@ -1116,6 +1116,24 @@ func (q *PeerAcceptedBuyQuoteEvent) Timestamp() time.Time {
 	return q.timestamp.UTC()
 }
 
+// MatchesOrder checks if the sell quote matches the provided order.
+func (q *PeerAcceptedBuyQuoteEvent) MatchesOrder(order BuyOrder) bool {
+	if q.Request.AssetSpecifier != order.AssetSpecifier {
+		return false
+	}
+
+	// If the order has no peer, we accept equality just based on the
+	// specifier.
+	if order.Peer.IsNone() {
+		return true
+	}
+
+	// If a peer is specified, ensure it matches the event's peer.
+	return fn.MapOptionZ(order.Peer, func(vertex route.Vertex) bool {
+		return q.Peer == vertex
+	})
+}
+
 // Ensure that the PeerAcceptedBuyQuoteEvent struct implements the Event
 // interface.
 var _ fn.Event = (*PeerAcceptedBuyQuoteEvent)(nil)
@@ -1197,6 +1215,24 @@ func (q *PeerAcceptedSellQuoteEvent) Timestamp() time.Time {
 	return q.timestamp.UTC()
 }
 
+// MatchesOrder checks if the sell quote matches the provided order.
+func (q *PeerAcceptedSellQuoteEvent) MatchesOrder(order SellOrder) bool {
+	if q.Request.AssetSpecifier != order.AssetSpecifier {
+		return false
+	}
+
+	// If the order has no peer, we accept equality just based on the
+	// specifier.
+	if order.Peer.IsNone() {
+		return true
+	}
+
+	// If a peer is specified, ensure it matches the event's peer.
+	return fn.MapOptionZ(order.Peer, func(vertex route.Vertex) bool {
+		return q.Peer == vertex
+	})
+}
+
 // Ensure that the PeerAcceptedSellQuoteEvent struct implements the Event
 // interface.
 var _ fn.Event = (*PeerAcceptedSellQuoteEvent)(nil)

rpcserver.go

@@ -6799,8 +6799,26 @@ func (r *rpcServer) AddAssetBuyOrder(ctx context.Context,
 	timeout := time.After(time.Second * time.Duration(req.TimeoutSeconds))
 
 	for {
+		type targetEventType = *rfq.PeerAcceptedBuyQuoteEvent
 		select {
 		case event := <-eventSubscriber.NewItemCreated.ChanOut():
+			acceptedQuote, ok := event.(targetEventType)
+			if !ok {
+				rpcsLog.Debugf("Received event of type %T "+
+					"but expected accepted sell quote, "+
+					"skipping", event)
+
+				continue
+			}
+
+			if !acceptedQuote.MatchesOrder(*buyOrder) {
+				rpcsLog.Debugf("Received event of type %T "+
+					"but order doesn't match, skipping",
+					event)
+
+				continue
+			}
+
 			resp, err := rfq.NewAddAssetBuyOrderResponse(event)
 			if err != nil {
 				return nil, fmt.Errorf("error marshalling "+
@@ -6973,8 +6991,26 @@ func (r *rpcServer) AddAssetSellOrder(ctx context.Context,
 	timeout := time.After(time.Second * time.Duration(req.TimeoutSeconds))
 
 	for {
+		type targetEventType = *rfq.PeerAcceptedSellQuoteEvent
 		select {
 		case event := <-eventSubscriber.NewItemCreated.ChanOut():
+			acceptedQuote, ok := event.(targetEventType)
+			if !ok {
+				rpcsLog.Debugf("Received event of type %T "+
+					"but expected accepted sell quote, "+
+					"skipping", event)
+
+				continue
+			}
+
+			if !acceptedQuote.MatchesOrder(*sellOrder) {
+				rpcsLog.Debugf("Received event of type %T "+
+					"but order doesn't match, skipping",
+					event)
+
+				continue
+			}
+
 			resp, err := rfq.NewAddAssetSellOrderResponse(event)
 			if err != nil {
 				return nil, fmt.Errorf("error marshalling "+