diff --git a/assets/client.go b/assets/client.go index 6b8b514f7..7eef98b03 100644 --- a/assets/client.go +++ b/assets/client.go @@ -4,6 +4,8 @@ import ( "context" "encoding/hex" "fmt" + "math" + "math/big" "os" "path/filepath" "sync" @@ -11,7 +13,6 @@ import ( "github.com/btcsuite/btcd/btcutil" "github.com/lightninglabs/taproot-assets/rfqmath" - "github.com/lightninglabs/taproot-assets/rpcutils" "github.com/lightninglabs/taproot-assets/taprpc" "github.com/lightninglabs/taproot-assets/taprpc/priceoraclerpc" "github.com/lightninglabs/taproot-assets/taprpc/rfqrpc" @@ -78,14 +79,19 @@ type TapdClient struct { rfqrpc.RfqClient universerpc.UniverseClient - cfg *TapdConfig - assetNameCache map[string]string - assetNameMutex sync.Mutex - cc *grpc.ClientConn + rfqTimeoutSeconds uint32 + assetNameCache map[string]string + assetNameMutex sync.RWMutex + cc *grpc.ClientConn } // NewTapdClient returns a new taproot assets client. func NewTapdClient(config *TapdConfig) (*TapdClient, error) { + rfqTimeoutSeconds, err := getRfqTimeoutSeconds(config.RFQtimeout) + if err != nil { + return nil, err + } + // Create the client connection to the server. conn, err := getClientConn(config) if err != nil { @@ -96,7 +102,7 @@ func NewTapdClient(config *TapdConfig) (*TapdClient, error) { client := &TapdClient{ assetNameCache: make(map[string]string), cc: conn, - cfg: config, + rfqTimeoutSeconds: rfqTimeoutSeconds, TaprootAssetsClient: taprpc.NewTaprootAssetsClient(conn), TaprootAssetChannelsClient: tapchannelrpc.NewTaprootAssetChannelsClient(conn), PriceOracleClient: priceoraclerpc.NewPriceOracleClient(conn), @@ -139,7 +145,7 @@ func (c *TapdClient) GetRfqForAsset(ctx context.Context, PeerPubKey: peerPubkey, PaymentMaxAmt: uint64(paymentMaxAmt), Expiry: uint64(expiry), - TimeoutSeconds: uint32(c.cfg.RFQtimeout.Seconds()), + TimeoutSeconds: c.rfqTimeoutSeconds, }) if err != nil { return nil, err @@ -152,21 +158,26 @@ func (c *TapdClient) GetRfqForAsset(ctx context.Context, rfq.GetRejectedQuote()) } - if rfq.GetAcceptedQuote() != nil { - return rfq.GetAcceptedQuote(), nil + acceptedQuote := rfq.GetAcceptedQuote() + if acceptedQuote == nil { + return nil, fmt.Errorf("no accepted quote") } - return nil, fmt.Errorf("no accepted quote") + _, err = unmarshalAssetRate(acceptedQuote.BidAssetRate) + if err != nil { + return nil, fmt.Errorf("invalid accepted quote asset rate: %w", + err) + } + + return acceptedQuote, nil } // GetAssetName returns the human-readable name of the asset. func (c *TapdClient) GetAssetName(ctx context.Context, assetId []byte) (string, error) { - c.assetNameMutex.Lock() - defer c.assetNameMutex.Unlock() assetIdStr := hex.EncodeToString(assetId) - if name, ok := c.assetNameCache[assetIdStr]; ok { + if name, ok := c.getCachedAssetName(assetIdStr); ok { return name, nil } @@ -192,11 +203,28 @@ func (c *TapdClient) GetAssetName(ctx context.Context, assetName = assetStats.AssetStats[0].Asset.AssetName } - c.assetNameCache[assetIdStr] = assetName + c.cacheAssetName(assetIdStr, assetName) return assetName, nil } +// getCachedAssetName returns an asset name from the cache. +func (c *TapdClient) getCachedAssetName(assetID string) (string, bool) { + c.assetNameMutex.RLock() + defer c.assetNameMutex.RUnlock() + + name, ok := c.assetNameCache[assetID] + return name, ok +} + +// cacheAssetName adds an asset name to the cache. +func (c *TapdClient) cacheAssetName(assetID, name string) { + c.assetNameMutex.Lock() + defer c.assetNameMutex.Unlock() + + c.assetNameCache[assetID] = name +} + // GetAssetPrice returns the price of an asset in satoshis. NOTE: this currently // uses the rfq process for the asset price. A future implementation should // use a price oracle to not spam a peer. @@ -220,7 +248,7 @@ func (c *TapdClient) GetAssetPrice(ctx context.Context, assetID string, }, PaymentMaxAmt: uint64(msatAmt), Expiry: uint64(rfqExpiry), - TimeoutSeconds: uint32(c.cfg.RFQtimeout.Seconds()), + TimeoutSeconds: c.rfqTimeoutSeconds, PeerPubKey: peerPubkey, }) if err != nil { @@ -254,7 +282,7 @@ func (c *TapdClient) GetAssetPrice(ctx context.Context, assetID string, func getSatsFromAssetAmt(assetAmt uint64, assetRate *rfqrpc.FixedPoint) ( btcutil.Amount, error) { - rateFP, err := rpcutils.UnmarshalRfqFixedPoint(assetRate) + rateFP, err := unmarshalAssetRate(assetRate) if err != nil { return 0, fmt.Errorf("cannot unmarshal asset rate: %w", err) } @@ -266,6 +294,33 @@ func getSatsFromAssetAmt(assetAmt uint64, assetRate *rfqrpc.FixedPoint) ( return msatAmt.ToSatoshis(), nil } +// unmarshalAssetRate validates and converts an RPC asset rate to the fixed +// point representation used for RFQ arithmetic. +func unmarshalAssetRate(assetRate *rfqrpc.FixedPoint) ( + *rfqmath.BigIntFixedPoint, error) { + + if assetRate == nil { + return nil, fmt.Errorf("asset rate cannot be nil") + } + if assetRate.Scale > math.MaxUint8 { + return nil, fmt.Errorf("scale value overflow: %v", assetRate.Scale) + } + + coefficient, ok := new(big.Int).SetString(assetRate.Coefficient, 10) + if !ok { + return nil, fmt.Errorf("invalid asset rate coefficient: %q", + assetRate.Coefficient) + } + if coefficient.Sign() <= 0 { + return nil, fmt.Errorf("asset rate coefficient must be positive") + } + + return &rfqmath.BigIntFixedPoint{ + Coefficient: rfqmath.NewBigInt(coefficient), + Scale: uint8(assetRate.Scale), + }, nil +} + // getPaymentMaxAmount returns the milisat amount we are willing to pay for the // payment. func getPaymentMaxAmount(satAmount btcutil.Amount, feeLimitMultiplier float64) ( @@ -288,6 +343,26 @@ func getPaymentMaxAmount(satAmount btcutil.Amount, feeLimitMultiplier float64) ( ) } +// getRfqTimeoutSeconds converts the configured RFQ timeout to the whole +// seconds accepted by tapd. Fractional seconds are rounded up so tapd's +// timeout is never shorter than the configured duration. +func getRfqTimeoutSeconds(timeout time.Duration) (uint32, error) { + if timeout <= 0 { + return 0, fmt.Errorf("RFQ timeout must be greater than zero") + } + + seconds := timeout / time.Second + if timeout%time.Second != 0 { + seconds++ + } + if seconds > time.Duration(math.MaxUint32) { + return 0, fmt.Errorf("RFQ timeout exceeds maximum of %v seconds", + uint64(math.MaxUint32)) + } + + return uint32(seconds), nil +} + func getClientConn(config *TapdConfig) (*grpc.ClientConn, error) { // Load the specified TLS certificate and build transport credentials. creds, err := credentials.NewClientTLSFromFile(config.TLSPath, "") diff --git a/assets/client_test.go b/assets/client_test.go index 8fa79092d..5941ae44d 100644 --- a/assets/client_test.go +++ b/assets/client_test.go @@ -1,20 +1,64 @@ package assets import ( + "context" + "encoding/hex" "encoding/pem" + "math" "net/http" "net/http/httptest" "os" "path/filepath" "testing" + "time" "github.com/btcsuite/btcd/btcutil" "github.com/lightninglabs/taproot-assets/taprpc/rfqrpc" + "github.com/lightninglabs/taproot-assets/taprpc/universerpc" "github.com/lightningnetwork/lnd/lnwire" "github.com/stretchr/testify/require" + "google.golang.org/grpc" "gopkg.in/macaroon.v2" ) +type blockingUniverseClient struct { + universerpc.UniverseClient + + queryStarted chan struct{} + releaseQuery chan struct{} +} + +func (b *blockingUniverseClient) QueryAssetStats(context.Context, + *universerpc.AssetStatsQuery, ...grpc.CallOption) ( + *universerpc.UniverseAssetStats, error) { + + close(b.queryStarted) + <-b.releaseQuery + + return &universerpc.UniverseAssetStats{ + AssetStats: []*universerpc.AssetStatsSnapshot{ + { + Asset: &universerpc.AssetStatsAsset{ + AssetName: "queried asset", + }, + }, + }, + }, nil +} + +type staticRfqClient struct { + rfqrpc.RfqClient + + response *rfqrpc.AddAssetSellOrderResponse +} + +func (s *staticRfqClient) AddAssetSellOrder(context.Context, + *rfqrpc.AddAssetSellOrderRequest, ...grpc.CallOption) ( + *rfqrpc.AddAssetSellOrderResponse, error) { + + return s.response, nil +} + // TestDefaultTapdConfig tests that the default tapd connection paths match // tapd's mainnet defaults. func TestDefaultTapdConfig(t *testing.T) { @@ -82,6 +126,144 @@ func TestTapdConfigClientConn(t *testing.T) { ) } +// TestGetAssetNameCachedLookupNotBlocked verifies that a slow universe query +// for one asset does not prevent another caller from reading a cached name. +func TestGetAssetNameCachedLookupNotBlocked(t *testing.T) { + const cachedName = "cached asset" + + cachedAssetID := []byte{1} + queryStarted := make(chan struct{}) + releaseQuery := make(chan struct{}) + client := &TapdClient{ + UniverseClient: &blockingUniverseClient{ + queryStarted: queryStarted, + releaseQuery: releaseQuery, + }, + assetNameCache: map[string]string{ + hex.EncodeToString(cachedAssetID): cachedName, + }, + } + + queryResult := make(chan error, 1) + go func() { + _, err := client.GetAssetName(context.Background(), []byte{2}) + queryResult <- err + }() + + select { + case <-queryStarted: + case <-time.After(time.Second): + t.Fatal("universe query did not start") + } + + type nameResult struct { + name string + err error + } + cachedResult := make(chan nameResult, 1) + go func() { + name, err := client.GetAssetName( + context.Background(), cachedAssetID, + ) + cachedResult <- nameResult{name: name, err: err} + }() + + select { + case result := <-cachedResult: + require.NoError(t, result.err) + require.Equal(t, cachedName, result.name) + case <-time.After(time.Second): + close(releaseQuery) + t.Fatal("cached lookup blocked behind universe query") + } + + close(releaseQuery) + require.NoError(t, <-queryResult) +} + +// TestGetRfqForAssetValidatesRate verifies that malformed accepted quote rates +// are rejected before they reach downstream RFQ arithmetic. +func TestGetRfqForAssetValidatesRate(t *testing.T) { + tests := []struct { + name string + assetRate *rfqrpc.FixedPoint + expectError bool + }{ + { + name: "valid", + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "100000", Scale: 0, + }, + }, + { + name: "nil", + assetRate: nil, + expectError: true, + }, + { + name: "malformed coefficient", + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "not-a-number", Scale: 0, + }, + expectError: true, + }, + { + name: "zero coefficient", + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "0", Scale: 0, + }, + expectError: true, + }, + { + name: "negative coefficient", + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "-1", Scale: 0, + }, + expectError: true, + }, + { + name: "scale overflow", + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "1", Scale: 256, + }, + expectError: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + acceptedQuote := &rfqrpc.PeerAcceptedSellQuote{ + BidAssetRate: test.assetRate, + } + acceptedResponse := + &rfqrpc.AddAssetSellOrderResponse_AcceptedQuote{ + AcceptedQuote: acceptedQuote, + } + client := &TapdClient{ + RfqClient: &staticRfqClient{ + response: &rfqrpc.AddAssetSellOrderResponse{ + Response: acceptedResponse, + }, + }, + rfqTimeoutSeconds: 60, + } + + quote, err := client.GetRfqForAsset( + context.Background(), 1000, []byte{1}, []byte{2}, + time.Now().Add(time.Minute).Unix(), 1, + ) + if test.expectError { + require.Error(t, err) + require.Nil(t, quote) + return + } + + require.NoError(t, err) + require.Same(t, acceptedQuote, quote) + }) + } +} + func TestGetPaymentMaxAmount(t *testing.T) { tests := []struct { satAmount btcutil.Amount @@ -141,6 +323,62 @@ func TestGetPaymentMaxAmount(t *testing.T) { } } +// TestGetRfqTimeoutSeconds verifies that configured durations are safely +// converted to tapd's whole-second timeout field. +func TestGetRfqTimeoutSeconds(t *testing.T) { + tests := []struct { + name string + timeout time.Duration + expectedSeconds uint32 + expectError bool + }{ + { + name: "whole seconds", + timeout: 60 * time.Second, + expectedSeconds: 60, + }, + { + name: "sub-second rounded up", + timeout: time.Millisecond, + expectedSeconds: 1, + }, + { + name: "fractional second rounded up", + timeout: time.Second + time.Nanosecond, + expectedSeconds: 2, + }, + { + name: "zero", + timeout: 0, + expectError: true, + }, + { + name: "negative", + timeout: -time.Second, + expectError: true, + }, + { + name: "overflow", + timeout: time.Duration(math.MaxUint32)*time.Second + + time.Nanosecond, + expectError: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + seconds, err := getRfqTimeoutSeconds(test.timeout) + if test.expectError { + require.Error(t, err) + return + } + + require.NoError(t, err) + require.Equal(t, test.expectedSeconds, seconds) + }) + } +} + func TestGetSatsFromAssetAmt(t *testing.T) { tests := []struct { assetAmt uint64 @@ -166,6 +404,39 @@ func TestGetSatsFromAssetAmt(t *testing.T) { expected: btcutil.Amount(0), expectError: false, }, + { + assetAmt: 1000, + assetRate: nil, + expectError: true, + }, + { + assetAmt: 1000, + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "not-a-number", Scale: 0, + }, + expectError: true, + }, + { + assetAmt: 1000, + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "0", Scale: 0, + }, + expectError: true, + }, + { + assetAmt: 1000, + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "-1", Scale: 0, + }, + expectError: true, + }, + { + assetAmt: 1000, + assetRate: &rfqrpc.FixedPoint{ + Coefficient: "1", Scale: 256, + }, + expectError: true, + }, } for _, test := range tests { diff --git a/docs/release-notes/release-notes-next.md b/docs/release-notes/release-notes-next.md index 4de57fdb4..7c9d492e5 100644 --- a/docs/release-notes/release-notes-next.md +++ b/docs/release-notes/release-notes-next.md @@ -6,6 +6,11 @@ #### Bug Fixes +* Taproot Asset Loop Out handling now validates RFQ timeouts and asset rates, + keeps cached asset-name lookups responsive during slow `tapd` queries, and + closes `tapd` connections cleanly during shutdown and startup failures. + [PR #1189](https://github.com/lightninglabs/loop/pull/1189) + #### Maintenance #### Contributors (Alphabetical Order) diff --git a/loopd/daemon.go b/loopd/daemon.go index f625aa270..319709d84 100644 --- a/loopd/daemon.go +++ b/loopd/daemon.go @@ -158,6 +158,15 @@ func (d *Daemon) Start() error { if err != nil { return err } + + defer func() { + if err == nil || d.assetClient == nil { + return + } + + d.assetClient.Close() + d.assetClient = nil + }() } // With lnd connected, initialize everything else, such as the swap @@ -178,15 +187,15 @@ func (d *Daemon) Start() error { // If we get here, we already have started several goroutines. So if // anything goes wrong now, we need to cleanly shut down again. - startErr := d.startWebServers() - if startErr != nil { - errorf("Error while starting daemon: %v", startErr) + err = d.startWebServers() + if err != nil { + errorf("Error while starting daemon: %v", err) d.Stop() stopErr := <-d.ErrChan if stopErr != nil { errorf("Error while stopping daemon: %v", stopErr) } - return startErr + return err } return nil @@ -1145,6 +1154,10 @@ func (d *Daemon) stop() { if d.clientCleanup != nil { d.clientCleanup() } + if d.assetClient != nil { + d.assetClient.Close() + d.assetClient = nil + } // Everything should be shutting down now, wait for completion. d.wg.Wait() diff --git a/loopd/view.go b/loopd/view.go index 73b401d86..86b9604a1 100644 --- a/loopd/view.go +++ b/loopd/view.go @@ -44,6 +44,7 @@ func view(config *Config, lisCfg *ListenerCfg) error { if err != nil { return err } + defer assetClient.Close() } swapClient, cleanup, err := getClient(