12583319996

Committed 02 Jan 2025 01:38PM UTC coverage: 57.522% (-1.1%) from 58.598%

Build # 12583319996

Build Type

Pull #9361

github

Committed by

starius

Commit Message

fn/ContextGuard: use context.AfterFunc to wait

Simplifies context cancellation handling by using context.AfterFunc instead of a
goroutine to wait for context cancellation. This approach avoids the overhead of
a goroutine during the waiting period.

For ctxQuitUnsafe, since g.quit is closed only in the Quit method (which also
cancels all associated contexts), waiting on context cancellation ensures the
same behavior without unnecessary dependency on g.quit.

Added a test to ensure that the Create method does not launch any goroutines.

Pull Request Pull Request #9361: fn: optimize context guard

Run Details

102587 of 178344 relevant lines covered (57.52%)

24734.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

36.04

/contractcourt/chain_arbitrator.go

package contractcourt

import (
        "errors"
        "fmt"
        "sync"
        "sync/atomic"
        "time"

        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btcd/wire"
        "github.com/btcsuite/btcwallet/walletdb"
        "github.com/lightningnetwork/lnd/chainio"
        "github.com/lightningnetwork/lnd/chainntnfs"
        "github.com/lightningnetwork/lnd/channeldb"
        "github.com/lightningnetwork/lnd/clock"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/input"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/labels"
        "github.com/lightningnetwork/lnd/lnwallet"
        "github.com/lightningnetwork/lnd/lnwallet/chainfee"
        "github.com/lightningnetwork/lnd/lnwire"
)

// ErrChainArbExiting signals that the chain arbitrator is shutting down.
var ErrChainArbExiting = errors.New("ChainArbitrator exiting")

// ResolutionMsg is a message sent by resolvers to outside sub-systems once an
// outgoing contract has been fully resolved. For multi-hop contracts, if we
// resolve the outgoing contract, we'll also need to ensure that the incoming
// contract is resolved as well. We package the items required to resolve the
// incoming contracts within this message.
type ResolutionMsg struct {
        // SourceChan identifies the channel that this message is being sent
        // from. This is the channel's short channel ID.
        SourceChan lnwire.ShortChannelID

        // HtlcIndex is the index of the contract within the original
        // commitment trace.
        HtlcIndex uint64

        // Failure will be non-nil if the incoming contract should be canceled
        // all together. This can happen if the outgoing contract was dust, if
        // if the outgoing HTLC timed out.
        Failure lnwire.FailureMessage

        // PreImage will be non-nil if the incoming contract can successfully
        // be redeemed. This can happen if we learn of the preimage from the
        // outgoing HTLC on-chain.
        PreImage *[32]byte
}

// ChainArbitratorConfig is a configuration struct that contains all the
// function closures and interface that required to arbitrate on-chain
// contracts for a particular chain.
type ChainArbitratorConfig struct {
        // ChainHash is the chain that this arbitrator is to operate within.
        ChainHash chainhash.Hash

        // IncomingBroadcastDelta is the delta that we'll use to decide when to
        // broadcast our commitment transaction if we have incoming htlcs. This
        // value should be set based on our current fee estimation of the
        // commitment transaction. We use this to determine when we should
        // broadcast instead of just the HTLC timeout, as we want to ensure
        // that the commitment transaction is already confirmed, by the time the
        // HTLC expires. Otherwise we may end up not settling the htlc on-chain
        // because the other party managed to time it out.
        IncomingBroadcastDelta uint32

        // OutgoingBroadcastDelta is the delta that we'll use to decide when to
        // broadcast our commitment transaction if there are active outgoing
        // htlcs. This value can be lower than the incoming broadcast delta.
        OutgoingBroadcastDelta uint32

        // NewSweepAddr is a function that returns a new address under control
        // by the wallet. We'll use this to sweep any no-delay outputs as a
        // result of unilateral channel closes.
        //
        // NOTE: This SHOULD return a p2wkh script.
        NewSweepAddr func() ([]byte, error)

        // PublishTx reliably broadcasts a transaction to the network. Once
        // this function exits without an error, then they transaction MUST
        // continually be rebroadcast if needed.
        PublishTx func(*wire.MsgTx, string) error

        // DeliverResolutionMsg is a function that will append an outgoing
        // message to the "out box" for a ChannelLink. This is used to cancel
        // backwards any HTLC's that are either dust, we're timing out, or
        // settling on-chain to the incoming link.
        DeliverResolutionMsg func(...ResolutionMsg) error

        // MarkLinkInactive is a function closure that the ChainArbitrator will
        // use to mark that active HTLC's shouldn't be attempted to be routed
        // over a particular channel. This function will be called in that a
        // ChannelArbitrator decides that it needs to go to chain in order to
        // resolve contracts.
        //
        // TODO(roasbeef): rename, routing based
        MarkLinkInactive func(wire.OutPoint) error

        // ContractBreach is a function closure that the ChainArbitrator will
        // use to notify the BreachArbitrator about a contract breach. It should
        // only return a non-nil error when the BreachArbitrator has preserved
        // the necessary breach info for this channel point. Once the breach
        // resolution is persisted in the ChannelArbitrator, it will be safe
        // to mark the channel closed.
        ContractBreach func(wire.OutPoint, *lnwallet.BreachRetribution) error

        // IsOurAddress is a function that returns true if the passed address
        // is known to the underlying wallet. Otherwise, false should be
        // returned.
        IsOurAddress func(btcutil.Address) bool

        // IncubateOutputs sends either an incoming HTLC, an outgoing HTLC, or
        // both to the utxo nursery. Once this function returns, the nursery
        // should have safely persisted the outputs to disk, and should start
        // the process of incubation. This is used when a resolver wishes to
        // pass off the output to the nursery as we're only waiting on an
        // absolute/relative item block.
        IncubateOutputs func(wire.OutPoint,
                fn.Option[lnwallet.OutgoingHtlcResolution],
                fn.Option[lnwallet.IncomingHtlcResolution],
                uint32, fn.Option[int32]) error

        // PreimageDB is a global store of all known pre-images. We'll use this
        // to decide if we should broadcast a commitment transaction to claim
        // an HTLC on-chain.
        PreimageDB WitnessBeacon

        // Notifier is an instance of a chain notifier we'll use to watch for
        // certain on-chain events.
        Notifier chainntnfs.ChainNotifier

        // Mempool is the a mempool watcher that allows us to watch for events
        // happened in mempool.
        Mempool chainntnfs.MempoolWatcher

        // Signer is a signer backed by the active lnd node. This should be
        // capable of producing a signature as specified by a valid
        // SignDescriptor.
        Signer input.Signer

        // FeeEstimator will be used to return fee estimates.
        FeeEstimator chainfee.Estimator

        // ChainIO allows us to query the state of the current main chain.
        ChainIO lnwallet.BlockChainIO

        // DisableChannel disables a channel, resulting in it not being able to
        // forward payments.
        DisableChannel func(wire.OutPoint) error

        // Sweeper allows resolvers to sweep their final outputs.
        Sweeper UtxoSweeper

        // Registry is the invoice database that is used by resolvers to lookup
        // preimages and settle invoices.
        Registry Registry

        // NotifyClosedChannel is a function closure that the ChainArbitrator
        // will use to notify the ChannelNotifier about a newly closed channel.
        NotifyClosedChannel func(wire.OutPoint)

        // NotifyFullyResolvedChannel is a function closure that the
        // ChainArbitrator will use to notify the ChannelNotifier about a newly
        // resolved channel. The main difference to NotifyClosedChannel is that
        // in case of a local force close the NotifyClosedChannel is called when
        // the published commitment transaction confirms while
        // NotifyFullyResolvedChannel is only called when the channel is fully
        // resolved (which includes sweeping any time locked funds).
        NotifyFullyResolvedChannel func(point wire.OutPoint)

        // OnionProcessor is used to decode onion payloads for on-chain
        // resolution.
        OnionProcessor OnionProcessor

        // PaymentsExpirationGracePeriod indicates a time window we let the
        // other node to cancel an outgoing htlc that our node has initiated and
        // has timed out.
        PaymentsExpirationGracePeriod time.Duration

        // IsForwardedHTLC checks for a given htlc, identified by channel id and
        // htlcIndex, if it is a forwarded one.
        IsForwardedHTLC func(chanID lnwire.ShortChannelID, htlcIndex uint64) bool

        // Clock is the clock implementation that ChannelArbitrator uses.
        // It is useful for testing.
        Clock clock.Clock

        // SubscribeBreachComplete is used by the breachResolver to register a
        // subscription that notifies when the breach resolution process is
        // complete.
        SubscribeBreachComplete func(op *wire.OutPoint, c chan struct{}) (
                bool, error)

        // PutFinalHtlcOutcome stores the final outcome of an htlc in the
        // database.
        PutFinalHtlcOutcome func(chanId lnwire.ShortChannelID,
                htlcId uint64, settled bool) error

        // HtlcNotifier is an interface that htlc events are sent to.
        HtlcNotifier HtlcNotifier

        // Budget is the configured budget for the arbitrator.
        Budget BudgetConfig

        // QueryIncomingCircuit is used to find the outgoing HTLC's
        // corresponding incoming HTLC circuit. It queries the circuit map for
        // a given outgoing circuit key and returns the incoming circuit key.
        //
        // TODO(yy): this is a hacky way to get around the cycling import issue
        // as we cannot import `htlcswitch` here. A proper way is to define an
        // interface here that asks for method `LookupOpenCircuit`,
        // meanwhile, turn `PaymentCircuit` into an interface or bring it to a
        // lower package.
        QueryIncomingCircuit func(circuit models.CircuitKey) *models.CircuitKey

        // AuxLeafStore is an optional store that can be used to store auxiliary
        // leaves for certain custom channel types.
        AuxLeafStore fn.Option[lnwallet.AuxLeafStore]

        // AuxSigner is an optional signer that can be used to sign auxiliary
        // leaves for certain custom channel types.
        AuxSigner fn.Option[lnwallet.AuxSigner]

        // AuxResolver is an optional interface that can be used to modify the
        // way contracts are resolved.
        AuxResolver fn.Option[lnwallet.AuxContractResolver]
}

// ChainArbitrator is a sub-system that oversees the on-chain resolution of all
// active, and channel that are in the "pending close" state. Within the
// contractcourt package, the ChainArbitrator manages a set of active
// ContractArbitrators. Each ContractArbitrators is responsible for watching
// the chain for any activity that affects the state of the channel, and also
// for monitoring each contract in order to determine if any on-chain activity is
// required. Outside sub-systems interact with the ChainArbitrator in order to
// forcibly exit a contract, update the set of live signals for each contract,
// and to receive reports on the state of contract resolution.
type ChainArbitrator struct {
        started int32 // To be used atomically.
        stopped int32 // To be used atomically.

        // Embed the blockbeat consumer struct to get access to the method
        // `NotifyBlockProcessed` and the `BlockbeatChan`.
        chainio.BeatConsumer

        sync.Mutex

        // activeChannels is a map of all the active contracts that are still
        // open, and not fully resolved.
        activeChannels map[wire.OutPoint]*ChannelArbitrator

        // activeWatchers is a map of all the active chainWatchers for channels
        // that are still considered open.
        activeWatchers map[wire.OutPoint]*chainWatcher

        // cfg is the config struct for the arbitrator that contains all
        // methods and interface it needs to operate.
        cfg ChainArbitratorConfig

        // chanSource will be used by the ChainArbitrator to fetch all the
        // active channels that it must still watch over.
        chanSource *channeldb.DB

        // beat is the current best known blockbeat.
        beat chainio.Blockbeat

        quit chan struct{}

        wg sync.WaitGroup
}

// NewChainArbitrator returns a new instance of the ChainArbitrator using the
// passed config struct, and backing persistent database.
func NewChainArbitrator(cfg ChainArbitratorConfig,
        db *channeldb.DB) *ChainArbitrator {

        c := &ChainArbitrator{
                cfg:            cfg,
                activeChannels: make(map[wire.OutPoint]*ChannelArbitrator),
                activeWatchers: make(map[wire.OutPoint]*chainWatcher),
                chanSource:     db,
                quit:           make(chan struct{}),
        }

        // Mount the block consumer.
        c.BeatConsumer = chainio.NewBeatConsumer(c.quit, c.Name())

        return c
}

// Compile-time check for the chainio.Consumer interface.
var _ chainio.Consumer = (*ChainArbitrator)(nil)

// arbChannel is a wrapper around an open channel that channel arbitrators
// interact with.
type arbChannel struct {
        // channel is the in-memory channel state.
        channel *channeldb.OpenChannel

        // c references the chain arbitrator and is used by arbChannel
        // internally.
        c *ChainArbitrator
}

// NewAnchorResolutions returns the anchor resolutions for currently valid
// commitment transactions.
//
// NOTE: Part of the ArbChannel interface.
func (a *arbChannel) NewAnchorResolutions() (*lnwallet.AnchorResolutions,
        error) {

        // Get a fresh copy of the database state to base the anchor resolutions
        // on. Unfortunately the channel instance that we have here isn't the
        // same instance that is used by the link.
        chanPoint := a.channel.FundingOutpoint

        channel, err := a.c.chanSource.ChannelStateDB().FetchChannel(chanPoint)
        if err != nil {
                return nil, err
        }

        var chanOpts []lnwallet.ChannelOpt
        a.c.cfg.AuxLeafStore.WhenSome(func(s lnwallet.AuxLeafStore) {
                chanOpts = append(chanOpts, lnwallet.WithLeafStore(s))
        })
        a.c.cfg.AuxSigner.WhenSome(func(s lnwallet.AuxSigner) {
                chanOpts = append(chanOpts, lnwallet.WithAuxSigner(s))
        })
        a.c.cfg.AuxResolver.WhenSome(func(s lnwallet.AuxContractResolver) {
                chanOpts = append(chanOpts, lnwallet.WithAuxResolver(s))
        })

        chanMachine, err := lnwallet.NewLightningChannel(
                a.c.cfg.Signer, channel, nil, chanOpts...,
        )
        if err != nil {
                return nil, err
        }

        return chanMachine.NewAnchorResolutions()
}

// ForceCloseChan should force close the contract that this attendant is
// watching over. We'll use this when we decide that we need to go to chain. It
// should in addition tell the switch to remove the corresponding link, such
// that we won't accept any new updates.
//
// NOTE: Part of the ArbChannel interface.
func (a *arbChannel) ForceCloseChan() (*wire.MsgTx, error) {
        // First, we mark the channel as borked, this ensure
        // that no new state transitions can happen, and also
        // that the link won't be loaded into the switch.
        if err := a.channel.MarkBorked(); err != nil {
                return nil, err
        }

        // With the channel marked as borked, we'll now remove
        // the link from the switch if its there. If the link
        // is active, then this method will block until it
        // exits.
        chanPoint := a.channel.FundingOutpoint

        if err := a.c.cfg.MarkLinkInactive(chanPoint); err != nil {
                log.Errorf("unable to mark link inactive: %v", err)
        }

        // Now that we know the link can't mutate the channel
        // state, we'll read the channel from disk the target
        // channel according to its channel point.
        channel, err := a.c.chanSource.ChannelStateDB().FetchChannel(chanPoint)
        if err != nil {
                return nil, err
        }

        var chanOpts []lnwallet.ChannelOpt
        a.c.cfg.AuxLeafStore.WhenSome(func(s lnwallet.AuxLeafStore) {
                chanOpts = append(chanOpts, lnwallet.WithLeafStore(s))
        })
        a.c.cfg.AuxSigner.WhenSome(func(s lnwallet.AuxSigner) {
                chanOpts = append(chanOpts, lnwallet.WithAuxSigner(s))
        })
        a.c.cfg.AuxResolver.WhenSome(func(s lnwallet.AuxContractResolver) {
                chanOpts = append(chanOpts, lnwallet.WithAuxResolver(s))
        })

        // Finally, we'll force close the channel completing
        // the force close workflow.
        chanMachine, err := lnwallet.NewLightningChannel(
                a.c.cfg.Signer, channel, nil, chanOpts...,
        )
        if err != nil {
                return nil, err
        }

        closeSummary, err := chanMachine.ForceClose(
                lnwallet.WithSkipContractResolutions(),
        )
        if err != nil {
                return nil, err
        }

        return closeSummary.CloseTx, nil
}

// newActiveChannelArbitrator creates a new instance of an active channel
// arbitrator given the state of the target channel.
func newActiveChannelArbitrator(channel *channeldb.OpenChannel,
        c *ChainArbitrator, chanEvents *ChainEventSubscription) (*ChannelArbitrator, error) {

        // TODO(roasbeef): fetch best height (or pass in) so can ensure block
        // epoch delivers all the notifications to

        chanPoint := channel.FundingOutpoint

        log.Tracef("Creating ChannelArbitrator for ChannelPoint(%v)", chanPoint)

        // Next we'll create the matching configuration struct that contains
        // all interfaces and methods the arbitrator needs to do its job.
        arbCfg := ChannelArbitratorConfig{
                ChanPoint:   chanPoint,
                Channel:     c.getArbChannel(channel),
                ShortChanID: channel.ShortChanID(),

                MarkCommitmentBroadcasted: channel.MarkCommitmentBroadcasted,
                MarkChannelClosed: func(summary *channeldb.ChannelCloseSummary,
                        statuses ...channeldb.ChannelStatus) error {

                        err := channel.CloseChannel(summary, statuses...)
                        if err != nil {
                                return err
                        }
                        c.cfg.NotifyClosedChannel(summary.ChanPoint)
                        return nil
                },
                IsPendingClose:        false,
                ChainArbitratorConfig: c.cfg,
                ChainEvents:           chanEvents,
                PutResolverReport: func(tx kvdb.RwTx,
                        report *channeldb.ResolverReport) error {

                        return c.chanSource.PutResolverReport(
                                tx, c.cfg.ChainHash, &chanPoint, report,
                        )
                },
                FetchHistoricalChannel: func() (*channeldb.OpenChannel, error) {
                        chanStateDB := c.chanSource.ChannelStateDB()
                        return chanStateDB.FetchHistoricalChannel(&chanPoint)
                },
                FindOutgoingHTLCDeadline: func(
                        htlc channeldb.HTLC) fn.Option[int32] {

                        return c.FindOutgoingHTLCDeadline(
                                channel.ShortChanID(), htlc,
                        )
                },
        }

        // The final component needed is an arbitrator log that the arbitrator
        // will use to keep track of its internal state using a backed
        // persistent log.
        //
        // TODO(roasbeef); abstraction leak...
        //  * rework: adaptor method to set log scope w/ factory func
        chanLog, err := newBoltArbitratorLog(
                c.chanSource.Backend, arbCfg, c.cfg.ChainHash, chanPoint,
        )
        if err != nil {
                return nil, err
        }

        arbCfg.MarkChannelResolved = func() error {
                if c.cfg.NotifyFullyResolvedChannel != nil {
                        c.cfg.NotifyFullyResolvedChannel(chanPoint)
                }

                return c.ResolveContract(chanPoint)
        }

        // Finally, we'll need to construct a series of htlc Sets based on all
        // currently known valid commitments.
        htlcSets := make(map[HtlcSetKey]htlcSet)
        htlcSets[LocalHtlcSet] = newHtlcSet(channel.LocalCommitment.Htlcs)
        htlcSets[RemoteHtlcSet] = newHtlcSet(channel.RemoteCommitment.Htlcs)

        pendingRemoteCommitment, err := channel.RemoteCommitChainTip()
        if err != nil && err != channeldb.ErrNoPendingCommit {
                return nil, err
        }
        if pendingRemoteCommitment != nil {
                htlcSets[RemotePendingHtlcSet] = newHtlcSet(
                        pendingRemoteCommitment.Commitment.Htlcs,
                )
        }

        return NewChannelArbitrator(
                arbCfg, htlcSets, chanLog,
        ), nil
}

// getArbChannel returns an open channel wrapper for use by channel arbitrators.
func (c *ChainArbitrator) getArbChannel(
        channel *channeldb.OpenChannel) *arbChannel {

        return &arbChannel{
                channel: channel,
                c:       c,
        }
}

// ResolveContract marks a contract as fully resolved within the database.
// This is only to be done once all contracts which were live on the channel
// before hitting the chain have been resolved.
func (c *ChainArbitrator) ResolveContract(chanPoint wire.OutPoint) error {
        log.Infof("Marking ChannelPoint(%v) fully resolved", chanPoint)

        // First, we'll we'll mark the channel as fully closed from the PoV of
        // the channel source.
        err := c.chanSource.ChannelStateDB().MarkChanFullyClosed(&chanPoint)
        if err != nil {
                log.Errorf("ChainArbitrator: unable to mark ChannelPoint(%v) "+
                        "fully closed: %v", chanPoint, err)
                return err
        }

        // Now that the channel has been marked as fully closed, we'll stop
        // both the channel arbitrator and chain watcher for this channel if
        // they're still active.
        var arbLog ArbitratorLog
        c.Lock()
        chainArb := c.activeChannels[chanPoint]
        delete(c.activeChannels, chanPoint)

        chainWatcher := c.activeWatchers[chanPoint]
        delete(c.activeWatchers, chanPoint)
        c.Unlock()

        if chainArb != nil {
                arbLog = chainArb.log

                if err := chainArb.Stop(); err != nil {
                        log.Warnf("unable to stop ChannelArbitrator(%v): %v",
                                chanPoint, err)
                }
        }
        if chainWatcher != nil {
                if err := chainWatcher.Stop(); err != nil {
                        log.Warnf("unable to stop ChainWatcher(%v): %v",
                                chanPoint, err)
                }
        }

        // Once this has been marked as resolved, we'll wipe the log that the
        // channel arbitrator was using to store its persistent state. We do
        // this after marking the channel resolved, as otherwise, the
        // arbitrator would be re-created, and think it was starting from the
        // default state.
        if arbLog != nil {
                if err := arbLog.WipeHistory(); err != nil {
                        return err
                }
        }

        return nil
}

// Start launches all goroutines that the ChainArbitrator needs to operate.
func (c *ChainArbitrator) Start(beat chainio.Blockbeat) error {
        if !atomic.CompareAndSwapInt32(&c.started, 0, 1) {
                return nil
        }

        // Set the current beat.
        c.beat = beat

        // First, we'll fetch all the channels that are still open, in order to
        // collect them within our set of active contracts.
        if err := c.loadOpenChannels(); err != nil {
                return err
        }

        // In addition to the channels that we know to be open, we'll also
        // launch arbitrators to finishing resolving any channels that are in
        // the pending close state.
        if err := c.loadPendingCloseChannels(); err != nil {
                return err
        }

        // Now, we'll start all chain watchers in parallel to shorten start up
        // duration. In neutrino mode, this allows spend registrations to take
        // advantage of batch spend reporting, instead of doing a single rescan
        // per chain watcher.
        //
        // NOTE: After this point, we Stop the chain arb to ensure that any
        // lingering goroutines are cleaned up before exiting.
        watcherErrs := make(chan error, len(c.activeWatchers))
        var wg sync.WaitGroup
        for _, watcher := range c.activeWatchers {
                wg.Add(1)
                go func(w *chainWatcher) {
                        defer wg.Done()
                        select {
                        case watcherErrs <- w.Start():
                        case <-c.quit:
                                watcherErrs <- ErrChainArbExiting
                        }
                }(watcher)
        }

        // Once all chain watchers have been started, seal the err chan to
        // signal the end of the err stream.
        go func() {
                wg.Wait()
                close(watcherErrs)
        }()

        // stopAndLog is a helper function which shuts down the chain arb and
        // logs errors if they occur.
        stopAndLog := func() {
                if err := c.Stop(); err != nil {
                        log.Errorf("ChainArbitrator could not shutdown: %v", err)
                }
        }

        // Handle all errors returned from spawning our chain watchers. If any
        // of them failed, we will stop the chain arb to shutdown any active
        // goroutines.
        for err := range watcherErrs {
                if err != nil {
                        stopAndLog()
                        return err
                }
        }

        // Before we start all of our arbitrators, we do a preliminary state
        // lookup so that we can combine all of these lookups in a single db
        // transaction.
        var startStates map[wire.OutPoint]*chanArbStartState

        err := kvdb.View(c.chanSource, func(tx walletdb.ReadTx) error {
                for _, arbitrator := range c.activeChannels {
                        startState, err := arbitrator.getStartState(tx)
                        if err != nil {
                                return err
                        }

                        startStates[arbitrator.cfg.ChanPoint] = startState
                }

                return nil
        }, func() {
                startStates = make(
                        map[wire.OutPoint]*chanArbStartState,
                        len(c.activeChannels),
                )
        })
        if err != nil {
                stopAndLog()
                return err
        }

        // Launch all the goroutines for each arbitrator so they can carry out
        // their duties.
        for _, arbitrator := range c.activeChannels {
                startState, ok := startStates[arbitrator.cfg.ChanPoint]
                if !ok {
                        stopAndLog()
                        return fmt.Errorf("arbitrator: %v has no start state",
                                arbitrator.cfg.ChanPoint)
                }

                if err := arbitrator.Start(startState, c.beat); err != nil {
                        stopAndLog()
                        return err
                }
        }

        // Start our goroutine which will dispatch blocks to each arbitrator.
        c.wg.Add(1)
        go func() {
                defer c.wg.Done()
                c.dispatchBlocks()
        }()

        log.Infof("ChainArbitrator starting at height %d with %d chain "+
                "watchers, %d channel arbitrators, and budget config=[%v]",
                c.beat.Height(), len(c.activeWatchers), len(c.activeChannels),
                &c.cfg.Budget)

        // TODO(roasbeef): eventually move all breach watching here

        return nil
}

// dispatchBlocks consumes a block epoch notification stream and dispatches
// blocks to each of the chain arb's active channel arbitrators. This function
// must be run in a goroutine.
func (c *ChainArbitrator) dispatchBlocks() {
        // Consume block epochs until we receive the instruction to shutdown.
        for {
                select {
                // Consume block epochs, exiting if our subscription is
                // terminated.
                case beat := <-c.BlockbeatChan:
                        // Set the current blockbeat.
                        c.beat = beat

                        // Send this blockbeat to all the active channels and
                        // wait for them to finish processing it.
                        c.handleBlockbeat(beat)

                // Exit if the chain arbitrator is shutting down.
                case <-c.quit:
                        return
                }
        }
}

// handleBlockbeat sends the blockbeat to all active channel arbitrator in
// parallel and wait for them to finish processing it.
func (c *ChainArbitrator) handleBlockbeat(beat chainio.Blockbeat) {
        // Read the active channels in a lock.
        c.Lock()

        // Create a slice to record active channel arbitrator.
        channels := make([]chainio.Consumer, 0, len(c.activeChannels))
        watchers := make([]chainio.Consumer, 0, len(c.activeWatchers))

        // Copy the active channels to the slice.
        for _, channel := range c.activeChannels {
                channels = append(channels, channel)
        }

        for _, watcher := range c.activeWatchers {
                watchers = append(watchers, watcher)
        }

        c.Unlock()

        // Iterate all the copied watchers and send the blockbeat to them.
        err := chainio.DispatchConcurrent(beat, watchers)
        if err != nil {
                log.Errorf("Notify blockbeat for chainWatcher failed: %v", err)
        }

        // Iterate all the copied channels and send the blockbeat to them.
        //
        // NOTE: This method will timeout if the processing of blocks of the
        // subsystems is too long (60s).
        err = chainio.DispatchConcurrent(beat, channels)
        if err != nil {
                log.Errorf("Notify blockbeat for ChannelArbitrator failed: %v",
                        err)
        }

        // Notify the chain arbitrator has processed the block.
        c.NotifyBlockProcessed(beat, err)
}

// republishClosingTxs will load any stored cooperative or unilateral closing
// transactions and republish them. This helps ensure propagation of the
// transactions in the event that prior publications failed.
func (c *ChainArbitrator) republishClosingTxs(
        channel *channeldb.OpenChannel) error {

        // If the channel has had its unilateral close broadcasted already,
        // republish it in case it didn't propagate.
        if channel.HasChanStatus(channeldb.ChanStatusCommitBroadcasted) {
                err := c.rebroadcast(
                        channel, channeldb.ChanStatusCommitBroadcasted,
                )
                if err != nil {
                        return err
                }
        }

        // If the channel has had its cooperative close broadcasted
        // already, republish it in case it didn't propagate.
        if channel.HasChanStatus(channeldb.ChanStatusCoopBroadcasted) {
                err := c.rebroadcast(
                        channel, channeldb.ChanStatusCoopBroadcasted,
                )
                if err != nil {
                        return err
                }
        }

        return nil
}

// rebroadcast is a helper method which will republish the unilateral or
// cooperative close transaction or a channel in a particular state.
//
// NOTE: There is no risk to calling this method if the channel isn't in either
// CommitmentBroadcasted or CoopBroadcasted, but the logs will be misleading.
func (c *ChainArbitrator) rebroadcast(channel *channeldb.OpenChannel,
        state channeldb.ChannelStatus) error {

        chanPoint := channel.FundingOutpoint

        var (
                closeTx *wire.MsgTx
                kind    string
                err     error
        )
        switch state {
        case channeldb.ChanStatusCommitBroadcasted:
                kind = "force"
                closeTx, err = channel.BroadcastedCommitment()

        case channeldb.ChanStatusCoopBroadcasted:
                kind = "coop"
                closeTx, err = channel.BroadcastedCooperative()

        default:
                return fmt.Errorf("unknown closing state: %v", state)
        }

        switch {
        // This can happen for channels that had their closing tx published
        // before we started storing it to disk.
        case err == channeldb.ErrNoCloseTx:
                log.Warnf("Channel %v is in state %v, but no %s closing tx "+
                        "to re-publish...", chanPoint, state, kind)
                return nil

        case err != nil:
                return err
        }

        log.Infof("Re-publishing %s close tx(%v) for channel %v",
                kind, closeTx.TxHash(), chanPoint)

        label := labels.MakeLabel(
                labels.LabelTypeChannelClose, &channel.ShortChannelID,
        )
        err = c.cfg.PublishTx(closeTx, label)
        if err != nil && err != lnwallet.ErrDoubleSpend {
                log.Warnf("Unable to broadcast %s close tx(%v): %v",
                        kind, closeTx.TxHash(), err)
        }

        return nil
}

// Stop signals the ChainArbitrator to trigger a graceful shutdown. Any active
// channel arbitrators will be signalled to exit, and this method will block
// until they've all exited.
func (c *ChainArbitrator) Stop() error {
        if !atomic.CompareAndSwapInt32(&c.stopped, 0, 1) {
                return nil
        }

        log.Info("ChainArbitrator shutting down...")
        defer log.Debug("ChainArbitrator shutdown complete")

        close(c.quit)

        var (
                activeWatchers = make(map[wire.OutPoint]*chainWatcher)
                activeChannels = make(map[wire.OutPoint]*ChannelArbitrator)
        )

        // Copy the current set of active watchers and arbitrators to shutdown.
        // We don't want to hold the lock when shutting down each watcher or
        // arbitrator individually, as they may need to acquire this mutex.
        c.Lock()
        for chanPoint, watcher := range c.activeWatchers {
                activeWatchers[chanPoint] = watcher
        }
        for chanPoint, arbitrator := range c.activeChannels {
                activeChannels[chanPoint] = arbitrator
        }
        c.Unlock()

        for chanPoint, watcher := range activeWatchers {
                log.Tracef("Attempting to stop ChainWatcher(%v)",
                        chanPoint)

                if err := watcher.Stop(); err != nil {
                        log.Errorf("unable to stop watcher for "+
                                "ChannelPoint(%v): %v", chanPoint, err)
                }
        }
        for chanPoint, arbitrator := range activeChannels {
                log.Tracef("Attempting to stop ChannelArbitrator(%v)",
                        chanPoint)

                if err := arbitrator.Stop(); err != nil {
                        log.Errorf("unable to stop arbitrator for "+
                                "ChannelPoint(%v): %v", chanPoint, err)
                }
        }

        c.wg.Wait()

        return nil
}

// ContractUpdate is a message packages the latest set of active HTLCs on a
// commitment, and also identifies which commitment received a new set of
// HTLCs.
type ContractUpdate struct {
        // HtlcKey identifies which commitment the HTLCs below are present on.
        HtlcKey HtlcSetKey

        // Htlcs are the of active HTLCs on the commitment identified by the
        // above HtlcKey.
        Htlcs []channeldb.HTLC
}

// ContractSignals is used by outside subsystems to notify a channel arbitrator
// of its ShortChannelID.
type ContractSignals struct {
        // ShortChanID is the up to date short channel ID for a contract. This
        // can change either if when the contract was added it didn't yet have
        // a stable identifier, or in the case of a reorg.
        ShortChanID lnwire.ShortChannelID
}

// UpdateContractSignals sends a set of active, up to date contract signals to
// the ChannelArbitrator which is has been assigned to the channel infield by
// the passed channel point.
func (c *ChainArbitrator) UpdateContractSignals(chanPoint wire.OutPoint,
        signals *ContractSignals) error {

        log.Infof("Attempting to update ContractSignals for ChannelPoint(%v)",
                chanPoint)

        c.Lock()
        arbitrator, ok := c.activeChannels[chanPoint]
        c.Unlock()
        if !ok {
                return fmt.Errorf("unable to find arbitrator")
        }

        arbitrator.UpdateContractSignals(signals)

        return nil
}

// NotifyContractUpdate lets a channel arbitrator know that a new
// ContractUpdate is available. This calls the ChannelArbitrator's internal
// method NotifyContractUpdate which waits for a response on a done chan before
// returning. This method will return an error if the ChannelArbitrator is not
// in the activeChannels map. However, this only happens if the arbitrator is
// resolved and the related link would already be shut down.
func (c *ChainArbitrator) NotifyContractUpdate(chanPoint wire.OutPoint,
        update *ContractUpdate) error {

        c.Lock()
        arbitrator, ok := c.activeChannels[chanPoint]
        c.Unlock()
        if !ok {
                return fmt.Errorf("can't find arbitrator for %v", chanPoint)
        }

        arbitrator.notifyContractUpdate(update)
        return nil
}

// GetChannelArbitrator safely returns the channel arbitrator for a given
// channel outpoint.
func (c *ChainArbitrator) GetChannelArbitrator(chanPoint wire.OutPoint) (
        *ChannelArbitrator, error) {

        c.Lock()
        arbitrator, ok := c.activeChannels[chanPoint]
        c.Unlock()
        if !ok {
                return nil, fmt.Errorf("unable to find arbitrator")
        }

        return arbitrator, nil
}

// forceCloseReq is a request sent from an outside sub-system to the arbitrator
// that watches a particular channel to broadcast the commitment transaction,
// and enter the resolution phase of the channel.
type forceCloseReq struct {
        // errResp is a channel that will be sent upon either in the case of
        // force close success (nil error), or in the case on an error.
        //
        // NOTE; This channel MUST be buffered.
        errResp chan error

        // closeTx is a channel that carries the transaction which ultimately
        // closed out the channel.
        closeTx chan *wire.MsgTx
}

// ForceCloseContract attempts to force close the channel infield by the passed
// channel point. A force close will immediately terminate the contract,
// causing it to enter the resolution phase. If the force close was successful,
// then the force close transaction itself will be returned.
//
// TODO(roasbeef): just return the summary itself?
func (c *ChainArbitrator) ForceCloseContract(chanPoint wire.OutPoint) (*wire.MsgTx, error) {
        c.Lock()
        arbitrator, ok := c.activeChannels[chanPoint]
        c.Unlock()
        if !ok {
                return nil, fmt.Errorf("unable to find arbitrator")
        }

        log.Infof("Attempting to force close ChannelPoint(%v)", chanPoint)

        // Before closing, we'll attempt to send a disable update for the
        // channel. We do so before closing the channel as otherwise the current
        // edge policy won't be retrievable from the graph.
        if err := c.cfg.DisableChannel(chanPoint); err != nil {
                log.Warnf("Unable to disable channel %v on "+
                        "close: %v", chanPoint, err)
        }

        errChan := make(chan error, 1)
        respChan := make(chan *wire.MsgTx, 1)

        // With the channel found, and the request crafted, we'll send over a
        // force close request to the arbitrator that watches this channel.
        select {
        case arbitrator.forceCloseReqs <- &forceCloseReq{
                errResp: errChan,
                closeTx: respChan,
        }:
        case <-c.quit:
                return nil, ErrChainArbExiting
        }

        // We'll await two responses: the error response, and the transaction
        // that closed out the channel.
        select {
        case err := <-errChan:
                if err != nil {
                        return nil, err
                }
        case <-c.quit:
                return nil, ErrChainArbExiting
        }

        var closeTx *wire.MsgTx
        select {
        case closeTx = <-respChan:
        case <-c.quit:
                return nil, ErrChainArbExiting
        }

        return closeTx, nil
}

// WatchNewChannel sends the ChainArbitrator a message to create a
// ChannelArbitrator tasked with watching over a new channel. Once a new
// channel has finished its final funding flow, it should be registered with
// the ChainArbitrator so we can properly react to any on-chain events.
func (c *ChainArbitrator) WatchNewChannel(newChan *channeldb.OpenChannel) error {
        c.Lock()
        defer c.Unlock()

        chanPoint := newChan.FundingOutpoint

        log.Infof("Creating new chainWatcher and ChannelArbitrator for "+
                "ChannelPoint(%v)", chanPoint)

        // If we're already watching this channel, then we'll ignore this
        // request.
        if _, ok := c.activeChannels[chanPoint]; ok {
                return nil
        }

        // First, also create an active chainWatcher for this channel to ensure
        // that we detect any relevant on chain events.
        chainWatcher, err := newChainWatcher(
                chainWatcherConfig{
                        chanState: newChan,
                        notifier:  c.cfg.Notifier,
                        signer:    c.cfg.Signer,
                        isOurAddr: c.cfg.IsOurAddress,
                        contractBreach: func(
                                retInfo *lnwallet.BreachRetribution) error {

                                return c.cfg.ContractBreach(
                                        chanPoint, retInfo,
                                )
                        },
                        extractStateNumHint: lnwallet.GetStateNumHint,
                        auxLeafStore:        c.cfg.AuxLeafStore,
                        auxResolver:         c.cfg.AuxResolver,
                },
        )
        if err != nil {
                return err
        }

        c.activeWatchers[chanPoint] = chainWatcher

        // We'll also create a new channel arbitrator instance using this new
        // channel, and our internal state.
        channelArb, err := newActiveChannelArbitrator(
                newChan, c, chainWatcher.SubscribeChannelEvents(),
        )
        if err != nil {
                return err
        }

        // With the arbitrator created, we'll add it to our set of active
        // arbitrators, then launch it.
        c.activeChannels[chanPoint] = channelArb

        if err := channelArb.Start(nil, c.beat); err != nil {
                return err
        }

        return chainWatcher.Start()
}

// SubscribeChannelEvents returns a new active subscription for the set of
// possible on-chain events for a particular channel. The struct can be used by
// callers to be notified whenever an event that changes the state of the
// channel on-chain occurs.
func (c *ChainArbitrator) SubscribeChannelEvents(
        chanPoint wire.OutPoint) (*ChainEventSubscription, error) {

        // First, we'll attempt to look up the active watcher for this channel.
        // If we can't find it, then we'll return an error back to the caller.
        c.Lock()
        watcher, ok := c.activeWatchers[chanPoint]
        c.Unlock()

        if !ok {
                return nil, fmt.Errorf("unable to find watcher for: %v",
                        chanPoint)
        }

        // With the watcher located, we'll request for it to create a new chain
        // event subscription client.
        return watcher.SubscribeChannelEvents(), nil
}

// FindOutgoingHTLCDeadline returns the deadline in absolute block height for
// the specified outgoing HTLC. For an outgoing HTLC, its deadline is defined
// by the timeout height of its corresponding incoming HTLC - this is the
// expiry height the that remote peer can spend his/her outgoing HTLC via the
// timeout path.
func (c *ChainArbitrator) FindOutgoingHTLCDeadline(scid lnwire.ShortChannelID,
        outgoingHTLC channeldb.HTLC) fn.Option[int32] {

        // Find the outgoing HTLC's corresponding incoming HTLC in the circuit
        // map.
        rHash := outgoingHTLC.RHash
        circuit := models.CircuitKey{
                ChanID: scid,
                HtlcID: outgoingHTLC.HtlcIndex,
        }
        incomingCircuit := c.cfg.QueryIncomingCircuit(circuit)

        // If there's no incoming circuit found, we will use the default
        // deadline.
        if incomingCircuit == nil {
                log.Warnf("ChannelArbitrator(%v): incoming circuit key not "+
                        "found for rHash=%x, using default deadline instead",
                        scid, rHash)

                return fn.None[int32]()
        }

        // If this is a locally initiated HTLC, it means we are the first hop.
        // In this case, we can relax the deadline.
        if incomingCircuit.ChanID.IsDefault() {
                log.Infof("ChannelArbitrator(%v): using default deadline for "+
                        "locally initiated HTLC for rHash=%x", scid, rHash)

                return fn.None[int32]()
        }

        log.Debugf("Found incoming circuit %v for rHash=%x using outgoing "+
                "circuit %v", incomingCircuit, rHash, circuit)

        c.Lock()
        defer c.Unlock()

        // Iterate over all active channels to find the incoming HTLC specified
        // by its circuit key.
        for cp, channelArb := range c.activeChannels {
                // Skip if the SCID doesn't match.
                if channelArb.cfg.ShortChanID != incomingCircuit.ChanID {
                        continue
                }

                // Make sure the channel arbitrator has the latest view of its
                // active HTLCs.
                channelArb.updateActiveHTLCs()

                // Iterate all the known HTLCs to find the targeted incoming
                // HTLC.
                for _, htlcs := range channelArb.activeHTLCs {
                        for _, htlc := range htlcs.incomingHTLCs {
                                // Skip if the index doesn't match.
                                if htlc.HtlcIndex != incomingCircuit.HtlcID {
                                        continue
                                }

                                log.Debugf("ChannelArbitrator(%v): found "+
                                        "incoming HTLC in channel=%v using "+
                                        "rHash=%x, refundTimeout=%v", scid,
                                        cp, rHash, htlc.RefundTimeout)

                                return fn.Some(int32(htlc.RefundTimeout))
                        }
                }
        }

        // If there's no incoming HTLC found, yet we have the incoming circuit,
        // something is wrong - in this case, we return the none deadline.
        log.Errorf("ChannelArbitrator(%v): incoming HTLC not found for "+
                "rHash=%x, using default deadline instead", scid, rHash)

        return fn.None[int32]()
}

// TODO(roasbeef): arbitration reports
//  * types: contested, waiting for success conf, etc

// NOTE: part of the `chainio.Consumer` interface.
func (c *ChainArbitrator) Name() string {
        return "ChainArbitrator"
}

// loadOpenChannels loads all channels that are currently open in the database
// and registers them with the chainWatcher for future notification.
func (c *ChainArbitrator) loadOpenChannels() error {
        openChannels, err := c.chanSource.ChannelStateDB().FetchAllChannels()
        if err != nil {
                return err
        }

        if len(openChannels) == 0 {
                return nil
        }

        log.Infof("Creating ChannelArbitrators for %v active channels",
                len(openChannels))

        // For each open channel, we'll configure then launch a corresponding
        // ChannelArbitrator.
        for _, channel := range openChannels {
                chanPoint := channel.FundingOutpoint
                channel := channel

                // First, we'll create an active chainWatcher for this channel
                // to ensure that we detect any relevant on chain events.
                breachClosure := func(ret *lnwallet.BreachRetribution) error {
                        return c.cfg.ContractBreach(chanPoint, ret)
                }

                chainWatcher, err := newChainWatcher(
                        chainWatcherConfig{
                                chanState:           channel,
                                notifier:            c.cfg.Notifier,
                                signer:              c.cfg.Signer,
                                isOurAddr:           c.cfg.IsOurAddress,
                                contractBreach:      breachClosure,
                                extractStateNumHint: lnwallet.GetStateNumHint,
                                auxLeafStore:        c.cfg.AuxLeafStore,
                                auxResolver:         c.cfg.AuxResolver,
                        },
                )
                if err != nil {
                        return err
                }

                c.activeWatchers[chanPoint] = chainWatcher
                channelArb, err := newActiveChannelArbitrator(
                        channel, c, chainWatcher.SubscribeChannelEvents(),
                )
                if err != nil {
                        return err
                }

                c.activeChannels[chanPoint] = channelArb

                // Republish any closing transactions for this channel.
                err = c.republishClosingTxs(channel)
                if err != nil {
                        log.Errorf("Failed to republish closing txs for "+
                                "channel %v", chanPoint)
                }
        }

        return nil
}

// loadPendingCloseChannels loads all channels that are currently pending
// closure in the database and registers them with the ChannelArbitrator to
// continue the resolution process.
func (c *ChainArbitrator) loadPendingCloseChannels() error {
        chanStateDB := c.chanSource.ChannelStateDB()

        closingChannels, err := chanStateDB.FetchClosedChannels(true)
        if err != nil {
                return err
        }

        if len(closingChannels) == 0 {
                return nil
        }

        log.Infof("Creating ChannelArbitrators for %v closing channels",
                len(closingChannels))

        // Next, for each channel is the closing state, we'll launch a
        // corresponding more restricted resolver, as we don't have to watch
        // the chain any longer, only resolve the contracts on the confirmed
        // commitment.
        //nolint:ll
        for _, closeChanInfo := range closingChannels {
                // We can leave off the CloseContract and ForceCloseChan
                // methods as the channel is already closed at this point.
                chanPoint := closeChanInfo.ChanPoint
                arbCfg := ChannelArbitratorConfig{
                        ChanPoint:             chanPoint,
                        ShortChanID:           closeChanInfo.ShortChanID,
                        ChainArbitratorConfig: c.cfg,
                        ChainEvents:           &ChainEventSubscription{},
                        IsPendingClose:        true,
                        ClosingHeight:         closeChanInfo.CloseHeight,
                        CloseType:             closeChanInfo.CloseType,
                        PutResolverReport: func(tx kvdb.RwTx,
                                report *channeldb.ResolverReport) error {

                                return c.chanSource.PutResolverReport(
                                        tx, c.cfg.ChainHash, &chanPoint, report,
                                )
                        },
                        FetchHistoricalChannel: func() (*channeldb.OpenChannel, error) {
                                return chanStateDB.FetchHistoricalChannel(&chanPoint)
                        },
                        FindOutgoingHTLCDeadline: func(
                                htlc channeldb.HTLC) fn.Option[int32] {

                                return c.FindOutgoingHTLCDeadline(
                                        closeChanInfo.ShortChanID, htlc,
                                )
                        },
                }
                chanLog, err := newBoltArbitratorLog(
                        c.chanSource.Backend, arbCfg, c.cfg.ChainHash, chanPoint,
                )
                if err != nil {
                        return err
                }
                arbCfg.MarkChannelResolved = func() error {
                        if c.cfg.NotifyFullyResolvedChannel != nil {
                                c.cfg.NotifyFullyResolvedChannel(chanPoint)
                        }

                        return c.ResolveContract(chanPoint)
                }

                // We create an empty map of HTLC's here since it's possible
                // that the channel is in StateDefault and updateActiveHTLCs is
                // called. We want to avoid writing to an empty map. Since the
                // channel is already in the process of being resolved, no new
                // HTLCs will be added.
                c.activeChannels[chanPoint] = NewChannelArbitrator(
                        arbCfg, make(map[HtlcSetKey]htlcSet), chanLog,
                )
        }

        return nil
}

// RedispatchBlockbeat resends the current blockbeat to the channels specified
// by the chanPoints. It is used when a channel is added to the chain
// arbitrator after it has been started, e.g., during the channel restore
// process.
func (c *ChainArbitrator) RedispatchBlockbeat(chanPoints []wire.OutPoint) {
        // Get the current blockbeat.
        beat := c.beat

        // Prepare two sets of consumers.
        channels := make([]chainio.Consumer, 0, len(chanPoints))
        watchers := make([]chainio.Consumer, 0, len(chanPoints))

        // Read the active channels in a lock.
        c.Lock()
        for _, op := range chanPoints {
                if channel, ok := c.activeChannels[op]; ok {
                        channels = append(channels, channel)
                }

                if watcher, ok := c.activeWatchers[op]; ok {
                        watchers = append(watchers, watcher)
                }
        }
        c.Unlock()

        // Iterate all the copied watchers and send the blockbeat to them.
        err := chainio.DispatchConcurrent(beat, watchers)
        if err != nil {
                log.Errorf("Notify blockbeat for chainWatcher failed: %v", err)
        }

        // Iterate all the copied channels and send the blockbeat to them.
        err = chainio.DispatchConcurrent(beat, channels)
        if err != nil {
                // Shutdown lnd if there's an error processing the block.
                log.Errorf("Notify blockbeat for ChannelArbitrator failed: %v",
                        err)
        }
}

lightningnetwork / lnd / 12583319996

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous