Merge pull request #396 from carlaKC/subscribesingle-eof

loopin: handle SubscribeSingleInvoice termination

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway v1.14.3
 	github.com/jessevdk/go-flags v1.4.0
 	github.com/lightninglabs/aperture v0.1.6-beta
-	github.com/lightninglabs/lndclient v0.11.1-6
+	github.com/lightninglabs/lndclient v0.11.1-9
 	github.com/lightninglabs/protobuf-hex-display v1.4.3-hex-display
 	github.com/lightningnetwork/lnd v0.13.0-beta.rc2
 	github.com/lightningnetwork/lnd/cert v1.0.3

go.sum

@@ -230,8 +230,8 @@ github.com/lightninglabs/aperture v0.1.6-beta/go.mod h1:9xl4mx778ZAzrB87nLHMqk+X
 github.com/lightninglabs/gozmq v0.0.0-20191113021534-d20a764486bf h1:HZKvJUHlcXI/f/O0Avg7t8sqkPo78HFzjmeYFl6DPnc=
 github.com/lightninglabs/gozmq v0.0.0-20191113021534-d20a764486bf/go.mod h1:vxmQPeIQxPf6Jf9rM8R+B4rKBqLA2AjttNxkFBL2Plk=
 github.com/lightninglabs/lndclient v0.11.0-4/go.mod h1:8/cTKNwgL87NX123gmlv3Xh6p1a7pvzu+40Un3PhHiI=
-github.com/lightninglabs/lndclient v0.11.1-6 h1:2L+0GIjAShSWsxEsRQ/ZbK+62SF4DBiCpnWBx8HSwyA=
-github.com/lightninglabs/lndclient v0.11.1-6/go.mod h1:qVFcrIXxsagpu3RC0SSSdVEs3QVxuv5YiHUYwDauUnc=
+github.com/lightninglabs/lndclient v0.11.1-9 h1:KTrCkOnqqP1gCsou0D7k7TOOC7HLboKDS35YREN7a3s=
+github.com/lightninglabs/lndclient v0.11.1-9/go.mod h1:qVFcrIXxsagpu3RC0SSSdVEs3QVxuv5YiHUYwDauUnc=
 github.com/lightninglabs/neutrino v0.11.0/go.mod h1:CuhF0iuzg9Sp2HO6ZgXgayviFTn1QHdSTJlMncK80wg=
 github.com/lightninglabs/neutrino v0.11.1-0.20200316235139-bffc52e8f200/go.mod h1:MlZmoKa7CJP3eR1s5yB7Rm5aSyadpKkxqAwLQmog7N0=
 github.com/lightninglabs/neutrino v0.12.1 h1:9umzk5kKNc/l3bAyak8ClSRP1qSulnjc6kppLYDnuqk=

loopin.go

@@ -779,12 +779,30 @@ func (s *loopInSwap) waitForSwapComplete(ctx context.Context,
 			htlcSpend = true
 
 		// Swap invoice ntfn error.
-		case err := <-swapInvoiceErr:
+		case err, ok := <-swapInvoiceErr:
+			// If the channel has been closed, the server has
+			// finished sending updates, so we set the channel to
+			// nil because we don't want to constantly select this
+			// case.
+			if !ok {
+				swapInvoiceErr = nil
+				continue
+			}
+
 			return err
 
 		// An update to the swap invoice occurred. Check the new state
 		// and update the swap state accordingly.
-		case update := <-swapInvoiceChan:
+		case update, ok := <-swapInvoiceChan:
+			// If the channel has been closed, the server has
+			// finished sending updates, so we set the channel to
+			// nil because we don't want to constantly select this
+			// case.
+			if !ok {
+				swapInvoiceChan = nil
+				continue
+			}
+
 			s.log.Infof("Received swap invoice update: %v",
 				update.State)
 

loopin_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"testing"
 
-	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/swap"
 	"github.com/lightninglabs/loop/test"
@@ -89,17 +88,11 @@ func TestLoopInSuccess(t *testing.T) {
 	<-ctx.lnd.RegisterSpendChannel
 
 	// Client starts listening for swap invoice updates.
-	subscription := <-ctx.lnd.SingleInvoiceSubcribeChannel
-	if subscription.Hash != ctx.server.swapHash {
-		t.Fatal("client subscribing to wrong invoice")
-	}
+	ctx.assertSubscribeInvoice(ctx.server.swapHash)
 
 	// Server has already paid invoice before spending the htlc. Signal
 	// settled.
-	subscription.Update <- lndclient.InvoiceUpdate{
-		State:   channeldb.ContractSettled,
-		AmtPaid: 49000,
-	}
+	ctx.updateInvoiceState(49000, channeldb.ContractSettled)
 
 	// Swap is expected to move to the state InvoiceSettled
 	ctx.assertState(loopdb.StateInvoiceSettled)
@@ -257,10 +250,7 @@ func testLoopInTimeout(t *testing.T,
 	<-ctx.lnd.RegisterSpendChannel
 
 	// Client starts listening for swap invoice updates.
-	subscription := <-ctx.lnd.SingleInvoiceSubcribeChannel
-	if subscription.Hash != ctx.server.swapHash {
-		t.Fatal("client subscribing to wrong invoice")
-	}
+	ctx.assertSubscribeInvoice(ctx.server.swapHash)
 
 	// Let htlc expire.
 	ctx.blockEpochChan <- s.LoopInContract.CltvExpiry
@@ -294,10 +284,8 @@ func testLoopInTimeout(t *testing.T,
 	// safely cancel the swap invoice.
 	<-ctx.lnd.FailInvoiceChannel
 
-	// Signal the the invoice was canceled.
-	subscription.Update <- lndclient.InvoiceUpdate{
-		State: channeldb.ContractCanceled,
-	}
+	// Signal that the invoice was canceled.
+	ctx.updateInvoiceState(0, channeldb.ContractCanceled)
 
 	ctx.assertState(loopdb.StateFailTimeout)
 	state := ctx.store.assertLoopInState(loopdb.StateFailTimeout)
@@ -501,18 +489,12 @@ func testLoopInResume(t *testing.T, state loopdb.SwapState, expired bool,
 	<-ctx.lnd.RegisterSpendChannel
 
 	// Client starts listening for swap invoice updates.
-	subscription := <-ctx.lnd.SingleInvoiceSubcribeChannel
-	if subscription.Hash != testPreimage.Hash() {
-		t.Fatal("client subscribing to wrong invoice")
-	}
+	ctx.assertSubscribeInvoice(testPreimage.Hash())
 
 	// Server has already paid invoice before spending the htlc. Signal
 	// settled.
-	invoiceUpdate := lndclient.InvoiceUpdate{
-		State:   channeldb.ContractSettled,
-		AmtPaid: 49000,
-	}
-	subscription.Update <- invoiceUpdate
+	amtPaid := btcutil.Amount(49000)
+	ctx.updateInvoiceState(amtPaid, channeldb.ContractSettled)
 
 	// Swap is expected to move to the state InvoiceSettled
 	ctx.assertState(loopdb.StateInvoiceSettled)
@@ -537,7 +519,6 @@ func testLoopInResume(t *testing.T, state loopdb.SwapState, expired bool,
 	// We expect our server fee to reflect as the difference between htlc
 	// value and invoice amount paid. We use our original on-chain cost, set
 	// earlier in the test, because we expect this value to be unchanged.
-	cost.Server = btcutil.Amount(htlcTx.TxOut[0].Value) -
-		invoiceUpdate.AmtPaid
+	cost.Server = btcutil.Amount(htlcTx.TxOut[0].Value) - amtPaid
 	require.Equal(t, cost, finalState.Cost)
 }

loopin_testcontext_test.go

@@ -4,9 +4,14 @@ import (
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btcutil"
+	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/sweep"
 	"github.com/lightninglabs/loop/test"
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/stretchr/testify/require"
 )
 
 type loopInTestContext struct {
@@ -18,6 +23,8 @@ type loopInTestContext struct {
 	cfg            *executeConfig
 	statusChan     chan SwapInfo
 	blockEpochChan chan interface{}
+
+	swapInvoiceSubscription *test.SingleInvoiceSubscription
 }
 
 func newLoopInTestContext(t *testing.T) *loopInTestContext {
@@ -61,3 +68,29 @@ func (c *loopInTestContext) assertState(expectedState loopdb.SwapState) {
 			state.State)
 	}
 }
+
+// assertSubscribeInvoice asserts that the client subscribes to invoice updates
+// for our swap invoice.
+func (c *loopInTestContext) assertSubscribeInvoice(hash lntypes.Hash) {
+	c.swapInvoiceSubscription = <-c.lnd.SingleInvoiceSubcribeChannel
+	require.Equal(c.t, hash, c.swapInvoiceSubscription.Hash)
+}
+
+// updateInvoiceState mocks an update to our swap invoice state.
+func (c *loopInTestContext) updateInvoiceState(amount btcutil.Amount,
+	state channeldb.ContractState) {
+
+	c.swapInvoiceSubscription.Update <- lndclient.InvoiceUpdate{
+		AmtPaid: amount,
+		State:   state,
+	}
+
+	// If we're in a final state, close our update channels as lndclient
+	// would.
+	if state == channeldb.ContractCanceled ||
+		state == channeldb.ContractSettled {
+
+		close(c.swapInvoiceSubscription.Update)
+		close(c.swapInvoiceSubscription.Err)
+	}
+}