Merge pull request #743 from bhandras/loop-out-timeout

loopout: configurable payment timeout for off-chain payments

Author: bhandras

cmd/loop/loopout.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"math"
 	"strconv"
 	"strings"
 	"time"
@@ -92,6 +93,14 @@ var loopOutCommand = cli.Command{
 				"Not setting this flag therefore might " +
 				"result in a lower swap fee",
 		},
+		cli.DurationFlag{
+			Name: "payment_timeout",
+			Usage: "the timeout for each individual off-chain " +
+				"payment attempt. If not set, the default " +
+				"timeout of 1 hour will be used. As the " +
+				"payment might be retried, the actual total " +
+				"time may be longer",
+		},
 		forceFlag,
 		labelFlag,
 		verboseFlag,
@@ -235,6 +244,25 @@ func loopOut(ctx *cli.Context) error {
 		}
 	}
 
+	var paymentTimeout int64
+	if ctx.IsSet("payment_timeout") {
+		parsedTimeout := ctx.Duration("payment_timeout")
+		if parsedTimeout.Truncate(time.Second) != parsedTimeout {
+			return fmt.Errorf("payment timeout must be a " +
+				"whole number of seconds")
+		}
+
+		paymentTimeout = int64(parsedTimeout.Seconds())
+		if paymentTimeout <= 0 {
+			return fmt.Errorf("payment timeout must be a " +
+				"positive value")
+		}
+
+		if paymentTimeout > math.MaxUint32 {
+			return fmt.Errorf("payment timeout is too large")
+		}
+	}
+
 	resp, err := client.LoopOut(context.Background(), &looprpc.LoopOutRequest{
 		Amt:                     int64(amt),
 		Dest:                    destAddr,
@@ -252,6 +280,7 @@ func loopOut(ctx *cli.Context) error {
 		SwapPublicationDeadline: uint64(swapDeadline.Unix()),
 		Label:                   label,
 		Initiator:               defaultInitiator,
+		PaymentTimeout:          uint32(paymentTimeout),
 	})
 	if err != nil {
 		return err

interface.go

@@ -92,6 +92,12 @@ type OutRequest struct {
 	// initiated the swap (loop CLI, autolooper, LiT UI and so on) and is
 	// appended to the user agent string.
 	Initiator string
+
+	// PaymentTimeout specifies the payment timeout for the individual
+	// off-chain payments. As the swap payment may be retried (depending on
+	// the configured maximum payment timeout) the total time spent may be
+	// a multiple of this value.
+	PaymentTimeout time.Duration
 }
 
 // Out contains the full details of a loop out request. This includes things

loopd/swapclient_server.go

@@ -101,6 +101,17 @@ func (s *swapClientServer) LoopOut(ctx context.Context,
 
 	log.Infof("Loop out request received")
 
+	// Note that LoopOutRequest.PaymentTimeout is unsigned and therefore
+	// cannot be negative.
+	paymentTimeout := time.Duration(in.PaymentTimeout) * time.Second
+
+	// Make sure we don't exceed the total allowed payment timeout.
+	if paymentTimeout > s.config.TotalPaymentTimeout {
+		return nil, fmt.Errorf("payment timeout %v exceeds maximum "+
+			"allowed timeout of %v", paymentTimeout,
+			s.config.TotalPaymentTimeout)
+	}
+
 	var sweepAddr btcutil.Address
 	var isExternalAddr bool
 	var err error
@@ -184,6 +195,7 @@ func (s *swapClientServer) LoopOut(ctx context.Context,
 		SwapPublicationDeadline: publicationDeadline,
 		Label:                   in.Label,
 		Initiator:               in.Initiator,
+		PaymentTimeout:          paymentTimeout,
 	}
 
 	switch {

loopdb/loopout.go

@@ -61,6 +61,10 @@ type LoopOutContract struct {
 	// allow the server to delay the publication in exchange for possibly
 	// lower fees.
 	SwapPublicationDeadline time.Time
+
+	// PaymentTimeout is the timeout for any individual off-chain payment
+	// attempt.
+	PaymentTimeout time.Duration
 }
 
 // ChannelSet stores a set of channels.

loopdb/sql_store.go

@@ -443,6 +443,7 @@ func loopOutToInsertArgs(hash lntypes.Hash,
 		PrepayInvoice:       loopOut.PrepayInvoice,
 		MaxPrepayRoutingFee: int64(loopOut.MaxPrepayRoutingFee),
 		PublicationDeadline: loopOut.SwapPublicationDeadline.UTC(),
+		PaymentTimeout:      int32(loopOut.PaymentTimeout.Seconds()),
 	}
 }
 
@@ -536,6 +537,9 @@ func ConvertLoopOutRow(network *chaincfg.Params, row sqlc.GetLoopOutSwapRow,
 			PrepayInvoice:           row.PrepayInvoice,
 			MaxPrepayRoutingFee:     btcutil.Amount(row.MaxPrepayRoutingFee),
 			SwapPublicationDeadline: row.PublicationDeadline,
+			PaymentTimeout: time.Duration(
+				row.PaymentTimeout,
+			) * time.Second,
 		},
 		Loop: Loop{
 			Hash: swapHash,

loopdb/sql_test.go

@@ -60,6 +60,7 @@ func TestSqliteLoopOutStore(t *testing.T) {
 		SweepConfTarget:         2,
 		HtlcConfirmations:       2,
 		SwapPublicationDeadline: initiationTime,
+		PaymentTimeout:          time.Second * 11,
 	}
 
 	t.Run("no outgoing set", func(t *testing.T) {
@@ -120,6 +121,8 @@ func testSqliteLoopOutStore(t *testing.T, pendingSwap *LoopOutContract) {
 		if expectedState == StatePreimageRevealed {
 			require.NotNil(t, swap.State().HtlcTxHash)
 		}
+
+		require.Equal(t, time.Second*11, swap.Contract.PaymentTimeout)
 	}
 
 	// If we create a new swap, then it should show up as being initialized

loopdb/sqlc/batch.sql.go

@@ -0,0 +1,3 @@
+-- payment_timeout is the timeout in seconds for each individual off-chain
+-- payment.
+ALTER TABLE loopout_swaps DROP COLUMN payment_timeout;

loopdb/sqlc/migrations/000007_payment_timeout.down.sql

@@ -0,0 +1,3 @@
+-- payment_timeout is the timeout in seconds for each individual off-chain
+-- payment.
+ALTER TABLE loopout_swaps ADD payment_timeout INTEGER NOT NULL DEFAULT 0;

loopdb/sqlc/migrations/000007_payment_timeout.up.sql

@@ -105,9 +105,10 @@ INSERT INTO loopout_swaps (
     prepay_invoice,
     max_prepay_routing_fee,
     publication_deadline,
-    single_sweep
+    single_sweep,
+    payment_timeout
 ) VALUES (
-    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11
+    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12
 );
 
 -- name: InsertLoopIn :exec
@@ -131,4 +132,4 @@ INSERT INTO htlc_keys(
     client_key_index
 ) VALUES (
     $1, $2, $3, $4, $5, $6, $7
-);
+);

loopdb/sqlc/models.go

@@ -195,6 +195,7 @@ func newLoopOutSwap(globalCtx context.Context, cfg *swapConfig,
 			ProtocolVersion: loopdb.CurrentProtocolVersion(),
 		},
 		OutgoingChanSet: chanSet,
+		PaymentTimeout:  request.PaymentTimeout,
 	}
 
 	swapKit := newSwapKit(
@@ -610,7 +611,8 @@ func (s *loopOutSwap) payInvoices(ctx context.Context) {
 	// Use the recommended routing plugin.
 	s.swapPaymentChan = s.payInvoice(
 		ctx, s.SwapInvoice, s.MaxSwapRoutingFee,
-		s.LoopOutContract.OutgoingChanSet, pluginType, true,
+		s.LoopOutContract.OutgoingChanSet,
+		s.LoopOutContract.PaymentTimeout, pluginType, true,
 	)
 
 	// Pay the prepay invoice. Won't use the routing plugin here as the
@@ -619,7 +621,8 @@ func (s *loopOutSwap) payInvoices(ctx context.Context) {
 	s.log.Infof("Sending prepayment %v", s.PrepayInvoice)
 	s.prePaymentChan = s.payInvoice(
 		ctx, s.PrepayInvoice, s.MaxPrepayRoutingFee,
-		s.LoopOutContract.OutgoingChanSet, RoutingPluginNone, false,
+		s.LoopOutContract.OutgoingChanSet,
+		s.LoopOutContract.PaymentTimeout, RoutingPluginNone, false,
 	)
 }
 
@@ -647,7 +650,7 @@ func (p paymentResult) failure() error {
 // payInvoice pays a single invoice.
 func (s *loopOutSwap) payInvoice(ctx context.Context, invoice string,
 	maxFee btcutil.Amount, outgoingChanIds loopdb.ChannelSet,
-	pluginType RoutingPluginType,
+	paymentTimeout time.Duration, pluginType RoutingPluginType,
 	reportPluginResult bool) chan paymentResult {
 
 	resultChan := make(chan paymentResult)
@@ -662,8 +665,8 @@ func (s *loopOutSwap) payInvoice(ctx context.Context, invoice string,
 		var result paymentResult
 
 		status, err := s.payInvoiceAsync(
-			ctx, invoice, maxFee, outgoingChanIds, pluginType,
-			reportPluginResult,
+			ctx, invoice, maxFee, outgoingChanIds, paymentTimeout,
+			pluginType, reportPluginResult,
 		)
 		if err != nil {
 			result.err = err
@@ -691,8 +694,9 @@ func (s *loopOutSwap) payInvoice(ctx context.Context, invoice string,
 // payInvoiceAsync is the asynchronously executed part of paying an invoice.
 func (s *loopOutSwap) payInvoiceAsync(ctx context.Context,
 	invoice string, maxFee btcutil.Amount,
-	outgoingChanIds loopdb.ChannelSet, pluginType RoutingPluginType,
-	reportPluginResult bool) (*lndclient.PaymentStatus, error) {
+	outgoingChanIds loopdb.ChannelSet, paymentTimeout time.Duration,
+	pluginType RoutingPluginType, reportPluginResult bool) (
+	*lndclient.PaymentStatus, error) {
 
 	// Extract hash from payment request. Unfortunately the request
 	// components aren't available directly.
@@ -705,7 +709,7 @@ func (s *loopOutSwap) payInvoiceAsync(ctx context.Context,
 	}
 
 	maxRetries := 1
-	paymentTimeout := s.executeConfig.totalPaymentTimeout
+	totalPaymentTimeout := s.executeConfig.totalPaymentTimeout
 
 	// Attempt to acquire and initialize the routing plugin.
 	routingPlugin, err := AcquireRoutingPlugin(
@@ -720,8 +724,30 @@ func (s *loopOutSwap) payInvoiceAsync(ctx context.Context,
 			pluginType, hash.String())
 
 		maxRetries = s.executeConfig.maxPaymentRetries
-		paymentTimeout /= time.Duration(maxRetries)
+
+		// If not set, default to the per payment timeout to the total
+		// payment timeout divied by the configured maximum retries.
+		if paymentTimeout == 0 {
+			paymentTimeout = totalPaymentTimeout /
+				time.Duration(maxRetries)
+		}
+
+		// If the payment timeout is too long, we need to adjust the
+		// number of retries to ensure we don't exceed the total
+		// payment timeout.
+		if paymentTimeout*time.Duration(maxRetries) >
+			totalPaymentTimeout {
+
+			maxRetries = int(totalPaymentTimeout / paymentTimeout)
+			s.log.Infof("Adjusted max routing plugin retries to "+
+				"%v to stay within total payment timeout",
+				maxRetries)
+		}
 		defer ReleaseRoutingPlugin(ctx)
+	} else if paymentTimeout == 0 {
+		// If not set, default the payment timeout to the total payment
+		// timeout.
+		paymentTimeout = totalPaymentTimeout
 	}
 
 	req := lndclient.SendPaymentRequest{