reservation: remove notifications from reservation

This commit removes the notification stream from
the reservation manager and replaces it with a
subscriber interface.

Author: sputn1ck

instantout/reservation/actions_test.go

@@ -47,24 +47,24 @@ type mockReservationClient struct {
 	mock.Mock
 }
 
-func (m *mockReservationClient) OpenReservation(ctx context.Context,
-	in *swapserverrpc.ServerOpenReservationRequest,
-	opts ...grpc.CallOption) (*swapserverrpc.ServerOpenReservationResponse,
+func (m *mockReservationClient) ReservationNotificationStream(
+	ctx context.Context, in *swapserverrpc.ReservationNotificationRequest,
+	opts ...grpc.CallOption,
+) (swapserverrpc.ReservationService_ReservationNotificationStreamClient,
 	error) {
 
 	args := m.Called(ctx, in, opts)
-	return args.Get(0).(*swapserverrpc.ServerOpenReservationResponse),
+	return args.Get(0).(swapserverrpc.ReservationService_ReservationNotificationStreamClient),
 		args.Error(1)
 }
 
-func (m *mockReservationClient) ReservationNotificationStream(
-	ctx context.Context, in *swapserverrpc.ReservationNotificationRequest,
-	opts ...grpc.CallOption,
-) (swapserverrpc.ReservationService_ReservationNotificationStreamClient,
+func (m *mockReservationClient) OpenReservation(ctx context.Context,
+	in *swapserverrpc.ServerOpenReservationRequest,
+	opts ...grpc.CallOption) (*swapserverrpc.ServerOpenReservationResponse,
 	error) {
 
 	args := m.Called(ctx, in, opts)
-	return args.Get(0).(swapserverrpc.ReservationService_ReservationNotificationStreamClient),
+	return args.Get(0).(*swapserverrpc.ServerOpenReservationResponse),
 		args.Error(1)
 }
 

instantout/reservation/fsm.go

@@ -28,8 +28,9 @@ type Config struct {
 	// swap server.
 	ReservationClient swapserverrpc.ReservationServiceClient
 
-	// FetchL402 is the function used to fetch the l402 token.
-	FetchL402 func(context.Context) error
+	// NotificationManager is the manager that handles the notification
+	// subscriptions.
+	NotificationManager NotificationManager
 }
 
 // FSM is the state machine that manages the reservation lifecycle.

instantout/reservation/interfaces.go

@@ -3,6 +3,8 @@ package reservation
 import (
 	"context"
 	"fmt"
+
+	"github.com/lightninglabs/loop/swapserverrpc"
 )
 
 var (
@@ -31,3 +33,10 @@ type Store interface {
 	// made.
 	ListReservations(ctx context.Context) ([]*Reservation, error)
 }
+
+// NotificationManager handles subscribing to incoming reservation
+// subscriptions.
+type NotificationManager interface {
+	SubscribeReservations(context.Context,
+	) <-chan *swapserverrpc.ServerReservationNotification
+}

instantout/reservation/manager.go

@@ -22,9 +22,6 @@ type Manager struct {
 	// activeReservations contains all the active reservationsFSMs.
 	activeReservations map[ID]*FSM
 
-	// hasL402 is true if the client has a valid L402.
-	hasL402 bool
-
 	runCtx context.Context
 
 	sync.Mutex
@@ -59,22 +56,15 @@ func (m *Manager) Run(ctx context.Context, height int32) error {
 		return err
 	}
 
-	reservationResChan := make(
-		chan *reservationrpc.ServerReservationNotification,
-	)
-
-	err = m.RegisterReservationNotifications(reservationResChan)
-	if err != nil {
-		return err
-	}
+	ntfnChan := m.cfg.NotificationManager.SubscribeReservations(ctx)
 
 	for {
 		select {
 		case height := <-newBlockChan:
 			log.Debugf("Received block %v", height)
 			currentHeight = height
 
-		case reservationRes := <-reservationResChan:
+		case reservationRes := <-ntfnChan:
 			log.Debugf("Received reservation %x",
 				reservationRes.ReservationId)
 			_, err := m.newReservation(
@@ -157,101 +147,6 @@ func (m *Manager) newReservation(ctx context.Context, currentHeight uint32,
 	return reservationFSM, nil
 }
 
-// fetchL402 fetches the L402 from the server. This method will keep on
-// retrying until it gets a valid response.
-func (m *Manager) fetchL402(ctx context.Context) {
-	// Add a 0 timer so that we initially fetch the L402 immediately.
-	timer := time.NewTimer(0)
-	for {
-		select {
-		case <-ctx.Done():
-			return
-
-		case <-timer.C:
-			err := m.cfg.FetchL402(ctx)
-			if err != nil {
-				log.Warnf("Error fetching L402: %v", err)
-				timer.Reset(time.Second * 10)
-				continue
-			}
-			m.hasL402 = true
-			return
-		}
-	}
-}
-
-// RegisterReservationNotifications registers a new reservation notification
-// stream.
-func (m *Manager) RegisterReservationNotifications(
-	reservationChan chan *reservationrpc.ServerReservationNotification) error {
-
-	// In order to create a valid l402 we first are going to call
-	// the FetchL402 method. As a client might not have outbound capacity
-	// yet, we'll retry until we get a valid response.
-	if !m.hasL402 {
-		m.fetchL402(m.runCtx)
-	}
-
-	ctx, cancel := context.WithCancel(m.runCtx)
-
-	// We'll now subscribe to the reservation notifications.
-	reservationStream, err := m.cfg.ReservationClient.
-		ReservationNotificationStream(
-			ctx, &reservationrpc.ReservationNotificationRequest{},
-		)
-	if err != nil {
-		cancel()
-		return err
-	}
-
-	log.Debugf("Successfully subscribed to reservation notifications")
-
-	// We'll now start a goroutine that will forward all the reservation
-	// notifications to the reservationChan.
-	go func() {
-		for {
-			reservationRes, err := reservationStream.Recv()
-			if err == nil && reservationRes != nil {
-				log.Debugf("Received reservation %x",
-					reservationRes.ReservationId)
-				reservationChan <- reservationRes
-				continue
-			}
-			log.Errorf("Error receiving "+
-				"reservation: %v", err)
-
-			cancel()
-
-			// If we encounter an error, we'll
-			// try to reconnect.
-			for {
-				select {
-				case <-m.runCtx.Done():
-					return
-
-				case <-time.After(time.Second * 10):
-					log.Debugf("Reconnecting to " +
-						"reservation notifications")
-					err = m.RegisterReservationNotifications(
-						reservationChan,
-					)
-					if err != nil {
-						log.Errorf("Error "+
-							"reconnecting: %v", err)
-						continue
-					}
-
-					// If we were able to reconnect, we'll
-					// return.
-					return
-				}
-			}
-		}
-	}()
-
-	return nil
-}
-
 // RecoverReservations tries to recover all reservations that are still active
 // from the database.
 func (m *Manager) RecoverReservations(ctx context.Context) error {

instantout/reservation/manager_test.go

@@ -13,7 +13,6 @@ import (
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
-	"google.golang.org/grpc"
 )
 
 var (
@@ -118,27 +117,22 @@ func newManagerTestContext(t *testing.T) *ManagerTestContext {
 
 	sendChan := make(chan *swapserverrpc.ServerReservationNotification)
 
-	mockReservationClient.On(
-		"ReservationNotificationStream", mock.Anything, mock.Anything,
-		mock.Anything,
-	).Return(
-		&dummyReservationNotificationServer{
-			SendChan: sendChan,
-		}, nil,
-	)
-
 	mockReservationClient.On(
 		"OpenReservation", mock.Anything, mock.Anything, mock.Anything,
 	).Return(
 		&swapserverrpc.ServerOpenReservationResponse{}, nil,
 	)
 
+	mockNtfnManager := &mockNtfnManager{
+		sendChan: sendChan,
+	}
+
 	cfg := &Config{
-		Store:             store,
-		Wallet:            mockLnd.WalletKit,
-		ChainNotifier:     mockLnd.ChainNotifier,
-		FetchL402:         func(context.Context) error { return nil },
-		ReservationClient: mockReservationClient,
+		Store:               store,
+		Wallet:              mockLnd.WalletKit,
+		ChainNotifier:       mockLnd.ChainNotifier,
+		ReservationClient:   mockReservationClient,
+		NotificationManager: mockNtfnManager,
 	}
 
 	manager := NewManager(cfg)
@@ -152,17 +146,15 @@ func newManagerTestContext(t *testing.T) *ManagerTestContext {
 	}
 }
 
-type dummyReservationNotificationServer struct {
-	grpc.ClientStream
-
-	// SendChan is the channel that is used to send notifications.
-	SendChan chan *swapserverrpc.ServerReservationNotification
+type mockNtfnManager struct {
+	sendChan chan *swapserverrpc.ServerReservationNotification
 }
 
-func (d *dummyReservationNotificationServer) Recv() (
-	*swapserverrpc.ServerReservationNotification, error) {
+func (m *mockNtfnManager) SubscribeReservations(
+	ctx context.Context,
+) <-chan *swapserverrpc.ServerReservationNotification {
 
-	return <-d.SendChan, nil
+	return m.sendChan
 }
 
 func mustDecodeID(id string) ID {