Merge pull request #210 from joostjager/mpp-probe

loopin: mpp pre-swap probe

Author: Roasbeef

loopdb/protocol_version.go

@@ -35,16 +35,20 @@ const (
 	// HTLC v2 scrips for swaps.
 	ProtocolVersionHtlcV2 ProtocolVersion = 5
 
+	// ProtocolVersionMultiLoopIn indicates that the client creates a probe
+	// invoice so that the server can perform a multi-path probe.
+	ProtocolVersionMultiLoopIn ProtocolVersion = 6
+
 	// ProtocolVersionUnrecorded is set for swaps were created before we
 	// started saving protocol version with swaps.
 	ProtocolVersionUnrecorded ProtocolVersion = math.MaxUint32
 
-	// CurrentRpcProtocolVersion defines the version of the RPC protocol
+	// CurrentRPCProtocolVersion defines the version of the RPC protocol
 	// that is currently supported by the loop client.
-	CurrentRPCProtocolVersion = looprpc.ProtocolVersion_HTLC_V2
+	CurrentRPCProtocolVersion = looprpc.ProtocolVersion_MULTI_LOOP_IN
 
-	// CurrentInteranlProtocolVersionInternal defines the RPC current
-	// protocol in the internal representation.
+	// CurrentInternalProtocolVersion defines the RPC current protocol in
+	// the internal representation.
 	CurrentInternalProtocolVersion = ProtocolVersion(CurrentRPCProtocolVersion)
 )
 

loopdb/protocol_version_test.go

@@ -20,6 +20,7 @@ func TestProtocolVersionSanity(t *testing.T) {
 		ProtocolVersionPreimagePush,
 		ProtocolVersionUserExpiryLoopOut,
 		ProtocolVersionHtlcV2,
+		ProtocolVersionMultiLoopIn,
 	}
 
 	rpcVersions := [...]looprpc.ProtocolVersion{
@@ -29,6 +30,7 @@ func TestProtocolVersionSanity(t *testing.T) {
 		looprpc.ProtocolVersion_PREIMAGE_PUSH_LOOP_OUT,
 		looprpc.ProtocolVersion_USER_EXPIRY_LOOP_OUT,
 		looprpc.ProtocolVersion_HTLC_V2,
+		looprpc.ProtocolVersion_MULTI_LOOP_IN,
 	}
 
 	require.Equal(t, len(versions), len(rpcVersions))

loopin.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/rand"
 	"crypto/sha256"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -135,17 +136,55 @@ func newLoopInSwap(globalCtx context.Context, cfg *swapConfig,
 		return nil, err
 	}
 
+	// Create the probe invoice in lnd. Derive the payment hash
+	// deterministically from the swap hash in such a way that the server
+	// can be sure that we don't know the preimage.
+	probeHash := lntypes.Hash(sha256.Sum256(swapHash[:]))
+	probeHash[0] ^= 1
+
+	log.Infof("Creating probe invoice %v", probeHash)
+	probeInvoice, err := cfg.lnd.Invoices.AddHoldInvoice(
+		globalCtx, &invoicesrpc.AddInvoiceData{
+			Hash:   &probeHash,
+			Value:  lnwire.NewMSatFromSatoshis(swapInvoiceAmt),
+			Memo:   "loop in probe",
+			Expiry: 3600,
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	// Create a cancellable context that is used for monitoring the probe.
+	probeWaitCtx, probeWaitCancel := context.WithCancel(globalCtx)
+
+	// Launch a goroutine to monitor the probe.
+	probeResult, err := awaitProbe(probeWaitCtx, *cfg.lnd, probeHash)
+	if err != nil {
+		probeWaitCancel()
+		return nil, fmt.Errorf("probe failed: %v", err)
+	}
+
 	// Post the swap parameters to the swap server. The response contains
 	// the server success key and the expiry height of the on-chain swap
 	// htlc.
 	log.Infof("Initiating swap request at height %v", currentHeight)
 	swapResp, err := cfg.server.NewLoopInSwap(globalCtx, swapHash,
-		request.Amount, senderKey, swapInvoice, request.LastHop,
+		request.Amount, senderKey, swapInvoice, probeInvoice,
+		request.LastHop,
 	)
+	probeWaitCancel()
 	if err != nil {
 		return nil, fmt.Errorf("cannot initiate swap: %v", err)
 	}
 
+	// Because the context is cancelled, it is guaranteed that we will be
+	// able to read from the probeResult channel.
+	err = <-probeResult
+	if err != nil {
+		return nil, fmt.Errorf("probe error: %v", err)
+	}
+
 	// Validate the response parameters the prevent us continuing with a
 	// swap that is based on parameters outside our allowed range.
 	err = validateLoopInContract(cfg.lnd, currentHeight, request, swapResp)
@@ -209,6 +248,72 @@ func newLoopInSwap(globalCtx context.Context, cfg *swapConfig,
 	}, nil
 }
 
+// awaitProbe waits for a probe payment to arrive and cancels it. This is a
+// workaround for the current lack of multi-path probing.
+func awaitProbe(ctx context.Context, lnd lndclient.LndServices,
+	probeHash lntypes.Hash) (chan error, error) {
+
+	// Subscribe to the probe invoice.
+	updateChan, errChan, err := lnd.Invoices.SubscribeSingleInvoice(
+		ctx, probeHash,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	// Wait in the background for the probe to arrive.
+	probeResult := make(chan error, 1)
+
+	go func() {
+		for {
+			select {
+			case update := <-updateChan:
+				switch update.State {
+				case channeldb.ContractAccepted:
+					log.Infof("Server probe successful")
+					probeResult <- nil
+
+					// Cancel probe invoice so that the
+					// server will know that its probe was
+					// successful.
+					err := lnd.Invoices.CancelInvoice(
+						ctx, probeHash,
+					)
+					if err != nil {
+						log.Errorf("Cancel probe "+
+							"invoice: %v", err)
+					}
+
+					return
+
+				case channeldb.ContractCanceled:
+					probeResult <- errors.New(
+						"probe invoice expired")
+
+					return
+
+				case channeldb.ContractSettled:
+					probeResult <- errors.New(
+						"impossible that probe " +
+							"invoice was settled")
+
+					return
+				}
+
+			case err := <-errChan:
+				probeResult <- err
+				return
+
+			case <-ctx.Done():
+				probeResult <- ctx.Err()
+				return
+			}
+		}
+	}()
+
+	return probeResult, nil
+}
+
 // resumeLoopInSwap returns a swap object representing a pending swap that has
 // been restored from the database.
 func resumeLoopInSwap(reqContext context.Context, cfg *swapConfig,

loopin_testcontext_test.go

@@ -22,7 +22,7 @@ type loopInTestContext struct {
 
 func newLoopInTestContext(t *testing.T) *loopInTestContext {
 	lnd := test.NewMockLnd()
-	server := newServerMock()
+	server := newServerMock(lnd)
 	store := newStoreMock(t)
 	sweeper := sweep.Sweeper{Lnd: &lnd.LndServices}
 

loopout_test.go

@@ -27,7 +27,7 @@ func TestLoopOutPaymentParameters(t *testing.T) {
 	// Set up test context objects.
 	lnd := test.NewMockLnd()
 	ctx := test.NewContext(t, lnd)
-	server := newServerMock()
+	server := newServerMock(lnd)
 	store := newStoreMock(t)
 
 	expiryChan := make(chan time.Time)
@@ -138,7 +138,7 @@ func TestLateHtlcPublish(t *testing.T) {
 
 	ctx := test.NewContext(t, lnd)
 
-	server := newServerMock()
+	server := newServerMock(lnd)
 
 	store := newStoreMock(t)
 
@@ -223,7 +223,7 @@ func TestCustomSweepConfTarget(t *testing.T) {
 
 	lnd := test.NewMockLnd()
 	ctx := test.NewContext(t, lnd)
-	server := newServerMock()
+	server := newServerMock(lnd)
 
 	// Use the highest sweep confirmation target before we attempt to use
 	// the default.
@@ -424,7 +424,7 @@ func TestPreimagePush(t *testing.T) {
 
 	lnd := test.NewMockLnd()
 	ctx := test.NewContext(t, lnd)
-	server := newServerMock()
+	server := newServerMock(lnd)
 
 	// Start with a high confirmation delta which will have a very high fee
 	// attached to it.