staticaddr: serialize address creation with a channel

Author: hieblmi

loopd/swapclient_server.go

@@ -1579,14 +1579,21 @@ func (s *swapClientServer) NewStaticAddress(ctx context.Context,
 	_ *looprpc.NewStaticAddressRequest) (
 	*looprpc.NewStaticAddressResponse, error) {
 
-	staticAddress, expiry, err := s.staticAddressManager.NewAddress(ctx)
+	params, err := s.staticAddressManager.NewAddress(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	address, err := s.staticAddressManager.GetTaprootAddress(
+		params.ClientPubkey, params.ServerPubkey, int64(params.Expiry),
+	)
 	if err != nil {
 		return nil, err
 	}
 
 	return &looprpc.NewStaticAddressResponse{
-		Address: staticAddress.String(),
-		Expiry:  uint32(expiry),
+		Address: address.String(),
+		Expiry:  params.Expiry,
 	}, nil
 }
 

staticaddr/address/manager.go

@@ -54,19 +54,68 @@ type Manager struct {
 	cfg *ManagerConfig
 
 	currentHeight atomic.Int32
+
+	// addrRequest is a channel used to request new static addresses from
+	// the manager. The manager employs a go worker routine that handles the
+	// requests.
+	addrRequest chan request
+}
+
+type request struct {
+	ctx      context.Context
+	respChan chan response
+}
+
+type response struct {
+	parameters *Parameters
+	err        error
 }
 
 // NewManager creates a new address manager.
 func NewManager(cfg *ManagerConfig, currentHeight int32) *Manager {
 	m := &Manager{
-		cfg: cfg,
+		cfg:         cfg,
+		addrRequest: make(chan request),
 	}
 	m.currentHeight.Store(currentHeight)
 
 	return m
 }
 
-// Run runs the address manager.
+// addrWorker is a worker that handles address creation requests. It calls
+// m.newAddress which blocks on server I/O and returns the address and expiry.
+func (m *Manager) addrWorker(ctx context.Context) {
+	for {
+		select {
+		case req := <-m.addrRequest:
+			m.handleAddrRequest(ctx, req)
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// handleAddrRequest is responsible for processing a single address request.
+func (m *Manager) handleAddrRequest(managerCtx context.Context, req request) {
+	addrParams, err := m.newAddress(req.ctx)
+
+	resp := response{
+		parameters: addrParams,
+		err:        err,
+	}
+
+	select {
+	case req.respChan <- resp:
+
+	case <-req.ctx.Done():
+
+	case <-managerCtx.Done():
+	}
+}
+
+// Run runs the address manager. It keeps track of the current block height and
+// creates new static addresses as needed.
 func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 	newBlockChan, newBlockErrChan, err :=
 		m.cfg.ChainNotifier.RegisterBlockEpochNtfn(ctx)
@@ -75,6 +124,10 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 		return err
 	}
 
+	// The address worker offloads the address creation with the server to a
+	// separate go routine.
+	go m.addrWorker(ctx)
+
 	// Communicate to the caller that the address manager has completed its
 	// initialization.
 	close(initChan)
@@ -95,49 +148,64 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 }
 
 // NewAddress creates a new static address with the server or returns an
-// existing one.
-func (m *Manager) NewAddress(ctx context.Context) (*btcutil.AddressTaproot,
-	int64, error) {
+// existing one. It now sends a request to the manager's Run loop which
+// executes the actual address creation logic.
+func (m *Manager) NewAddress(ctx context.Context) (*Parameters, error) {
+	respChan := make(chan response, 1)
+	req := request{
+		ctx:      ctx,
+		respChan: respChan,
+	}
+
+	// Send the new address request to the manager run loop.
+	select {
+	case m.addrRequest <- req:
 
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+
+	// Wait for the response from the manager run loop.
+	select {
+	case resp := <-respChan:
+		return resp.parameters, resp.err
+
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+// newAddress contains the body of the former NewAddress method and performs the
+// actual address creation/lookup according to the requested type.
+func (m *Manager) newAddress(ctx context.Context) (*Parameters, error) {
 	// If there's already a static address in the database, we can return
 	// it.
 	m.Lock()
 	addresses, err := m.cfg.Store.GetAllStaticAddresses(ctx)
 	if err != nil {
 		m.Unlock()
 
-		return nil, 0, err
+		return nil, err
 	}
 	if len(addresses) > 0 {
-		clientPubKey := addresses[0].ClientPubkey
-		serverPubKey := addresses[0].ServerPubkey
-		expiry := int64(addresses[0].Expiry)
-
-		defer m.Unlock()
-
-		address, err := m.GetTaprootAddress(
-			clientPubKey, serverPubKey, expiry,
-		)
-		if err != nil {
-			return nil, 0, err
-		}
+		m.Unlock()
 
-		return address, expiry, nil
+		return addresses[0], nil
 	}
 	m.Unlock()
 
 	// We are fetching a new L402 token from the server. There is one static
 	// address per L402 token allowed.
 	err = m.cfg.FetchL402(ctx)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	clientPubKey, err := m.cfg.WalletKit.DeriveNextKey(
 		ctx, swap.StaticAddressKeyFamily,
 	)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	// Send our clientPubKey to the server and wait for the server to
@@ -150,27 +218,27 @@ func (m *Manager) NewAddress(ctx context.Context) (*btcutil.AddressTaproot,
 		},
 	)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	serverParams := resp.GetParams()
 
 	serverPubKey, err := btcec.ParsePubKey(serverParams.ServerKey)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	staticAddress, err := script.NewStaticAddress(
 		input.MuSig2Version100RC2, int64(serverParams.Expiry),
 		clientPubKey.PubKey, serverPubKey,
 	)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	pkScript, err := staticAddress.StaticAddressScript()
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	// Create the static address from the parameters the server provided and
@@ -191,7 +259,7 @@ func (m *Manager) NewAddress(ctx context.Context) (*btcutil.AddressTaproot,
 	}
 	err = m.cfg.Store.CreateStaticAddress(ctx, addrParams)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	// Import the static address tapscript into our lnd wallet, so we can
@@ -201,20 +269,13 @@ func (m *Manager) NewAddress(ctx context.Context) (*btcutil.AddressTaproot,
 	)
 	addr, err := m.cfg.WalletKit.ImportTaprootScript(ctx, tapScript)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	log.Infof("Imported static address taproot script to lnd wallet: %v",
 		addr)
 
-	address, err := m.GetTaprootAddress(
-		clientPubKey.PubKey, serverPubKey, int64(serverParams.Expiry),
-	)
-	if err != nil {
-		return nil, 0, err
-	}
-
-	return address, int64(serverParams.Expiry), nil
+	return addrParams, nil
 }
 
 // GetTaprootAddress returns a taproot address for the given client and server

staticaddr/address/manager_test.go

@@ -108,14 +108,19 @@ func TestManager(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create a new static address.
-	taprootAddress, expiry, err := testContext.manager.NewAddress(ctxb)
+	params, err := testContext.manager.NewAddress(ctxb)
+	require.NoError(t, err)
+
+	address, err := testContext.manager.GetTaprootAddress(
+		params.ClientPubkey, params.ServerPubkey, int64(params.Expiry),
+	)
 	require.NoError(t, err)
 
 	// The addresses have to match.
-	require.Equal(t, expectedAddress.String(), taprootAddress.String())
+	require.Equal(t, expectedAddress.String(), address.String())
 
 	// The expiry has to match.
-	require.EqualValues(t, defaultExpiry, expiry)
+	require.EqualValues(t, defaultExpiry, params.Expiry)
 }
 
 // GenerateExpectedTaprootAddress generates the expected taproot address that