multi: consume and log sever state updates

swap_server_client.go

@@ -6,7 +6,10 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"io"
 	"net"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/btcsuite/btcd/btcec"
@@ -24,6 +27,21 @@ import (
 // supported by the loop client.
 const protocolVersion = looprpc.ProtocolVersion_PREIMAGE_PUSH_LOOP_OUT
 
+var (
+	// errServerSubscriptionComplete is returned when our subscription to
+	// server updates exits because the server has no more updates to
+	// provide us, because its part in the swap is complete.
+	errServerSubscriptionComplete = errors.New("server finished serving " +
+		"updates")
+
+	// errSubscriptionFailed is returned when our subscription returns with
+	// and EOF, indicating that the server restarted, we had an unexpected
+	// network failure. Since we do not have restart-recovery, we note that
+	// we will not resume our subscription once this error occurs.
+	errSubscriptionFailed = errors.New("failed, no further updates will " +
+		"be provided")
+)
+
 type swapServerClient interface {
 	GetLoopOutTerms(ctx context.Context) (
 		*LoopOutTerms, error)
@@ -51,11 +69,31 @@ type swapServerClient interface {
 		swapHash lntypes.Hash, amount btcutil.Amount,
 		senderKey [33]byte, swapInvoice string, lastHop *route.Vertex) (
 		*newLoopInResponse, error)
+
+	// SubscribeLoopOutUpdates subscribes to loop out server state.
+	SubscribeLoopOutUpdates(ctx context.Context,
+		hash lntypes.Hash) (<-chan *ServerUpdate, <-chan error, error)
+
+	// SubscribeLoopInUpdates subscribes to loop in server state.
+	SubscribeLoopInUpdates(ctx context.Context,
+		hash lntypes.Hash) (<-chan *ServerUpdate, <-chan error, error)
 }
 
 type grpcSwapServerClient struct {
 	server looprpc.SwapServerClient
 	conn   *grpc.ClientConn
+
+	wg sync.WaitGroup
+}
+
+// stop sends the signal for the server's goroutines to shutdown and waits for
+// them to complete.
+func (s *grpcSwapServerClient) stop() {
+	if err := s.conn.Close(); err != nil {
+		log.Warnf("could not close connection: %v", err)
+	}
+
+	s.wg.Wait()
 }
 
 var _ swapServerClient = (*grpcSwapServerClient)(nil)
@@ -275,8 +313,144 @@ func (s *grpcSwapServerClient) NewLoopInSwap(ctx context.Context,
 	}, nil
 }
 
-func (s *grpcSwapServerClient) Close() {
-	s.conn.Close()
+// ServerUpdate summarizes an update from the swap server.
+type ServerUpdate struct {
+	// State is the state that the server has sent us.
+	State looprpc.ServerSwapState
+
+	// Timestamp is the time of the server state update.
+	Timestamp time.Time
+}
+
+// SubscribeLoopInUpdates subscribes to loop in server state and pipes updates
+// into the channel provided.
+func (s *grpcSwapServerClient) SubscribeLoopInUpdates(ctx context.Context,
+	hash lntypes.Hash) (<-chan *ServerUpdate, <-chan error, error) {
+
+	resp, err := s.server.SubscribeLoopInUpdates(
+		ctx, &looprpc.SubscribeUpdatesRequest{
+			ProtocolVersion: protocolVersion,
+			SwapHash:        hash[:],
+		},
+	)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	receive := func() (*ServerUpdate, error) {
+		response, err := resp.Recv()
+		if err != nil {
+			return nil, err
+		}
+
+		return &ServerUpdate{
+			State:     response.State,
+			Timestamp: time.Unix(0, response.TimestampNs),
+		}, nil
+	}
+
+	updateChan, errChan := s.makeServerUpdate(ctx, receive)
+	return updateChan, errChan, nil
+}
+
+// SubscribeLoopOutUpdates subscribes to loop out server state and pipes updates
+// into the channel provided.
+func (s *grpcSwapServerClient) SubscribeLoopOutUpdates(ctx context.Context,
+	hash lntypes.Hash) (<-chan *ServerUpdate, <-chan error, error) {
+
+	resp, err := s.server.SubscribeLoopOutUpdates(
+		ctx, &looprpc.SubscribeUpdatesRequest{
+			ProtocolVersion: protocolVersion,
+			SwapHash:        hash[:],
+		},
+	)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	receive := func() (*ServerUpdate, error) {
+		response, err := resp.Recv()
+		if err != nil {
+			return nil, err
+		}
+
+		return &ServerUpdate{
+			State:     response.State,
+			Timestamp: time.Unix(0, response.TimestampNs),
+		}, nil
+	}
+
+	updateChan, errChan := s.makeServerUpdate(ctx, receive)
+	return updateChan, errChan, nil
+}
+
+// makeServerUpdate takes a stream receive function and a channel that it
+// should pipe updates into. It sends events into the updates channel until
+// the client cancels, server client shuts down or the subscription is cancelled
+// server side.
+func (s *grpcSwapServerClient) makeServerUpdate(ctx context.Context,
+	receive func() (*ServerUpdate, error)) (<-chan *ServerUpdate,
+	<-chan error) {
+
+	// We will return exactly one error from this function so we buffer
+	// our error channel so that the function exit is not dependent on
+	// the error being read.
+	errChan := make(chan error, 1)
+	updateChan := make(chan *ServerUpdate)
+
+	// Create a goroutine that will pipe updates in to our updates channel.
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+
+		for {
+			// Try to receive from our stream. If there are no items
+			// to consume, this call will block. If our stream is
+			// cancelled by the server we will receive an error.
+			response, err := receive()
+			switch err {
+			// If we get a nil error, we proceed with to delivering
+			// the update we have just received.
+			case nil:
+
+			// If we get an EOF error, the server is finished
+			// sending us updates, so we return with a non-nil
+			// a subscription complete error to inform the caller
+			// that they will no longer receive updates.
+			case io.EOF:
+				errChan <- errServerSubscriptionComplete
+				return
+
+			// If we receive a non-nil error, we exit.
+			default:
+				// If we get a transport is closing error, we
+				// send a server restarting error so that the
+				// caller is informed that we will not get
+				// any more updates from the server (since we
+				// don't have retry logic yet).
+				if isErrConClosing(err) {
+					errChan <- errSubscriptionFailed
+				} else {
+					errChan <- err
+				}
+
+				return
+			}
+
+			select {
+			// Try to send our update to the update channel.
+			case updateChan <- response:
+
+			// If the client cancels their context, we exit with
+			// no error.
+			case <-ctx.Done():
+				errChan <- nil
+				return
+			}
+		}
+	}()
+
+	return updateChan, errChan
 }
 
 // getSwapServerConn returns a connection to the swap server. A non-empty
@@ -286,9 +460,14 @@ func getSwapServerConn(address, proxyAddress string, insecure bool,
 	tlsPath string, interceptor *lsat.Interceptor) (*grpc.ClientConn, error) {
 
 	// Create a dial options array.
-	opts := []grpc.DialOption{grpc.WithUnaryInterceptor(
-		interceptor.UnaryInterceptor,
-	)}
+	opts := []grpc.DialOption{
+		grpc.WithUnaryInterceptor(
+			interceptor.UnaryInterceptor,
+		),
+		grpc.WithStreamInterceptor(
+			interceptor.StreamInterceptor,
+		),
+	}
 
 	// There are three options to connect to a swap server, either insecure,
 	// using a self-signed certificate or with a certificate signed by a
@@ -331,6 +510,18 @@ func getSwapServerConn(address, proxyAddress string, insecure bool,
 	return conn, nil
 }
 
+// isErrConClosing identifies whether we have received a "transport is closing"
+// error from a grpc stream, indicating that the server has shutdown. We need
+// to string match this error because ErrConnClosing is part of an internal
+// grpc package, so cannot be used directly.
+func isErrConClosing(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	return strings.Contains(err.Error(), "transport is closing")
+}
+
 type newLoopOutResponse struct {
 	swapInvoice   string
 	prepayInvoice string