Merge pull request #1643 from lightninglabs/wip/supplycommit/add-multi-sm-manager

supplycommit: add MultiStateMachineManager for managing group state machines

Author: ffranr

config.go

@@ -17,6 +17,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/tapfreighter"
 	"github.com/lightninglabs/taproot-assets/tapgarden"
 	"github.com/lightninglabs/taproot-assets/universe"
+	"github.com/lightninglabs/taproot-assets/universe/supplycommit"
 	"github.com/lightningnetwork/lnd"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/signal"
@@ -186,6 +187,15 @@ 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.
+	SupplyCommitManager *supplycommit.MultiStateMachineManager
+
 	UniverseArchive *universe.Archive
 
 	UniverseSyncer universe.Syncer

chain_bridge.go renamed to lnd_services.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)

log.go

@@ -14,6 +14,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/tapgarden"
 	"github.com/lightninglabs/taproot-assets/tapsend"
 	"github.com/lightninglabs/taproot-assets/universe"
+	"github.com/lightninglabs/taproot-assets/universe/supplycommit"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/signal"
 )
@@ -109,6 +110,10 @@ func SetupLoggers(root *build.SubLoggerManager,
 	AddSubLogger(root, address.Subsystem, interceptor, address.UseLogger)
 	AddSubLogger(root, tapsend.Subsystem, interceptor, tapsend.UseLogger)
 	AddSubLogger(root, universe.Subsystem, interceptor, universe.UseLogger)
+	AddSubLogger(
+		root, supplycommit.Subsystem, interceptor,
+		supplycommit.UseLogger,
+	)
 	AddSubLogger(
 		root, commitment.Subsystem, interceptor, commitment.UseLogger,
 	)

server.go

@@ -211,6 +211,18 @@ 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",
+			err)
+	}
+
 	// Start the auxiliary components.
 	if err := s.cfg.AuxLeafSigner.Start(); err != nil {
 		return fmt.Errorf("unable to start aux leaf signer: %w", err)
@@ -691,6 +703,18 @@ 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",
+			err)
+	}
+
 	if err := s.cfg.AuxLeafSigner.Stop(); err != nil {
 		return err
 	}

tapcfg/server.go

@@ -23,6 +23,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/tapgarden"
 	"github.com/lightninglabs/taproot-assets/tapscript"
 	"github.com/lightninglabs/taproot-assets/universe"
+	"github.com/lightninglabs/taproot-assets/universe/supplycommit"
 	"github.com/lightningnetwork/lnd"
 	"github.com/lightningnetwork/lnd/clock"
 	"github.com/lightningnetwork/lnd/signal"
@@ -110,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 {
@@ -161,6 +163,7 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger,
 	groupVerifier := tapgarden.GenGroupVerifier(
 		context.Background(), assetMintingStore,
 	)
+
 	uniArchiveCfg := universe.ArchiveConfig{
 		// nolint: lll
 		NewBaseTree: func(id universe.Identifier) universe.StorageBackend {
@@ -583,6 +586,40 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger,
 			"access status: %w", err)
 	}
 
+	// Construct the supply commit manager, which is used to
+	// formulate universe supply commitment transactions.
+	//
+	// Construct database backends for the supply commitment state machines.
+	supplyCommitDb := tapdb.NewTransactionExecutor(
+		db, func(tx *sql.Tx) tapdb.SupplyCommitStore {
+			return db.WithTx(tx)
+		},
+	)
+	supplyCommitStore := tapdb.NewSupplyCommitMachine(supplyCommitDb)
+
+	// Construct the supply tree database backend.
+	supplyTreeDb := tapdb.NewTransactionExecutor(
+		db, func(tx *sql.Tx) tapdb.BaseUniverseStore {
+			return db.WithTx(tx)
+		},
+	)
+	supplyTreeStore := tapdb.NewSupplyTreeStore(supplyTreeDb)
+
+	// Create the supply commitment state machine manager, which is used to
+	// 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,
+			DaemonAdapters: lndFsmDaemonAdapters,
+			StateLog:       supplyCommitStore,
+			ChainParams:    *tapChainParams.Params,
+		},
+	)
+
 	return &tap.Config{
 		DebugLevel:            cfg.DebugLevel,
 		RuntimeID:             runtimeID,
@@ -633,6 +670,8 @@ func genServerConfig(cfg *Config, cfgLogger btclog.Logger,
 		AssetWallet:              assetWallet,
 		CoinSelect:               coinSelect,
 		ChainPorter:              chainPorter,
+		FsmDaemonAdapters:        lndFsmDaemonAdapters,
+		SupplyCommitManager:      supplyCommitManager,
 		UniverseArchive:          uniArchive,
 		UniverseSyncer:           universeSyncer,
 		UniverseFederation:       universeFederation,