assets: harden tapd client lifecycle and RFQ handling

assets/client.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
+	"math"
+	"math/big"
 	"os"
 	"path/filepath"
 	"sync"
@@ -78,14 +80,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 +103,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 +146,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
@@ -163,12 +170,13 @@ func (c *TapdClient) GetRfqForAsset(ctx context.Context,
 func (c *TapdClient) GetAssetName(ctx context.Context,
 	assetId []byte) (string, error) {
 
-	c.assetNameMutex.Lock()
-	defer c.assetNameMutex.Unlock()
 	assetIdStr := hex.EncodeToString(assetId)
+	c.assetNameMutex.RLock()
 	if name, ok := c.assetNameCache[assetIdStr]; ok {
+		c.assetNameMutex.RUnlock()
 		return name, nil
 	}
+	c.assetNameMutex.RUnlock()
 
 	assetStats, err := c.UniverseClient.QueryAssetStats(
 		ctx, &universerpc.AssetStatsQuery{
@@ -192,7 +200,9 @@ func (c *TapdClient) GetAssetName(ctx context.Context,
 		assetName = assetStats.AssetStats[0].Asset.AssetName
 	}
 
+	c.assetNameMutex.Lock()
 	c.assetNameCache[assetIdStr] = assetName
+	c.assetNameMutex.Unlock()
 
 	return assetName, nil
 }
@@ -220,7 +230,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,6 +264,19 @@ func (c *TapdClient) GetAssetPrice(ctx context.Context, assetID string,
 func getSatsFromAssetAmt(assetAmt uint64, assetRate *rfqrpc.FixedPoint) (
 	btcutil.Amount, error) {
 
+	if assetRate == nil {
+		return 0, fmt.Errorf("asset rate cannot be nil")
+	}
+
+	coefficient, ok := new(big.Int).SetString(assetRate.Coefficient, 10)
+	if !ok {
+		return 0, fmt.Errorf("invalid asset rate coefficient: %q",
+			assetRate.Coefficient)
+	}
+	if coefficient.Sign() <= 0 {
+		return 0, fmt.Errorf("asset rate coefficient must be positive")
+	}
+
 	rateFP, err := rpcutils.UnmarshalRfqFixedPoint(assetRate)
 	if err != nil {
 		return 0, fmt.Errorf("cannot unmarshal asset rate: %w", err)
@@ -288,6 +311,26 @@ func getPaymentMaxAmount(satAmount btcutil.Amount, feeLimitMultiplier float64) (
 	)
 }
 
+// getRfqTimeoutSeconds converts the configured RFQ timeout to the whole
+// seconds accepted by tapd. Sub-second precision is rounded up so a positive
+// timeout can never become tapd's invalid zero value.
+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, "")

assets/client_test.go

@@ -1,20 +1,51 @@
 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
+}
+
 // TestDefaultTapdConfig tests that the default tapd connection paths match
 // tapd's mainnet defaults.
 func TestDefaultTapdConfig(t *testing.T) {
@@ -82,6 +113,61 @@ 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)
+}
+
 func TestGetPaymentMaxAmount(t *testing.T) {
 	tests := []struct {
 		satAmount          btcutil.Amount
@@ -141,6 +227,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 +308,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 {

loopd/daemon.go

@@ -166,13 +166,23 @@ func (d *Daemon) Start() error {
 	// and we can just return the error.
 	err = d.initialize(true)
 	if errors.Is(err, bbolterrors.ErrTimeout) {
+		if d.assetClient != nil {
+			d.assetClient.Close()
+			d.assetClient = nil
+		}
+
 		// We're trying to be started as a standalone Loop daemon, most
 		// likely LiT is already running and blocking the DB
 		return fmt.Errorf("%v: make sure no other loop daemon process "+
 			"(standalone or embedded in lightning-terminal) is"+
 			"running", err)
 	}
 	if err != nil {
+		if d.assetClient != nil {
+			d.assetClient.Close()
+			d.assetClient = nil
+		}
+
 		return err
 	}
 
@@ -1145,6 +1155,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()

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(