rfq: add support for quote policy persistence and restoration

Author: darioAnongba

rfq/manager.go

@@ -118,6 +118,9 @@ type ManagerCfg struct {
 	// helps us communicate custom feature bits with our peer.
 	AuxChanNegotiator *tapfeatures.AuxChannelNegotiator
 
+	// PolicyStore provides persistence for agreed RFQ policies.
+	PolicyStore PolicyStore
+
 	// AcceptPriceDeviationPpm is the price deviation in
 	// parts per million that is accepted by the RFQ negotiator.
 	//
@@ -264,12 +267,18 @@ func (m *Manager) startSubsystems(ctx context.Context) error {
 		SpecifierChecker:  m.AssetMatchesSpecifier,
 		NoOpHTLCs:         m.cfg.NoOpHTLCs,
 		AuxChanNegotiator: m.cfg.AuxChanNegotiator,
+		PolicyStore:       m.cfg.PolicyStore,
 	})
 	if err != nil {
 		return fmt.Errorf("error initializing RFQ order handler: %w",
 			err)
 	}
 
+	if err := m.restorePersistedPolicies(ctx); err != nil {
+		return fmt.Errorf("error restoring persisted RFQ "+
+			"policies: %w", err)
+	}
+
 	if err := m.orderHandler.Start(); err != nil {
 		return fmt.Errorf("unable to start RFQ order handler: %w", err)
 	}
@@ -313,6 +322,29 @@ func (m *Manager) startSubsystems(ctx context.Context) error {
 	return err
 }
 
+// restorePersistedPolicies loads persisted RFQ policies from the store
+// and rehydrates the order handler and local cache state.
+func (m *Manager) restorePersistedPolicies(ctx context.Context) error {
+	buyAccepts, sellAccepts, err := m.cfg.PolicyStore.FetchActivePolicies(
+		ctx,
+	)
+	if err != nil {
+		return fmt.Errorf("error fetching persisted policies: %w", err)
+	}
+
+	m.orderHandler.restorePersistedPolicies(buyAccepts, sellAccepts)
+
+	for _, accept := range buyAccepts {
+		m.localAcceptedBuyQuotes.Store(accept.ShortChannelId(), accept)
+	}
+
+	for _, accept := range sellAccepts {
+		m.localAcceptedSellQuotes.Store(accept.ShortChannelId(), accept)
+	}
+
+	return nil
+}
+
 // handleError logs an error and sends it to the main server error channel if
 // it is a critical error.
 func (m *Manager) handleError(err error) {

rfq/manager_test.go

@@ -4,13 +4,15 @@ import (
 	"context"
 	"encoding/binary"
 	"testing"
+	"time"
 
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/decred/dcrd/dcrec/secp256k1/v4"
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/taproot-assets/address"
 	"github.com/lightninglabs/taproot-assets/asset"
 	"github.com/lightninglabs/taproot-assets/proof"
+	"github.com/lightninglabs/taproot-assets/rfqmsg"
 	tpchmsg "github.com/lightninglabs/taproot-assets/tapchannelmsg"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/routing/route"
@@ -59,6 +61,32 @@ var (
 	peer2        = route.Vertex{77}
 )
 
+type nopPolicyStore struct{}
+
+func (nopPolicyStore) StoreSalePolicy(context.Context,
+	rfqmsg.BuyAccept) error {
+
+	return nil
+}
+
+func (nopPolicyStore) StorePurchasePolicy(context.Context,
+	rfqmsg.SellAccept) error {
+
+	return nil
+}
+
+func (nopPolicyStore) FetchActivePolicies(context.Context) (
+	[]rfqmsg.BuyAccept, []rfqmsg.SellAccept, error) {
+
+	return nil, nil, nil
+}
+
+func (nopPolicyStore) DeactivatePolicy(context.Context,
+	rfqmsg.ID, time.Time) error {
+
+	return nil
+}
+
 // GroupLookupMock mocks the GroupLookup interface that is required by the
 // rfq manager to check asset IDs against asset specifiers.
 type GroupLookupMock struct{}
@@ -141,6 +169,7 @@ func assertComputeChannelAssetBalance(t *testing.T,
 	mockGroupLookup := &GroupLookupMock{}
 	cfg := ManagerCfg{
 		GroupLookup: mockGroupLookup,
+		PolicyStore: nopPolicyStore{},
 	}
 	manager, err := NewManager(cfg)
 	require.NoError(t, err)

rfq/order.go

@@ -88,6 +88,9 @@ type Policy interface {
 	GenerateInterceptorResponse(
 		lndclient.InterceptedHtlc) (*lndclient.InterceptedHtlcResponse,
 		error)
+
+	// QuoteID returns the RFQ ID if it exists, otherwise false.
+	QuoteID() (rfqmsg.ID, bool)
 }
 
 // AssetSalePolicy is a struct that holds the terms which determine whether an
@@ -325,6 +328,11 @@ func (c *AssetSalePolicy) GenerateInterceptorResponse(
 	}, nil
 }
 
+// QuoteID returns the RFQ identifier that originated this policy.
+func (c *AssetSalePolicy) QuoteID() (rfqmsg.ID, bool) {
+	return c.AcceptedQuoteId, true
+}
+
 // Ensure that AssetSalePolicy implements the Policy interface.
 var _ Policy = (*AssetSalePolicy)(nil)
 
@@ -537,6 +545,11 @@ func (c *AssetPurchasePolicy) GenerateInterceptorResponse(
 	}, nil
 }
 
+// QuoteID returns the RFQ identifier that originated this policy.
+func (c *AssetPurchasePolicy) QuoteID() (rfqmsg.ID, bool) {
+	return c.AcceptedQuoteId, true
+}
+
 // Ensure that AssetPurchasePolicy implements the Policy interface.
 var _ Policy = (*AssetPurchasePolicy)(nil)
 
@@ -674,6 +687,13 @@ func (a *AssetForwardPolicy) GenerateInterceptorResponse(
 	}, nil
 }
 
+// QuoteID returns the RFQ identifier that originated this policy.
+//
+// Forward policies do not map to a single quote, hence we return false.
+func (a *AssetForwardPolicy) QuoteID() (rfqmsg.ID, bool) {
+	return rfqmsg.ID{}, false
+}
+
 // Ensure that AssetForwardPolicy implements the Policy interface.
 var _ Policy = (*AssetForwardPolicy)(nil)
 
@@ -707,6 +727,9 @@ type OrderHandlerCfg struct {
 	// that is encapsulated in the init and reestablish peer messages. This
 	// helps us communicate custom feature bits with our peer.
 	AuxChanNegotiator *tapfeatures.AuxChannelNegotiator
+
+	// PolicyStore persists agreed RFQ policies.
+	PolicyStore PolicyStore
 }
 
 // OrderHandler orchestrates management of accepted quote bundles. It monitors
@@ -719,6 +742,9 @@ type OrderHandler struct {
 	// cfg holds the configuration parameters for the RFQ order handler.
 	cfg OrderHandlerCfg
 
+	// policyStore provides persistence for agreed policies.
+	policyStore PolicyStore
+
 	// policies is a map of serialised short channel IDs (SCIDs) to
 	// associated asset transaction policies.
 	policies lnutils.SyncMap[SerialisedScid, Policy]
@@ -736,8 +762,9 @@ type OrderHandler struct {
 // NewOrderHandler creates a new struct instance.
 func NewOrderHandler(cfg OrderHandlerCfg) (*OrderHandler, error) {
 	return &OrderHandler{
-		cfg:      cfg,
-		policies: lnutils.SyncMap[SerialisedScid, Policy]{},
+		cfg:         cfg,
+		policyStore: cfg.PolicyStore,
+		policies:    lnutils.SyncMap[SerialisedScid, Policy]{},
 		ContextGuard: &fn.ContextGuard{
 			DefaultTimeout: DefaultTimeout,
 			Quit:           make(chan struct{}),
@@ -965,6 +992,11 @@ func (h *OrderHandler) RegisterAssetSalePolicy(buyAccept rfqmsg.BuyAccept) {
 		buyAccept, h.cfg.NoOpHTLCs, h.cfg.AuxChanNegotiator,
 	)
 
+	if err := h.storeSalePolicy(buyAccept); err != nil {
+		log.Errorf("Unable to persist asset sale policy (id=%x): %v",
+			buyAccept.ID[:], err)
+	}
+
 	h.policies.Store(policy.AcceptedQuoteId.Scid(), policy)
 }
 
@@ -978,9 +1010,72 @@ func (h *OrderHandler) RegisterAssetPurchasePolicy(
 		"sell accept message: %s", sellAccept.String())
 
 	policy := NewAssetPurchasePolicy(sellAccept)
+
+	if err := h.storePurchasePolicy(sellAccept); err != nil {
+		log.Errorf("Unable to persist asset purchase policy "+
+			"(id=%x): %v", sellAccept.ID[:], err)
+	}
+
 	h.policies.Store(policy.scid, policy)
 }
 
+func (h *OrderHandler) storeSalePolicy(buyAccept rfqmsg.BuyAccept) error {
+	ctx, cancel := h.WithCtxQuit()
+	defer cancel()
+
+	return h.policyStore.StoreSalePolicy(ctx, buyAccept)
+}
+
+func (h *OrderHandler) storePurchasePolicy(
+	sellAccept rfqmsg.SellAccept) error {
+
+	ctx, cancel := h.WithCtxQuit()
+	defer cancel()
+
+	return h.policyStore.StorePurchasePolicy(ctx, sellAccept)
+}
+
+func (h *OrderHandler) restorePersistedPolicies(buyAccepts []rfqmsg.BuyAccept,
+	sellAccepts []rfqmsg.SellAccept) {
+
+	for _, accept := range buyAccepts {
+		policy := NewAssetSalePolicy(
+			accept, h.cfg.NoOpHTLCs, h.cfg.AuxChanNegotiator,
+		)
+		h.policies.Store(policy.AcceptedQuoteId.Scid(), policy)
+	}
+
+	for _, accept := range sellAccepts {
+		policy := NewAssetPurchasePolicy(accept)
+		h.policies.Store(policy.scid, policy)
+	}
+}
+
+func (h *OrderHandler) removePolicy(scid SerialisedScid,
+	policy Policy) {
+
+	h.policies.Delete(scid)
+	h.deactivatePersistedPolicy(policy)
+}
+
+func (h *OrderHandler) deactivatePersistedPolicy(policy Policy) {
+	id, ok := policy.QuoteID()
+	if !ok {
+		return
+	}
+
+	ctx, cancel := h.WithCtxQuit()
+	defer cancel()
+
+	err := h.policyStore.DeactivatePolicy(
+		ctx, id, time.Now().UTC(),
+	)
+	if err != nil {
+		log.Errorf("Unable to deactivate persisted policy id=%x: %v",
+			id[:], err)
+	}
+}
+
 // fetchPolicy fetches a policy which is relevant to a given HTLC. If a policy
 // is not found, false is returned. Expired policies are not returned and are
 // removed from the cache.
@@ -1049,12 +1144,12 @@ func (h *OrderHandler) fetchPolicy(htlc lndclient.InterceptedHtlc) (Policy,
 		outgoingPolicy := outPolicy
 
 		if incomingPolicy.HasExpired() {
-			scid := incomingPolicy.Scid()
-			h.policies.Delete(SerialisedScid(scid))
+			inScid := SerialisedScid(incomingPolicy.Scid())
+			h.removePolicy(inScid, incomingPolicy)
 		}
 		if outgoingPolicy.HasExpired() {
-			scid := outgoingPolicy.Scid()
-			h.policies.Delete(SerialisedScid(scid))
+			outScid := SerialisedScid(outgoingPolicy.Scid())
+			h.removePolicy(outScid, outgoingPolicy)
 		}
 
 		// If either the incoming or outgoing policy has expired, we
@@ -1098,7 +1193,7 @@ func (h *OrderHandler) fetchPolicy(htlc lndclient.InterceptedHtlc) (Policy,
 	scid := *foundScid
 
 	if policy.HasExpired() {
-		h.policies.Delete(scid)
+		h.removePolicy(scid, policy)
 		return nil, false, nil
 	}
 
@@ -1114,7 +1209,7 @@ func (h *OrderHandler) cleanupStalePolicies() {
 		func(scid SerialisedScid, policy Policy) error {
 			if policy.HasExpired() {
 				staleCounter++
-				h.policies.Delete(scid)
+				h.removePolicy(scid, policy)
 			}
 
 			return nil

rfq/policystore.go

@@ -0,0 +1,24 @@
+package rfq
+
+import (
+	"context"
+	"time"
+
+	"github.com/lightninglabs/taproot-assets/rfqmsg"
+)
+
+// PolicyStore abstracts persistence of RFQ policies.
+type PolicyStore interface {
+	// StoreSalePolicy stores an asset sale policy.
+	StoreSalePolicy(ctx context.Context, accept rfqmsg.BuyAccept) error
+
+	// StorePurchasePolicy stores an asset purchase policy.
+	StorePurchasePolicy(ctx context.Context, accept rfqmsg.SellAccept) error
+
+	// FetchActivePolicies fetches all active asset policies.
+	FetchActivePolicies(ctx context.Context) ([]rfqmsg.BuyAccept,
+		[]rfqmsg.SellAccept, error)
+
+	// DeactivatePolicy deactivates an asset sale or purchase policy.
+	DeactivatePolicy(ctx context.Context, id rfqmsg.ID, at time.Time) error
+}