multi: add LndFsmDaemonAdapters for state machine manager

Introduce LndFsmDaemonAdapters, a struct implementing the
protofsm.DaemonAdapters interface. This is passed into
supplycommit.NewMultiStateMachineManager, which then propagates it to
each asset group-specific state machine.

Author: ffranr

chain_bridge.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"sync"
+	"time"
 
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
@@ -22,6 +24,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnrpc/routerrpc"
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/protofsm"
 	"github.com/lightningnetwork/lnd/routing/route"
 )
 
@@ -31,6 +34,10 @@ const (
 	// 3200kB of memory (4 bytes for the block height and 4 bytes for the
 	// timestamp, not including any map/cache overhead).
 	maxNumBlocksInCache = 400_000
+
+	// DefaultTimeout is the default timeout we use for RPC and database
+	// operations.
+	DefaultTimeout = 30 * time.Second
 )
 
 // cacheableTimestamp is a wrapper around an uint32 that can be used as a value
@@ -458,3 +465,184 @@ func (l *LndInvoicesClient) HtlcModifier(ctx context.Context,
 // Ensure LndInvoicesClient implements the tapchannel.InvoiceHtlcModifier
 // interface.
 var _ tapchannel.InvoiceHtlcModifier = (*LndInvoicesClient)(nil)
+
+// LndFsmDaemonAdapters is a struct that implements the protofsm.DaemonAdapters
+// interface.
+type LndFsmDaemonAdapters struct {
+	// lnd is the LND services client that will be used to interact with
+	// the LND node.
+	lnd *lndclient.LndServices
+
+	// retryConfig is the retry configuration that will be used for
+	// operations that may fail due to temporary issues, such as network
+	// errors or RPC timeouts.
+	retryConfig fn.RetryConfig
+
+	// msgTransport is the message transport client that will be used to
+	// send messages to peers.
+	msgTransport LndMsgTransportClient
+
+	// chainBridge is the chain bridge that will be used to interact with
+	// the blockchain.
+	chainBridge LndRpcChainBridge
+
+	// ContextGuard manages the context and quit channel for this service.
+	fn.ContextGuard
+
+	startOnce sync.Once
+	stopOnce  sync.Once
+}
+
+// NewLndFsmDaemonAdapters creates a new instance of LndFsmDaemonAdapters.
+func NewLndFsmDaemonAdapters(lnd *lndclient.LndServices) *LndFsmDaemonAdapters {
+	retryConfig := fn.DefaultRetryConfig()
+
+	msgTransport := NewLndMsgTransportClient(lnd)
+
+	// Initialize the chain bridge without the asset store, as it is not
+	// needed for the FSM adapters.
+	chainBridge := NewLndRpcChainBridge(lnd, nil)
+	chainBridge.retryConfig = retryConfig
+
+	return &LndFsmDaemonAdapters{
+		lnd:          lnd,
+		retryConfig:  retryConfig,
+		msgTransport: *msgTransport,
+		chainBridge:  *chainBridge,
+		ContextGuard: fn.ContextGuard{
+			DefaultTimeout: DefaultTimeout,
+			Quit:           make(chan struct{}),
+		},
+	}
+}
+
+// Start attempts to start the service.
+func (l *LndFsmDaemonAdapters) Start() error {
+	l.startOnce.Do(func() {})
+	return nil
+}
+
+// Stop signals for the service to gracefully exit.
+func (l *LndFsmDaemonAdapters) Stop() error {
+	l.stopOnce.Do(func() {
+		// Signal the quit channel which will cancel all active
+		// contexts.
+		close(l.Quit)
+	})
+
+	return nil
+}
+
+// SendMessages sends a slice of lnwire.Message to the peer with the given
+// public key.
+func (l *LndFsmDaemonAdapters) SendMessages(peer btcec.PublicKey,
+	messages []lnwire.Message) error {
+
+	// Convert messages to a slice of CustomMessage.
+	customMessages := make([]lndclient.CustomMessage, 0, len(messages))
+	for idx := range messages {
+		msg := messages[idx]
+
+		var buf bytes.Buffer
+		if err := msg.Encode(&buf, 0); err != nil {
+			return fmt.Errorf("unable to encode message: %w", err)
+		}
+
+		customMsg := lndclient.CustomMessage{
+			Peer:    route.NewVertex(&peer),
+			MsgType: uint32(msg.MsgType()),
+			Data:    buf.Bytes(),
+		}
+
+		customMessages = append(customMessages, customMsg)
+	}
+
+	ctx, cancel := l.WithCtxQuitNoTimeout()
+	defer cancel()
+
+	// Send each message in turn.
+	for idx := range customMessages {
+		msg := customMessages[idx]
+
+		err := l.msgTransport.SendCustomMessage(ctx, msg)
+		if err != nil {
+			return fmt.Errorf("unable to send custom message: %w",
+				err)
+		}
+	}
+
+	return nil
+}
+
+// BroadcastTransaction attempts to broadcast a transaction to the
+// network. It uses the chain bridge to publish the transaction.
+func (l *LndFsmDaemonAdapters) BroadcastTransaction(tx *wire.MsgTx,
+	label string) error {
+
+	ctx, cancel := l.WithCtxQuitNoTimeout()
+	defer cancel()
+
+	return l.chainBridge.PublishTransaction(ctx, tx, label)
+}
+
+// RegisterConfirmationsNtfn registers an intent to be notified once the
+// transaction with the given txid reaches the specified number of
+// confirmations.
+func (l *LndFsmDaemonAdapters) RegisterConfirmationsNtfn(
+	txid *chainhash.Hash, pkScript []byte, numConfs uint32,
+	heightHint uint32,
+	optFuncs ...chainntnfs.NotifierOption) (*chainntnfs.ConfirmationEvent,
+	error) {
+
+	opts := chainntnfs.DefaultNotifierOptions()
+	for _, optFunc := range optFuncs {
+		optFunc(opts)
+	}
+
+	lndCliOpt := make([]lndclient.NotifierOption, 0, len(optFuncs))
+	if opts.IncludeBlock {
+		lndCliOpt = append(lndCliOpt, lndclient.WithIncludeBlock())
+	}
+
+	ctx, cancel := l.WithCtxQuitNoTimeout()
+	spendDetail, _, err := l.lnd.ChainNotifier.RegisterConfirmationsNtfn(
+		ctx, txid, pkScript, int32(numConfs), int32(heightHint),
+		lndCliOpt...,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("unable to register for conf: %w", err)
+	}
+
+	return &chainntnfs.ConfirmationEvent{
+		Confirmed:    spendDetail,
+		Updates:      make(chan uint32),
+		NegativeConf: make(chan int32),
+		Done:         make(chan struct{}),
+		Cancel:       cancel,
+	}, nil
+}
+
+// RegisterSpendNtfn registers an intent to be notified once the outpoint
+// is spent on-chain.
+func (l *LndFsmDaemonAdapters) RegisterSpendNtfn(outpoint *wire.OutPoint,
+	pkScript []byte, heightHint uint32) (*chainntnfs.SpendEvent, error) {
+
+	ctx, cancel := l.WithCtxQuitNoTimeout()
+	spendDetail, _, err := l.lnd.ChainNotifier.RegisterSpendNtfn(
+		ctx, outpoint, pkScript, int32(heightHint),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("unable to register for spend: %w", err)
+	}
+
+	return &chainntnfs.SpendEvent{
+		Spend:  spendDetail,
+		Reorg:  make(chan struct{}, 1),
+		Done:   make(chan struct{}, 1),
+		Cancel: cancel,
+	}, nil
+}
+
+// Ensure LndFsmDaemonAdapters implements the protofsm.DaemonAdapters
+// interface.
+var _ protofsm.DaemonAdapters = (*LndFsmDaemonAdapters)(nil)

config.go

@@ -187,6 +187,10 @@ type Config struct {
 
 	ChainPorter tapfreighter.Porter
 
+	// FsmDaemonAdapters is a set of adapters that allow a state machine to
+	// interact with external daemons.
+	FsmDaemonAdapters *LndFsmDaemonAdapters
+
 	// SupplyCommitManager is a service that is used to manage supply
 	// commitments for assets. Supply commitments are issuer published
 	// attestations of the total supply of an asset.

server.go

@@ -211,6 +211,12 @@ func (s *Server) initialize(interceptorChain *rpcperms.InterceptorChain) error {
 		return fmt.Errorf("unable to start RFQ manager: %w", err)
 	}
 
+	// Start FSM daemon adapters.
+	if err := s.cfg.FsmDaemonAdapters.Start(); err != nil {
+		return fmt.Errorf("unable to start FSM daemon adapters: %w",
+			err)
+	}
+
 	// Start universe supply commitment manager.
 	if err := s.cfg.SupplyCommitManager.Start(); err != nil {
 		return fmt.Errorf("unable to start supply commit manager: %w",
@@ -697,6 +703,12 @@ func (s *Server) Stop() error {
 		return err
 	}
 
+	// Stop FSM daemon adapters.
+	if err := s.cfg.FsmDaemonAdapters.Stop(); err != nil {
+		return fmt.Errorf("unable to stop FSM daemon adapters: %w",
+			err)
+	}
+
 	// Stop universe supply commitment manager.
 	if err := s.cfg.SupplyCommitManager.Stop(); err != nil {
 		return fmt.Errorf("unable to stop supply commit manager: %w",

tapcfg/server.go

@@ -111,6 +111,7 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger,
 	lndRouterClient := tap.NewLndRouterClient(lndServices)
 	lndInvoicesClient := tap.NewLndInvoicesClient(lndServices)
 	lndFeatureBitsVerifier := tap.NewLndFeatureBitVerifier(lndServices)
+	lndFsmDaemonAdapters := tap.NewLndFsmDaemonAdapters(lndServices)
 
 	uniDB := tapdb.NewTransactionExecutor(
 		db, func(tx *sql.Tx) tapdb.BaseUniverseStore {
@@ -608,13 +609,14 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger,
 	// manage the supply commitment state machines for each asset group.
 	supplyCommitManager := supplycommit.NewMultiStateMachineManager(
 		supplycommit.MultiStateMachineManagerCfg{
-			TreeView:    supplyTreeStore,
-			Commitments: supplyCommitStore,
-			Wallet:      walletAnchor,
-			KeyRing:     keyRing,
-			Chain:       chainBridge,
-			StateLog:    supplyCommitStore,
-			ChainParams: *tapChainParams.Params,
+			TreeView:       supplyTreeStore,
+			Commitments:    supplyCommitStore,
+			Wallet:         walletAnchor,
+			KeyRing:        keyRing,
+			Chain:          chainBridge,
+			DaemonAdapters: lndFsmDaemonAdapters,
+			StateLog:       supplyCommitStore,
+			ChainParams:    *tapChainParams.Params,
 		},
 	)
 
@@ -668,6 +670,7 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger,
 		AssetWallet:              assetWallet,
 		CoinSelect:               coinSelect,
 		ChainPorter:              chainPorter,
+		FsmDaemonAdapters:        lndFsmDaemonAdapters,
 		SupplyCommitManager:      supplyCommitManager,
 		UniverseArchive:          uniArchive,
 		UniverseSyncer:           universeSyncer,

universe/supplycommit/multi_sm_manager.go

@@ -21,6 +21,18 @@ const (
 	DefaultTimeout = 30 * time.Second
 )
 
+// DaemonAdapters is a wrapper around the protofsm.DaemonAdapters interface
+// with the addition of Start and Stop methods.
+type DaemonAdapters interface {
+	protofsm.DaemonAdapters
+
+	// Start starts the daemon adapters handler service.
+	Start() error
+
+	// Stop stops the daemon adapters handler service.
+	Stop() error
+}
+
 // MultiStateMachineManagerCfg is the configuration for the
 // MultiStateMachineManager. It contains all the dependencies needed to
 // manage multiple supply commitment state machines, one for each asset group.
@@ -45,6 +57,10 @@ type MultiStateMachineManagerCfg struct {
 	// TODO(roasbeef): can make a slimmer version of
 	Chain tapgarden.ChainBridge
 
+	// DaemonAdapters is a set of adapters that allow the state machine to
+	// interact with external daemons whilst processing internal events.
+	DaemonAdapters DaemonAdapters
+
 	// StateLog is the main state log that is used to track the state of the
 	// state machine. This is used to persist the state of the state machine
 	// across restarts.
@@ -150,6 +166,7 @@ func (m *MultiStateMachineManager) fetchStateMachine(
 	fsmCfg := protofsm.StateMachineCfg[Event, *Environment]{
 		InitialState: initialState,
 		Env:          env,
+		Daemon:       m.cfg.DaemonAdapters,
 	}
 	newSm := protofsm.NewStateMachine[Event, *Environment](fsmCfg)