Merge pull request #90 from guggero/connect-fix

poold: fix connection issues and retry mechanism

account/manager.go

@@ -399,11 +399,6 @@ func (m *Manager) resumeAccount(ctx context.Context, account *Account, // nolint
 		return fmt.Errorf("unable to construct account output: %v", err)
 	}
 
-	terms, err := m.cfg.Auctioneer.Terms(ctx)
-	if err != nil {
-		return fmt.Errorf("could not query auctioneer terms: %v", err)
-	}
-
 	var accountTx *wire.MsgTx
 	switch account.State {
 	// In StateInitiated, we'll attempt to fund our account.
@@ -534,6 +529,11 @@ func (m *Manager) resumeAccount(ctx context.Context, account *Account, // nolint
 		if err != nil {
 			return err
 		}
+		terms, err := m.cfg.Auctioneer.Terms(ctx)
+		if err != nil {
+			return fmt.Errorf("could not query auctioneer terms: "+
+				"%v", err)
+		}
 
 		// Proceed to watch for the account on-chain.
 		numConfs := NumConfsForValue(
@@ -564,6 +564,15 @@ func (m *Manager) resumeAccount(ctx context.Context, account *Account, // nolint
 	// and would therefore not be noticed by us. The account would stay
 	// pending forever in that case.
 	case StatePendingUpdate, StatePendingBatch:
+		// We need to know the maximum account value to scale the number
+		// of confirmations the same way the auctioneer does to avoid
+		// getting the state out of sync.
+		terms, err := m.cfg.Auctioneer.Terms(ctx)
+		if err != nil {
+			return fmt.Errorf("could not query auctioneer terms: "+
+				"%v", err)
+		}
+
 		numConfs := NumConfsForValue(
 			account.Value, terms.MaxAccountValue,
 		)

auctioneer/account_subscription.go

@@ -21,7 +21,8 @@ type acctSubscription struct {
 	sendMsg    func(*poolrpc.ClientAuctionMessage) error
 	signer     lndclient.SignerClient
 	msgChan    chan *poolrpc.ServerAuctionMessage
-	quit       <-chan struct{}
+	errChan    chan error
+	quit       chan struct{}
 }
 
 // authenticate performs the 3-way authentication handshake between the trader
@@ -58,7 +59,7 @@ func (s *acctSubscription) authenticate(ctx context.Context) error {
 	}
 
 	// We can't sign anything if we haven't received the server's challenge
-	// yet. So we'll wait for the message to arrive.
+	// yet. So we'll wait for the message or an error to arrive.
 	select {
 	case srvMsg, more := <-s.msgChan:
 		if !more {
@@ -94,11 +95,15 @@ func (s *acctSubscription) authenticate(ctx context.Context) error {
 			},
 		})
 
+	case err := <-s.errChan:
+		return fmt.Errorf("error during authentication, before "+
+			"sending subscribe: %v", err)
+
 	case <-ctx.Done():
 		return fmt.Errorf("context canceled before challenge was " +
 			"received")
 
 	case <-s.quit:
-		return ErrClientShutdown
+		return ErrAuthCanceled
 	}
 }

auctioneer/account_subscription_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/hex"
+	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -193,3 +194,56 @@ func TestAccountSubscriptionAuthenticateContextClose(t *testing.T) {
 		t.Fatalf("did not receive commit message before timeout")
 	}
 }
+
+// TestAccountSubscriptionAuthenticateError tests that the 3-way
+// authentication handshake returns an error correctly if the server sends one
+// in the last step.
+func TestAccountSubscriptionAuthenticateError(t *testing.T) {
+	var (
+		msgChan    = make(chan *poolrpc.ClientAuctionMessage)
+		srvMsgChan = make(chan *poolrpc.ServerAuctionMessage)
+		errChan    = make(chan error)
+		sendMsg    = func(msg *poolrpc.ClientAuctionMessage) error {
+			msgChan <- msg
+			return nil
+		}
+		sub = &acctSubscription{
+			acctKey: testAccountDesc,
+			sendMsg: sendMsg,
+			signer:  testSigner,
+			msgChan: srvMsgChan,
+			errChan: make(chan error),
+		}
+	)
+
+	// First, kick off the auth handshake in a goroutine. Every step will
+	// block because we don't use buffered channels.
+	go func() {
+		errChan <- sub.authenticate(context.Background())
+	}()
+
+	// Step 1: We expect a commitment message.
+	select {
+	case msg := <-msgChan:
+		if _, ok := msg.Msg.(*poolrpc.ClientAuctionMessage_Commit); !ok {
+			t.Fatalf("unexpected message type: %v", msg)
+		}
+
+	case <-time.After(defaultTimeout):
+		t.Fatalf("did not receive commit message before timeout")
+	}
+
+	// Step 2: Simulate the server sending an error message next.
+	sub.errChan <- fmt.Errorf("invalid signature")
+
+	// There should be an error in the chan now.
+	select {
+	case err := <-errChan:
+		if !strings.Contains(err.Error(), "invalid signature") {
+			t.Fatalf("unexpected error: %v", err)
+		}
+
+	case <-time.After(defaultTimeout):
+		t.Fatalf("did not receive commit message before timeout")
+	}
+}

auctioneer/client.go

@@ -47,6 +47,10 @@ var (
 	// ErrClientShutdown is the error that is returned if the trader client
 	// itself is shutting down.
 	ErrClientShutdown = errors.New("client shutting down")
+
+	// ErrAuthCanceled is returned if the authentication process of a single
+	// account subscription is aborted.
+	ErrAuthCanceled = errors.New("authentication was canceled")
 )
 
 // Config holds the configuration options for the auctioneer client.
@@ -93,6 +97,7 @@ type Client struct {
 	stopped uint32
 
 	StreamErrChan  chan error
+	errChanSwitch  *ErrChanSwitch
 	FromServerChan chan *poolrpc.ServerAuctionMessage
 
 	serverConn *grpc.ClientConn
@@ -116,10 +121,13 @@ func NewClient(cfg *Config) (*Client, error) {
 		return nil, err
 	}
 
+	mainErrChan := make(chan error)
+	errChanSwitch := NewErrChanSwitch(mainErrChan)
 	return &Client{
 		cfg:             cfg,
 		FromServerChan:  make(chan *poolrpc.ServerAuctionMessage),
-		StreamErrChan:   make(chan error),
+		StreamErrChan:   mainErrChan,
+		errChanSwitch:   errChanSwitch,
 		quit:            make(chan struct{}),
 		subscribedAccts: make(map[[33]byte]*acctSubscription),
 	}, nil
@@ -140,6 +148,8 @@ func (c *Client) Start() error {
 	c.serverConn = serverConn
 	c.client = poolrpc.NewChannelAuctioneerClient(serverConn)
 
+	c.errChanSwitch.Start()
+
 	return nil
 }
 
@@ -189,6 +199,7 @@ func (c *Client) Stop() error {
 	}
 	c.wg.Wait()
 	close(c.FromServerChan)
+	c.errChanSwitch.Stop()
 	return c.serverConn.Close()
 }
 
@@ -207,6 +218,7 @@ func (c *Client) closeStream() error {
 
 	// Close all pending subscriptions.
 	for _, subscription := range c.subscribedAccts {
+		close(subscription.quit)
 		close(subscription.msgChan)
 	}
 
@@ -467,6 +479,14 @@ func (c *Client) connectAndAuthenticate(ctx context.Context,
 		}
 	}
 
+	// For the duration this subscription is active, we need to redirect any
+	// errors sent from the auctioneer to a different channel so the
+	// subscription can react to it. Once we're done, we restore the
+	// original channel.
+	tempErrChan := make(chan error)
+	c.errChanSwitch.Divert(tempErrChan)
+	defer c.errChanSwitch.Restore()
+
 	// Before we can expect to receive any updates, we need to perform the
 	// 3-way authentication handshake.
 	sub = &acctSubscription{
@@ -474,7 +494,8 @@ func (c *Client) connectAndAuthenticate(ctx context.Context,
 		sendMsg: c.SendAuctionMessage,
 		signer:  c.cfg.Signer,
 		msgChan: make(chan *poolrpc.ServerAuctionMessage),
-		quit:    c.quit,
+		errChan: tempErrChan,
+		quit:    make(chan struct{}),
 	}
 	c.subscribedAccts[acctPubKey] = sub
 	err := sub.authenticate(ctx)
@@ -537,6 +558,13 @@ func (c *Client) connectAndAuthenticate(ctx context.Context,
 				"received: %v", srvMsg)
 		}
 
+	case err := <-tempErrChan:
+		return nil, false, fmt.Errorf("error during authentication "+
+			"when waiting for final step: %v", err)
+
+	case <-sub.quit:
+		return nil, false, ErrAuthCanceled
+
 	case <-c.quit:
 		return nil, false, ErrClientShutdown
 	}
@@ -718,14 +746,13 @@ func (c *Client) SendAuctionMessage(msg *poolrpc.ClientAuctionMessage) error {
 // wait blocks for a given amount of time but returns immediately if the client
 // is shutting down.
 func (c *Client) wait(backoff time.Duration) error {
-	if backoff > 0 {
-		select {
-		case <-time.After(backoff):
-		case <-c.quit:
-			return ErrClientShutdown
-		}
+	select {
+	case <-time.After(backoff):
+		return nil
+
+	case <-c.quit:
+		return ErrClientShutdown
 	}
-	return nil
 }
 
 // connectServerStream opens the initial connection to the server for the stream
@@ -735,17 +762,22 @@ func (c *Client) connectServerStream(initialBackoff time.Duration,
 
 	var (
 		backoff = initialBackoff
+		ctxb    = context.Background()
 		ctx     context.Context
 		err     error
 	)
 	for i := 0; i < numRetries; i++ {
 		// Wait before connecting in case this is a reconnect trial.
-		err = c.wait(backoff)
-		if err != nil {
-			return err
+		if backoff != 0 {
+			err = c.wait(backoff)
+			if err != nil {
+				return err
+			}
 		}
-		ctx, c.streamCancel = context.WithCancel(context.Background())
-		c.serverStream, err = c.client.SubscribeBatchAuction(ctx)
+
+		// Try connecting by querying a "cheap" RPC that the server can
+		// answer from memory only.
+		_, err = c.client.Terms(ctxb, &poolrpc.TermsRequest{})
 		if err == nil {
 			log.Debugf("Connected successfully to server after "+
 				"%d tries", i+1)
@@ -763,9 +795,11 @@ func (c *Client) connectServerStream(initialBackoff time.Duration,
 		}
 		log.Debugf("Connect failed with error, canceling and backing "+
 			"off for %s: %v", backoff, err)
-		c.streamCancel()
-		log.Infof("Connection to server failed, will try again in %v",
-			backoff)
+
+		if i < numRetries-1 {
+			log.Infof("Connection to server failed, will try again "+
+				"in %v", backoff)
+		}
 	}
 	if err != nil {
 		log.Errorf("Connection to server failed after %d retries",
@@ -773,6 +807,15 @@ func (c *Client) connectServerStream(initialBackoff time.Duration,
 		return err
 	}
 
+	// Now that we know the connection itself is established, we also re-
+	// connect the long-lived stream.
+	ctx, c.streamCancel = context.WithCancel(ctxb)
+	c.serverStream, err = c.client.SubscribeBatchAuction(ctx)
+	if err != nil {
+		log.Errorf("Subscribing to batch auction failed: %v", err)
+		return err
+	}
+
 	// Read incoming messages and send them to the channel where
 	// the order manager is listening on. We can only send our first message
 	// to the server after we've received the challenge, which we'll track
@@ -813,7 +856,7 @@ func (c *Client) readIncomingStream() { // nolint:gocyclo
 		// error, usually "transport is closing".
 		case err == io.EOF:
 			select {
-			case c.StreamErrChan <- ErrServerShutdown:
+			case c.errChanSwitch.ErrChan() <- ErrServerShutdown:
 			case <-c.quit:
 			}
 			return
@@ -831,7 +874,7 @@ func (c *Client) readIncomingStream() { // nolint:gocyclo
 
 			// Any other error we want to report back.
 			select {
-			case c.StreamErrChan <- err:
+			case c.errChanSwitch.ErrChan() <- err:
 			case <-c.quit:
 			}
 			return
@@ -856,8 +899,8 @@ func (c *Client) readIncomingStream() { // nolint:gocyclo
 				}
 			}
 			if acctSub == nil {
-				c.StreamErrChan <- fmt.Errorf("no sub"+
-					"scription found for commit hash %x",
+				c.errChanSwitch.ErrChan() <- fmt.Errorf("no "+
+					"subscription found for commit hash %x",
 					commitHash)
 				return
 			}
@@ -874,7 +917,7 @@ func (c *Client) readIncomingStream() { // nolint:gocyclo
 		case *poolrpc.ServerAuctionMessage_Success:
 			err := c.sendToSubscription(t.Success.TraderKey, msg)
 			if err != nil {
-				c.StreamErrChan <- err
+				c.errChanSwitch.ErrChan() <- err
 				return
 			}
 
@@ -884,7 +927,7 @@ func (c *Client) readIncomingStream() { // nolint:gocyclo
 		case *poolrpc.ServerAuctionMessage_Account:
 			err := c.sendToSubscription(t.Account.TraderKey, msg)
 			if err != nil {
-				c.StreamErrChan <- err
+				c.errChanSwitch.ErrChan() <- err
 				return
 			}
 
@@ -901,7 +944,7 @@ func (c *Client) readIncomingStream() { // nolint:gocyclo
 				err := c.HandleServerShutdown(nil)
 				if err != nil {
 					select {
-					case c.StreamErrChan <- err:
+					case c.errChanSwitch.ErrChan() <- err:
 					case <-c.quit:
 					}
 				}
@@ -917,7 +960,7 @@ func (c *Client) readIncomingStream() { // nolint:gocyclo
 					t.Error.TraderKey, msg,
 				)
 				if err != nil {
-					c.StreamErrChan <- err
+					c.errChanSwitch.ErrChan() <- err
 					return
 				}
 
@@ -989,11 +1032,10 @@ func (c *Client) HandleServerShutdown(err error) error {
 	// which requires access to the lock as well.
 	c.streamMutex.Lock()
 	err = c.connectServerStream(c.cfg.MinBackoff, reconnectRetries)
+	c.streamMutex.Unlock()
 	if err != nil {
-		c.streamMutex.Unlock()
 		return err
 	}
-	c.streamMutex.Unlock()
 
 	// With the connection re-established, check whether we need to mark our
 	// pending batch as finalized, or if we need to remove it due to the

auctioneer/err_chan_switch.go

@@ -0,0 +1,98 @@
+package auctioneer
+
+import "sync"
+
+// ErrChanSwitch is a type that can switch incoming error messages between a
+// main channel and a temporary channel in a concurrency safe way.
+type ErrChanSwitch struct {
+	mainChan chan<- error
+	tempChan chan<- error
+	diverted bool
+
+	incomingChan chan error
+
+	sync.Mutex
+	wg sync.WaitGroup
+
+	quit chan struct{}
+}
+
+// NewErrChanSwitch creates a new error channel switcher with the given main
+// channel that error messages are forwarded to by default.
+func NewErrChanSwitch(mainChan chan<- error) *ErrChanSwitch {
+	return &ErrChanSwitch{
+		mainChan:     mainChan,
+		diverted:     false,
+		incomingChan: make(chan error),
+		quit:         make(chan struct{}),
+	}
+}
+
+// ErrChan returns the incoming channel that errors can be sent to that are then
+// switched in a concurrency safe way.
+func (s *ErrChanSwitch) ErrChan() chan<- error {
+	return s.incomingChan
+}
+
+// Divert causes all incoming error messages to be sent to the given temporary
+// channel from now on instead of the main error channel.
+func (s *ErrChanSwitch) Divert(tempChan chan<- error) {
+	s.Lock()
+	defer s.Unlock()
+
+	s.tempChan = tempChan
+	s.diverted = true
+}
+
+// Restore causes all incoming error messages to be sent to the main channel
+// again from now on.
+func (s *ErrChanSwitch) Restore() {
+	s.Lock()
+	defer s.Unlock()
+
+	s.tempChan = nil
+	s.diverted = false
+}
+
+// Start spins up the internal goroutine that processes incoming messages.
+func (s *ErrChanSwitch) Start() {
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		s.run()
+	}()
+}
+
+// Stop shuts down the goroutine that processes incoming messages.
+func (s *ErrChanSwitch) Stop() {
+	close(s.quit)
+	s.wg.Wait()
+}
+
+// run is the main event loop where we receive errors on the incoming channel
+// and send them to the correct outgoing channel in a concurrency safe way.
+func (s *ErrChanSwitch) run() {
+	for {
+		select {
+		case msg := <-s.incomingChan:
+			// Make sure that while we are processing a message the
+			// channels can't be switched.
+			s.Lock()
+			if s.diverted {
+				select {
+				case s.tempChan <- msg:
+				case <-s.quit:
+				}
+			} else {
+				select {
+				case s.mainChan <- msg:
+				case <-s.quit:
+				}
+			}
+			s.Unlock()
+
+		case <-s.quit:
+			return
+		}
+	}
+}