13587276956

Committed 28 Feb 2025 11:32AM UTC coverage: 58.857% (+7.8%) from 51.098%

Build # 13587276956

Build Type

Pull #9567

github

Committed by

ellemouton

Commit Message

itest: make sure to not hit the natural ChannelUpdate rate limit

Channel Updates have a natural rate limit of 1 update per second due to
the fact that the timestamp carried in the update is only accurate to
the second. So we need to ensure that the next update we send in the
burst is at least 1 second after the last one.

Pull Request Pull Request #9567: itest: make sure to not hit the natural ChannelUpdate rate limit

Run Details

136630 of 232137 relevant lines covered (58.86%)

19201.78 hits per line

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

77.92

/htlcswitch/link.go

package htlcswitch

import (
        "bytes"
        "context"
        crand "crypto/rand"
        "crypto/sha256"
        "errors"
        "fmt"
        prand "math/rand"
        "sync"
        "sync/atomic"
        "time"

        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/wire"
        "github.com/btcsuite/btclog/v2"
        "github.com/lightningnetwork/lnd/channeldb"
        "github.com/lightningnetwork/lnd/contractcourt"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/htlcswitch/hodl"
        "github.com/lightningnetwork/lnd/htlcswitch/hop"
        "github.com/lightningnetwork/lnd/input"
        "github.com/lightningnetwork/lnd/invoices"
        "github.com/lightningnetwork/lnd/lnpeer"
        "github.com/lightningnetwork/lnd/lntypes"
        "github.com/lightningnetwork/lnd/lnutils"
        "github.com/lightningnetwork/lnd/lnwallet"
        "github.com/lightningnetwork/lnd/lnwallet/chainfee"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/queue"
        "github.com/lightningnetwork/lnd/record"
        "github.com/lightningnetwork/lnd/ticker"
        "github.com/lightningnetwork/lnd/tlv"
)

func init() {
        prand.Seed(time.Now().UnixNano())
}

const (
        // DefaultMaxOutgoingCltvExpiry is the maximum outgoing time lock that
        // the node accepts for forwarded payments. The value is relative to the
        // current block height. The reason to have a maximum is to prevent
        // funds getting locked up unreasonably long. Otherwise, an attacker
        // willing to lock its own funds too, could force the funds of this node
        // to be locked up for an indefinite (max int32) number of blocks.
        //
        // The value 2016 corresponds to on average two weeks worth of blocks
        // and is based on the maximum number of hops (20), the default CLTV
        // delta (40), and some extra margin to account for the other lightning
        // implementations and past lnd versions which used to have a default
        // CLTV delta of 144.
        DefaultMaxOutgoingCltvExpiry = 2016

        // DefaultMinLinkFeeUpdateTimeout represents the minimum interval in
        // which a link should propose to update its commitment fee rate.
        DefaultMinLinkFeeUpdateTimeout = 10 * time.Minute

        // DefaultMaxLinkFeeUpdateTimeout represents the maximum interval in
        // which a link should propose to update its commitment fee rate.
        DefaultMaxLinkFeeUpdateTimeout = 60 * time.Minute

        // DefaultMaxLinkFeeAllocation is the highest allocation we'll allow
        // a channel's commitment fee to be of its balance. This only applies to
        // the initiator of the channel.
        DefaultMaxLinkFeeAllocation float64 = 0.5
)

// ExpectedFee computes the expected fee for a given htlc amount. The value
// returned from this function is to be used as a sanity check when forwarding
// HTLC's to ensure that an incoming HTLC properly adheres to our propagated
// forwarding policy.
//
// TODO(roasbeef): also add in current available channel bandwidth, inverse
// func
func ExpectedFee(f models.ForwardingPolicy,
        htlcAmt lnwire.MilliSatoshi) lnwire.MilliSatoshi {

        return f.BaseFee + (htlcAmt*f.FeeRate)/1000000
}

// ChannelLinkConfig defines the configuration for the channel link. ALL
// elements within the configuration MUST be non-nil for channel link to carry
// out its duties.
type ChannelLinkConfig struct {
        // FwrdingPolicy is the initial forwarding policy to be used when
        // deciding whether to forwarding incoming HTLC's or not. This value
        // can be updated with subsequent calls to UpdateForwardingPolicy
        // targeted at a given ChannelLink concrete interface implementation.
        FwrdingPolicy models.ForwardingPolicy

        // Circuits provides restricted access to the switch's circuit map,
        // allowing the link to open and close circuits.
        Circuits CircuitModifier

        // BestHeight returns the best known height.
        BestHeight func() uint32

        // ForwardPackets attempts to forward the batch of htlcs through the
        // switch. The function returns and error in case it fails to send one or
        // more packets. The link's quit signal should be provided to allow
        // cancellation of forwarding during link shutdown.
        ForwardPackets func(<-chan struct{}, bool, ...*htlcPacket) error

        // DecodeHopIterators facilitates batched decoding of HTLC Sphinx onion
        // blobs, which are then used to inform how to forward an HTLC.
        //
        // NOTE: This function assumes the same set of readers and preimages
        // are always presented for the same identifier.
        DecodeHopIterators func([]byte, []hop.DecodeHopIteratorRequest) (
                []hop.DecodeHopIteratorResponse, error)

        // ExtractErrorEncrypter function is responsible for decoding HTLC
        // Sphinx onion blob, and creating onion failure obfuscator.
        ExtractErrorEncrypter hop.ErrorEncrypterExtracter

        // FetchLastChannelUpdate retrieves the latest routing policy for a
        // target channel. This channel will typically be the outgoing channel
        // specified when we receive an incoming HTLC.  This will be used to
        // provide payment senders our latest policy when sending encrypted
        // error messages.
        FetchLastChannelUpdate func(lnwire.ShortChannelID) (
                *lnwire.ChannelUpdate1, error)

        // Peer is a lightning network node with which we have the channel link
        // opened.
        Peer lnpeer.Peer

        // Registry is a sub-system which responsible for managing the invoices
        // in thread-safe manner.
        Registry InvoiceDatabase

        // PreimageCache is a global witness beacon that houses any new
        // preimages discovered by other links. We'll use this to add new
        // witnesses that we discover which will notify any sub-systems
        // subscribed to new events.
        PreimageCache contractcourt.WitnessBeacon

        // OnChannelFailure is a function closure that we'll call if the
        // channel failed for some reason. Depending on the severity of the
        // error, the closure potentially must force close this channel and
        // disconnect the peer.
        //
        // NOTE: The method must return in order for the ChannelLink to be able
        // to shut down properly.
        OnChannelFailure func(lnwire.ChannelID, lnwire.ShortChannelID,
                LinkFailureError)

        // UpdateContractSignals is a function closure that we'll use to update
        // outside sub-systems with this channel's latest ShortChannelID.
        UpdateContractSignals func(*contractcourt.ContractSignals) error

        // NotifyContractUpdate is a function closure that we'll use to update
        // the contractcourt and more specifically the ChannelArbitrator of the
        // latest channel state.
        NotifyContractUpdate func(*contractcourt.ContractUpdate) error

        // ChainEvents is an active subscription to the chain watcher for this
        // channel to be notified of any on-chain activity related to this
        // channel.
        ChainEvents *contractcourt.ChainEventSubscription

        // FeeEstimator is an instance of a live fee estimator which will be
        // used to dynamically regulate the current fee of the commitment
        // transaction to ensure timely confirmation.
        FeeEstimator chainfee.Estimator

        // hodl.Mask is a bitvector composed of hodl.Flags, specifying breakpoints
        // for HTLC forwarding internal to the switch.
        //
        // NOTE: This should only be used for testing.
        HodlMask hodl.Mask

        // SyncStates is used to indicate that we need send the channel
        // reestablishment message to the remote peer. It should be done if our
        // clients have been restarted, or remote peer have been reconnected.
        SyncStates bool

        // BatchTicker is the ticker that determines the interval that we'll
        // use to check the batch to see if there're any updates we should
        // flush out. By batching updates into a single commit, we attempt to
        // increase throughput by maximizing the number of updates coalesced
        // into a single commit.
        BatchTicker ticker.Ticker

        // FwdPkgGCTicker is the ticker determining the frequency at which
        // garbage collection of forwarding packages occurs. We use a
        // time-based approach, as opposed to block epochs, as to not hinder
        // syncing.
        FwdPkgGCTicker ticker.Ticker

        // PendingCommitTicker is a ticker that allows the link to determine if
        // a locally initiated commitment dance gets stuck waiting for the
        // remote party to revoke.
        PendingCommitTicker ticker.Ticker

        // BatchSize is the max size of a batch of updates done to the link
        // before we do a state update.
        BatchSize uint32

        // UnsafeReplay will cause a link to replay the adds in its latest
        // commitment txn after the link is restarted. This should only be used
        // in testing, it is here to ensure the sphinx replay detection on the
        // receiving node is persistent.
        UnsafeReplay bool

        // MinUpdateTimeout represents the minimum interval in which a link
        // will propose to update its commitment fee rate. A random timeout will
        // be selected between this and MaxUpdateTimeout.
        MinUpdateTimeout time.Duration

        // MaxUpdateTimeout represents the maximum interval in which a link
        // will propose to update its commitment fee rate. A random timeout will
        // be selected between this and MinUpdateTimeout.
        MaxUpdateTimeout time.Duration

        // OutgoingCltvRejectDelta defines the number of blocks before expiry of
        // an htlc where we don't offer an htlc anymore. This should be at least
        // the outgoing broadcast delta, because in any case we don't want to
        // risk offering an htlc that triggers channel closure.
        OutgoingCltvRejectDelta uint32

        // TowerClient is an optional engine that manages the signing,
        // encrypting, and uploading of justice transactions to the daemon's
        // configured set of watchtowers for legacy channels.
        TowerClient TowerClient

        // MaxOutgoingCltvExpiry is the maximum outgoing timelock that the link
        // should accept for a forwarded HTLC. The value is relative to the
        // current block height.
        MaxOutgoingCltvExpiry uint32

        // MaxFeeAllocation is the highest allocation we'll allow a channel's
        // commitment fee to be of its balance. This only applies to the
        // initiator of the channel.
        MaxFeeAllocation float64

        // MaxAnchorsCommitFeeRate is the max commitment fee rate we'll use as
        // the initiator for channels of the anchor type.
        MaxAnchorsCommitFeeRate chainfee.SatPerKWeight

        // NotifyActiveLink allows the link to tell the ChannelNotifier when a
        // link is first started.
        NotifyActiveLink func(wire.OutPoint)

        // NotifyActiveChannel allows the link to tell the ChannelNotifier when
        // channels becomes active.
        NotifyActiveChannel func(wire.OutPoint)

        // NotifyInactiveChannel allows the switch to tell the ChannelNotifier
        // when channels become inactive.
        NotifyInactiveChannel func(wire.OutPoint)

        // NotifyInactiveLinkEvent allows the switch to tell the
        // ChannelNotifier when a channel link become inactive.
        NotifyInactiveLinkEvent func(wire.OutPoint)

        // HtlcNotifier is an instance of a htlcNotifier which we will pipe htlc
        // events through.
        HtlcNotifier htlcNotifier

        // FailAliasUpdate is a function used to fail an HTLC for an
        // option_scid_alias channel.
        FailAliasUpdate func(sid lnwire.ShortChannelID,
                incoming bool) *lnwire.ChannelUpdate1

        // GetAliases is used by the link and switch to fetch the set of
        // aliases for a given link.
        GetAliases func(base lnwire.ShortChannelID) []lnwire.ShortChannelID

        // PreviouslySentShutdown is an optional value that is set if, at the
        // time of the link being started, persisted shutdown info was found for
        // the channel. This value being set means that we previously sent a
        // Shutdown message to our peer, and so we should do so again on
        // re-establish and should not allow anymore HTLC adds on the outgoing
        // direction of the link.
        PreviouslySentShutdown fn.Option[lnwire.Shutdown]

        // Adds the option to disable forwarding payments in blinded routes
        // by failing back any blinding-related payloads as if they were
        // invalid.
        DisallowRouteBlinding bool

        // DisallowQuiescence is a flag that can be used to disable the
        // quiescence protocol.
        DisallowQuiescence bool

        // MaxFeeExposure is the threshold in milli-satoshis after which we'll
        // restrict the flow of HTLCs and fee updates.
        MaxFeeExposure lnwire.MilliSatoshi

        // ShouldFwdExpEndorsement is a closure that indicates whether the link
        // should forward experimental endorsement signals.
        ShouldFwdExpEndorsement func() bool

        // AuxTrafficShaper is an optional auxiliary traffic shaper that can be
        // used to manage the bandwidth of the link.
        AuxTrafficShaper fn.Option[AuxTrafficShaper]
}

// channelLink is the service which drives a channel's commitment update
// state-machine. In the event that an HTLC needs to be propagated to another
// link, the forward handler from config is used which sends HTLC to the
// switch. Additionally, the link encapsulate logic of commitment protocol
// message ordering and updates.
type channelLink struct {
        // The following fields are only meant to be used *atomically*
        started       int32
        reestablished int32
        shutdown      int32

        // failed should be set to true in case a link error happens, making
        // sure we don't process any more updates.
        failed bool

        // keystoneBatch represents a volatile list of keystones that must be
        // written before attempting to sign the next commitment txn. These
        // represent all the HTLC's forwarded to the link from the switch. Once
        // we lock them into our outgoing commitment, then the circuit has a
        // keystone, and is fully opened.
        keystoneBatch []Keystone

        // openedCircuits is the set of all payment circuits that will be open
        // once we make our next commitment. After making the commitment we'll
        // ACK all these from our mailbox to ensure that they don't get
        // re-delivered if we reconnect.
        openedCircuits []CircuitKey

        // closedCircuits is the set of all payment circuits that will be
        // closed once we make our next commitment. After taking the commitment
        // we'll ACK all these to ensure that they don't get re-delivered if we
        // reconnect.
        closedCircuits []CircuitKey

        // channel is a lightning network channel to which we apply htlc
        // updates.
        channel *lnwallet.LightningChannel

        // cfg is a structure which carries all dependable fields/handlers
        // which may affect behaviour of the service.
        cfg ChannelLinkConfig

        // mailBox is the main interface between the outside world and the
        // link. All incoming messages will be sent over this mailBox. Messages
        // include new updates from our connected peer, and new packets to be
        // forwarded sent by the switch.
        mailBox MailBox

        // upstream is a channel that new messages sent from the remote peer to
        // the local peer will be sent across.
        upstream chan lnwire.Message

        // downstream is a channel in which new multi-hop HTLC's to be
        // forwarded will be sent across. Messages from this channel are sent
        // by the HTLC switch.
        downstream chan *htlcPacket

        // updateFeeTimer is the timer responsible for updating the link's
        // commitment fee every time it fires.
        updateFeeTimer *time.Timer

        // uncommittedPreimages stores a list of all preimages that have been
        // learned since receiving the last CommitSig from the remote peer. The
        // batch will be flushed just before accepting the subsequent CommitSig
        // or on shutdown to avoid doing a write for each preimage received.
        uncommittedPreimages []lntypes.Preimage

        sync.RWMutex

        // hodlQueue is used to receive exit hop htlc resolutions from invoice
        // registry.
        hodlQueue *queue.ConcurrentQueue

        // hodlMap stores related htlc data for a circuit key. It allows
        // resolving those htlcs when we receive a message on hodlQueue.
        hodlMap map[models.CircuitKey]hodlHtlc

        // log is a link-specific logging instance.
        log btclog.Logger

        // isOutgoingAddBlocked tracks whether the channelLink can send an
        // UpdateAddHTLC.
        isOutgoingAddBlocked atomic.Bool

        // isIncomingAddBlocked tracks whether the channelLink can receive an
        // UpdateAddHTLC.
        isIncomingAddBlocked atomic.Bool

        // flushHooks is a hookMap that is triggered when we reach a channel
        // state with no live HTLCs.
        flushHooks hookMap

        // outgoingCommitHooks is a hookMap that is triggered after we send our
        // next CommitSig.
        outgoingCommitHooks hookMap

        // incomingCommitHooks is a hookMap that is triggered after we receive
        // our next CommitSig.
        incomingCommitHooks hookMap

        // quiescer is the state machine that tracks where this channel is with
        // respect to the quiescence protocol.
        quiescer Quiescer

        // quiescenceReqs is a queue of requests to quiesce this link. The
        // members of the queue are send-only channels we should call back with
        // the result.
        quiescenceReqs chan StfuReq

        // cg is a helper that encapsulates a wait group and quit channel and
        // allows contexts that either block or cancel on those depending on
        // the use case.
        cg *fn.ContextGuard
}

// hookMap is a data structure that is used to track the hooks that need to be
// called in various parts of the channelLink's lifecycle.
//
// WARNING: NOT thread-safe.
type hookMap struct {
        // allocIdx keeps track of the next id we haven't yet allocated.
        allocIdx atomic.Uint64

        // transient is a map of hooks that are only called the next time invoke
        // is called. These hooks are deleted during invoke.
        transient map[uint64]func()

        // newTransients is a channel that we use to accept new hooks into the
        // hookMap.
        newTransients chan func()
}

// newHookMap initializes a new empty hookMap.
func newHookMap() hookMap {
        return hookMap{
                allocIdx:      atomic.Uint64{},
                transient:     make(map[uint64]func()),
                newTransients: make(chan func()),
        }
}

// alloc allocates space in the hook map for the supplied hook, the second
// argument determines whether it goes into the transient or persistent part
// of the hookMap.
func (m *hookMap) alloc(hook func()) uint64 {
        // We assume we never overflow a uint64. Seems OK.
        hookID := m.allocIdx.Add(1)
        if hookID == 0 {
                panic("hookMap allocIdx overflow")
        }
        m.transient[hookID] = hook

        return hookID
}

// invoke is used on a hook map to call all the registered hooks and then clear
// out the transient hooks so they are not called again.
func (m *hookMap) invoke() {
        for _, hook := range m.transient {
                hook()
        }

        m.transient = make(map[uint64]func())
}

// hodlHtlc contains htlc data that is required for resolution.
type hodlHtlc struct {
        add        lnwire.UpdateAddHTLC
        sourceRef  channeldb.AddRef
        obfuscator hop.ErrorEncrypter
}

// NewChannelLink creates a new instance of a ChannelLink given a configuration
// and active channel that will be used to verify/apply updates to.
func NewChannelLink(cfg ChannelLinkConfig,
        channel *lnwallet.LightningChannel) ChannelLink {

        logPrefix := fmt.Sprintf("ChannelLink(%v):", channel.ChannelPoint())

        // If the max fee exposure isn't set, use the default.
        if cfg.MaxFeeExposure == 0 {
                cfg.MaxFeeExposure = DefaultMaxFeeExposure
        }

        var qsm Quiescer
        if !cfg.DisallowQuiescence {
                qsm = NewQuiescer(QuiescerCfg{
                        chanID: lnwire.NewChanIDFromOutPoint(
                                channel.ChannelPoint(),
                        ),
                        channelInitiator: channel.Initiator(),
                        sendMsg: func(s lnwire.Stfu) error {
                                return cfg.Peer.SendMessage(false, &s)
                        },
                        timeoutDuration: defaultQuiescenceTimeout,
                        onTimeout: func() {
                                cfg.Peer.Disconnect(ErrQuiescenceTimeout)
                        },
                })
        } else {
                qsm = &quiescerNoop{}
        }

        quiescenceReqs := make(
                chan fn.Req[fn.Unit, fn.Result[lntypes.ChannelParty]], 1,
        )

        return &channelLink{
                cfg:                 cfg,
                channel:             channel,
                hodlMap:             make(map[models.CircuitKey]hodlHtlc),
                hodlQueue:           queue.NewConcurrentQueue(10),
                log:                 log.WithPrefix(logPrefix),
                flushHooks:          newHookMap(),
                outgoingCommitHooks: newHookMap(),
                incomingCommitHooks: newHookMap(),
                quiescer:            qsm,
                quiescenceReqs:      quiescenceReqs,
                cg:                  fn.NewContextGuard(),
        }
}

// A compile time check to ensure channelLink implements the ChannelLink
// interface.
var _ ChannelLink = (*channelLink)(nil)

// Start starts all helper goroutines required for the operation of the channel
// link.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) Start() error {
        if !atomic.CompareAndSwapInt32(&l.started, 0, 1) {
                err := fmt.Errorf("channel link(%v): already started", l)
                l.log.Warn("already started")
                return err
        }

        l.log.Info("starting")

        // If the config supplied watchtower client, ensure the channel is
        // registered before trying to use it during operation.
        if l.cfg.TowerClient != nil {
                err := l.cfg.TowerClient.RegisterChannel(
                        l.ChanID(), l.channel.State().ChanType,
                )
                if err != nil {
                        return err
                }
        }

        l.mailBox.ResetMessages()
        l.hodlQueue.Start()

        // Before launching the htlcManager messages, revert any circuits that
        // were marked open in the switch's circuit map, but did not make it
        // into a commitment txn. We use the next local htlc index as the cut
        // off point, since all indexes below that are committed. This action
        // is only performed if the link's final short channel ID has been
        // assigned, otherwise we would try to trim the htlcs belonging to the
        // all-zero, hop.Source ID.
        if l.ShortChanID() != hop.Source {
                localHtlcIndex, err := l.channel.NextLocalHtlcIndex()
                if err != nil {
                        return fmt.Errorf("unable to retrieve next local "+
                                "htlc index: %v", err)
                }

                // NOTE: This is automatically done by the switch when it
                // starts up, but is necessary to prevent inconsistencies in
                // the case that the link flaps. This is a result of a link's
                // life-cycle being shorter than that of the switch.
                chanID := l.ShortChanID()
                err = l.cfg.Circuits.TrimOpenCircuits(chanID, localHtlcIndex)
                if err != nil {
                        return fmt.Errorf("unable to trim circuits above "+
                                "local htlc index %d: %v", localHtlcIndex, err)
                }

                // Since the link is live, before we start the link we'll update
                // the ChainArbitrator with the set of new channel signals for
                // this channel.
                //
                // TODO(roasbeef): split goroutines within channel arb to avoid
                go func() {
                        signals := &contractcourt.ContractSignals{
                                ShortChanID: l.channel.ShortChanID(),
                        }

                        err := l.cfg.UpdateContractSignals(signals)
                        if err != nil {
                                l.log.Errorf("unable to update signals")
                        }
                }()
        }

        l.updateFeeTimer = time.NewTimer(l.randomFeeUpdateTimeout())

        l.cg.WgAdd(1)
        go l.htlcManager(context.TODO())

        return nil
}

// Stop gracefully stops all active helper goroutines, then waits until they've
// exited.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) Stop() {
        if !atomic.CompareAndSwapInt32(&l.shutdown, 0, 1) {
                l.log.Warn("already stopped")
                return
        }

        l.log.Info("stopping")

        // As the link is stopping, we are no longer interested in htlc
        // resolutions coming from the invoice registry.
        l.cfg.Registry.HodlUnsubscribeAll(l.hodlQueue.ChanIn())

        if l.cfg.ChainEvents.Cancel != nil {
                l.cfg.ChainEvents.Cancel()
        }

        // Ensure the channel for the timer is drained.
        if l.updateFeeTimer != nil {
                if !l.updateFeeTimer.Stop() {
                        select {
                        case <-l.updateFeeTimer.C:
                        default:
                        }
                }
        }

        if l.hodlQueue != nil {
                l.hodlQueue.Stop()
        }

        l.cg.Quit()
        l.cg.WgWait()

        // Now that the htlcManager has completely exited, reset the packet
        // courier. This allows the mailbox to revaluate any lingering Adds that
        // were delivered but didn't make it on a commitment to be failed back
        // if the link is offline for an extended period of time. The error is
        // ignored since it can only fail when the daemon is exiting.
        _ = l.mailBox.ResetPackets()

        // As a final precaution, we will attempt to flush any uncommitted
        // preimages to the preimage cache. The preimages should be re-delivered
        // after channel reestablishment, however this adds an extra layer of
        // protection in case the peer never returns. Without this, we will be
        // unable to settle any contracts depending on the preimages even though
        // we had learned them at some point.
        err := l.cfg.PreimageCache.AddPreimages(l.uncommittedPreimages...)
        if err != nil {
                l.log.Errorf("unable to add preimages=%v to cache: %v",
                        l.uncommittedPreimages, err)
        }
}

// WaitForShutdown blocks until the link finishes shutting down, which includes
// termination of all dependent goroutines.
func (l *channelLink) WaitForShutdown() {
        l.cg.WgWait()
}

// EligibleToForward returns a bool indicating if the channel is able to
// actively accept requests to forward HTLC's. We're able to forward HTLC's if
// we are eligible to update AND the channel isn't currently flushing the
// outgoing half of the channel.
//
// NOTE: MUST NOT be called from the main event loop.
func (l *channelLink) EligibleToForward() bool {
        l.RLock()
        defer l.RUnlock()

        return l.eligibleToForward()
}

// eligibleToForward returns a bool indicating if the channel is able to
// actively accept requests to forward HTLC's. We're able to forward HTLC's if
// we are eligible to update AND the channel isn't currently flushing the
// outgoing half of the channel.
//
// NOTE: MUST be called from the main event loop.
func (l *channelLink) eligibleToForward() bool {
        return l.eligibleToUpdate() && !l.IsFlushing(Outgoing)
}

// eligibleToUpdate returns a bool indicating if the channel is able to update
// channel state. We're able to update channel state if we know the remote
// party's next revocation point. Otherwise, we can't initiate new channel
// state. We also require that the short channel ID not be the all-zero source
// ID, meaning that the channel has had its ID finalized.
//
// NOTE: MUST be called from the main event loop.
func (l *channelLink) eligibleToUpdate() bool {
        return l.channel.RemoteNextRevocation() != nil &&
                l.channel.ShortChanID() != hop.Source &&
                l.isReestablished() &&
                l.quiescer.CanSendUpdates()
}

// EnableAdds sets the ChannelUpdateHandler state to allow UpdateAddHtlc's in
// the specified direction. It returns true if the state was changed and false
// if the desired state was already set before the method was called.
func (l *channelLink) EnableAdds(linkDirection LinkDirection) bool {
        if linkDirection == Outgoing {
                return l.isOutgoingAddBlocked.Swap(false)
        }

        return l.isIncomingAddBlocked.Swap(false)
}

// DisableAdds sets the ChannelUpdateHandler state to allow UpdateAddHtlc's in
// the specified direction. It returns true if the state was changed and false
// if the desired state was already set before the method was called.
func (l *channelLink) DisableAdds(linkDirection LinkDirection) bool {
        if linkDirection == Outgoing {
                return !l.isOutgoingAddBlocked.Swap(true)
        }

        return !l.isIncomingAddBlocked.Swap(true)
}

// IsFlushing returns true when UpdateAddHtlc's are disabled in the direction of
// the argument.
func (l *channelLink) IsFlushing(linkDirection LinkDirection) bool {
        if linkDirection == Outgoing {
                return l.isOutgoingAddBlocked.Load()
        }

        return l.isIncomingAddBlocked.Load()
}

// OnFlushedOnce adds a hook that will be called the next time the channel
// state reaches zero htlcs. This hook will only ever be called once. If the
// channel state already has zero htlcs, then this will be called immediately.
func (l *channelLink) OnFlushedOnce(hook func()) {
        select {
        case l.flushHooks.newTransients <- hook:
        case <-l.cg.Done():
        }
}

// OnCommitOnce adds a hook that will be called the next time a CommitSig
// message is sent in the argument's LinkDirection. This hook will only ever be
// called once. If no CommitSig is owed in the argument's LinkDirection, then
// we will call this hook be run immediately.
func (l *channelLink) OnCommitOnce(direction LinkDirection, hook func()) {
        var queue chan func()

        if direction == Outgoing {
                queue = l.outgoingCommitHooks.newTransients
        } else {
                queue = l.incomingCommitHooks.newTransients
        }

        select {
        case queue <- hook:
        case <-l.cg.Done():
        }
}

// InitStfu allows us to initiate quiescence on this link. It returns a receive
// only channel that will block until quiescence has been achieved, or
// definitively fails.
//
// This operation has been added to allow channels to be quiesced via RPC. It
// may be removed or reworked in the future as RPC initiated quiescence is a
// holdover until we have downstream protocols that use it.
func (l *channelLink) InitStfu() <-chan fn.Result[lntypes.ChannelParty] {
        req, out := fn.NewReq[fn.Unit, fn.Result[lntypes.ChannelParty]](
                fn.Unit{},
        )

        select {
        case l.quiescenceReqs <- req:
        case <-l.cg.Done():
                req.Resolve(fn.Err[lntypes.ChannelParty](ErrLinkShuttingDown))
        }

        return out
}

// isReestablished returns true if the link has successfully completed the
// channel reestablishment dance.
func (l *channelLink) isReestablished() bool {
        return atomic.LoadInt32(&l.reestablished) == 1
}

// markReestablished signals that the remote peer has successfully exchanged
// channel reestablish messages and that the channel is ready to process
// subsequent messages.
func (l *channelLink) markReestablished() {
        atomic.StoreInt32(&l.reestablished, 1)
}

// IsUnadvertised returns true if the underlying channel is unadvertised.
func (l *channelLink) IsUnadvertised() bool {
        state := l.channel.State()
        return state.ChannelFlags&lnwire.FFAnnounceChannel == 0
}

// sampleNetworkFee samples the current fee rate on the network to get into the
// chain in a timely manner. The returned value is expressed in fee-per-kw, as
// this is the native rate used when computing the fee for commitment
// transactions, and the second-level HTLC transactions.
func (l *channelLink) sampleNetworkFee() (chainfee.SatPerKWeight, error) {
        // We'll first query for the sat/kw recommended to be confirmed within 3
        // blocks.
        feePerKw, err := l.cfg.FeeEstimator.EstimateFeePerKW(3)
        if err != nil {
                return 0, err
        }

        l.log.Debugf("sampled fee rate for 3 block conf: %v sat/kw",
                int64(feePerKw))

        return feePerKw, nil
}

// shouldAdjustCommitFee returns true if we should update our commitment fee to
// match that of the network fee. We'll only update our commitment fee if the
// network fee is +/- 10% to our commitment fee or if our current commitment
// fee is below the minimum relay fee.
func shouldAdjustCommitFee(netFee, chanFee,
        minRelayFee chainfee.SatPerKWeight) bool {

        switch {
        // If the network fee is greater than our current commitment fee and
        // our current commitment fee is below the minimum relay fee then
        // we should switch to it no matter if it is less than a 10% increase.
        case netFee > chanFee && chanFee < minRelayFee:
                return true

        // If the network fee is greater than the commitment fee, then we'll
        // switch to it if it's at least 10% greater than the commit fee.
        case netFee > chanFee && netFee >= (chanFee+(chanFee*10)/100):
                return true

        // If the network fee is less than our commitment fee, then we'll
        // switch to it if it's at least 10% less than the commitment fee.
        case netFee < chanFee && netFee <= (chanFee-(chanFee*10)/100):
                return true

        // Otherwise, we won't modify our fee.
        default:
                return false
        }
}

// failCb is used to cut down on the argument verbosity.
type failCb func(update *lnwire.ChannelUpdate1) lnwire.FailureMessage

// createFailureWithUpdate creates a ChannelUpdate when failing an incoming or
// outgoing HTLC. It may return a FailureMessage that references a channel's
// alias. If the channel does not have an alias, then the regular channel
// update from disk will be returned.
func (l *channelLink) createFailureWithUpdate(incoming bool,
        outgoingScid lnwire.ShortChannelID, cb failCb) lnwire.FailureMessage {

        // Determine which SCID to use in case we need to use aliases in the
        // ChannelUpdate.
        scid := outgoingScid
        if incoming {
                scid = l.ShortChanID()
        }

        // Try using the FailAliasUpdate function. If it returns nil, fallback
        // to the non-alias behavior.
        update := l.cfg.FailAliasUpdate(scid, incoming)
        if update == nil {
                // Fallback to the non-alias behavior.
                var err error
                update, err = l.cfg.FetchLastChannelUpdate(l.ShortChanID())
                if err != nil {
                        return &lnwire.FailTemporaryNodeFailure{}
                }
        }

        return cb(update)
}

// syncChanState attempts to synchronize channel states with the remote party.
// This method is to be called upon reconnection after the initial funding
// flow. We'll compare out commitment chains with the remote party, and re-send
// either a danging commit signature, a revocation, or both.
func (l *channelLink) syncChanStates(ctx context.Context) error {
        chanState := l.channel.State()

        l.log.Infof("Attempting to re-synchronize channel: %v", chanState)

        // First, we'll generate our ChanSync message to send to the other
        // side. Based on this message, the remote party will decide if they
        // need to retransmit any data or not.
        localChanSyncMsg, err := chanState.ChanSyncMsg()
        if err != nil {
                return fmt.Errorf("unable to generate chan sync message for "+
                        "ChannelPoint(%v)", l.channel.ChannelPoint())
        }
        if err := l.cfg.Peer.SendMessage(true, localChanSyncMsg); err != nil {
                return fmt.Errorf("unable to send chan sync message for "+
                        "ChannelPoint(%v): %v", l.channel.ChannelPoint(), err)
        }

        var msgsToReSend []lnwire.Message

        // Next, we'll wait indefinitely to receive the ChanSync message. The
        // first message sent MUST be the ChanSync message.
        select {
        case msg := <-l.upstream:
                l.log.Tracef("Received msg=%v from peer(%x)", msg.MsgType(),
                        l.cfg.Peer.PubKey())

                remoteChanSyncMsg, ok := msg.(*lnwire.ChannelReestablish)
                if !ok {
                        return fmt.Errorf("first message sent to sync "+
                                "should be ChannelReestablish, instead "+
                                "received: %T", msg)
                }

                // If the remote party indicates that they think we haven't
                // done any state updates yet, then we'll retransmit the
                // channel_ready message first. We do this, as at this point
                // we can't be sure if they've really received the
                // ChannelReady message.
                if remoteChanSyncMsg.NextLocalCommitHeight == 1 &&
                        localChanSyncMsg.NextLocalCommitHeight == 1 &&
                        !l.channel.IsPending() {

                        l.log.Infof("resending ChannelReady message to peer")

                        nextRevocation, err := l.channel.NextRevocationKey()
                        if err != nil {
                                return fmt.Errorf("unable to create next "+
                                        "revocation: %v", err)
                        }

                        channelReadyMsg := lnwire.NewChannelReady(
                                l.ChanID(), nextRevocation,
                        )

                        // If this is a taproot channel, then we'll send the
                        // very same nonce that we sent above, as they should
                        // take the latest verification nonce we send.
                        if chanState.ChanType.IsTaproot() {
                                //nolint:ll
                                channelReadyMsg.NextLocalNonce = localChanSyncMsg.LocalNonce
                        }

                        // For channels that negotiated the option-scid-alias
                        // feature bit, ensure that we send over the alias in
                        // the channel_ready message. We'll send the first
                        // alias we find for the channel since it does not
                        // matter which alias we send. We'll error out if no
                        // aliases are found.
                        if l.negotiatedAliasFeature() {
                                aliases := l.getAliases()
                                if len(aliases) == 0 {
                                        // This shouldn't happen since we
                                        // always add at least one alias before
                                        // the channel reaches the link.
                                        return fmt.Errorf("no aliases found")
                                }

                                // getAliases returns a copy of the alias slice
                                // so it is ok to use a pointer to the first
                                // entry.
                                channelReadyMsg.AliasScid = &aliases[0]
                        }

                        err = l.cfg.Peer.SendMessage(false, channelReadyMsg)
                        if err != nil {
                                return fmt.Errorf("unable to re-send "+
                                        "ChannelReady: %v", err)
                        }
                }

                // In any case, we'll then process their ChanSync message.
                l.log.Info("received re-establishment message from remote side")

                var (
                        openedCircuits []CircuitKey
                        closedCircuits []CircuitKey
                )

                // We've just received a ChanSync message from the remote
                // party, so we'll process the message  in order to determine
                // if we need to re-transmit any messages to the remote party.
                ctx, cancel := l.cg.Create(ctx)
                defer cancel()
                msgsToReSend, openedCircuits, closedCircuits, err =
                        l.channel.ProcessChanSyncMsg(ctx, remoteChanSyncMsg)
                if err != nil {
                        return err
                }

                // Repopulate any identifiers for circuits that may have been
                // opened or unclosed. This may happen if we needed to
                // retransmit a commitment signature message.
                l.openedCircuits = openedCircuits
                l.closedCircuits = closedCircuits

                // Ensure that all packets have been have been removed from the
                // link's mailbox.
                if err := l.ackDownStreamPackets(); err != nil {
                        return err
                }

                if len(msgsToReSend) > 0 {
                        l.log.Infof("sending %v updates to synchronize the "+
                                "state", len(msgsToReSend))
                }

                // If we have any messages to retransmit, we'll do so
                // immediately so we return to a synchronized state as soon as
                // possible.
                for _, msg := range msgsToReSend {
                        l.cfg.Peer.SendMessage(false, msg)
                }

        case <-l.cg.Done():
                return ErrLinkShuttingDown
        }

        return nil
}

// resolveFwdPkgs loads any forwarding packages for this link from disk, and
// reprocesses them in order. The primary goal is to make sure that any HTLCs
// we previously received are reinstated in memory, and forwarded to the switch
// if necessary. After a restart, this will also delete any previously
// completed packages.
func (l *channelLink) resolveFwdPkgs(ctx context.Context) error {
        fwdPkgs, err := l.channel.LoadFwdPkgs()
        if err != nil {
                return err
        }

        l.log.Debugf("loaded %d fwd pks", len(fwdPkgs))

        for _, fwdPkg := range fwdPkgs {
                if err := l.resolveFwdPkg(fwdPkg); err != nil {
                        return err
                }
        }

        // If any of our reprocessing steps require an update to the commitment
        // txn, we initiate a state transition to capture all relevant changes.
        if l.channel.NumPendingUpdates(lntypes.Local, lntypes.Remote) > 0 {
                return l.updateCommitTx(ctx)
        }

        return nil
}

// resolveFwdPkg interprets the FwdState of the provided package, either
// reprocesses any outstanding htlcs in the package, or performs garbage
// collection on the package.
func (l *channelLink) resolveFwdPkg(fwdPkg *channeldb.FwdPkg) error {
        // Remove any completed packages to clear up space.
        if fwdPkg.State == channeldb.FwdStateCompleted {
                l.log.Debugf("removing completed fwd pkg for height=%d",
                        fwdPkg.Height)

                err := l.channel.RemoveFwdPkgs(fwdPkg.Height)
                if err != nil {
                        l.log.Errorf("unable to remove fwd pkg for height=%d: "+
                                "%v", fwdPkg.Height, err)
                        return err
                }
        }

        // Otherwise this is either a new package or one has gone through
        // processing, but contains htlcs that need to be restored in memory.
        // We replay this forwarding package to make sure our local mem state
        // is resurrected, we mimic any original responses back to the remote
        // party, and re-forward the relevant HTLCs to the switch.

        // If the package is fully acked but not completed, it must still have
        // settles and fails to propagate.
        if !fwdPkg.SettleFailFilter.IsFull() {
                l.processRemoteSettleFails(fwdPkg)
        }

        // Finally, replay *ALL ADDS* in this forwarding package. The
        // downstream logic is able to filter out any duplicates, but we must
        // shove the entire, original set of adds down the pipeline so that the
        // batch of adds presented to the sphinx router does not ever change.
        if !fwdPkg.AckFilter.IsFull() {
                l.processRemoteAdds(fwdPkg)

                // If the link failed during processing the adds, we must
                // return to ensure we won't attempted to update the state
                // further.
                if l.failed {
                        return fmt.Errorf("link failed while " +
                                "processing remote adds")
                }
        }

        return nil
}

// fwdPkgGarbager periodically reads all forwarding packages from disk and
// removes those that can be discarded. It is safe to do this entirely in the
// background, since all state is coordinated on disk. This also ensures the
// link can continue to process messages and interleave database accesses.
//
// NOTE: This MUST be run as a goroutine.
func (l *channelLink) fwdPkgGarbager() {
        defer l.cg.WgDone()

        l.cfg.FwdPkgGCTicker.Resume()
        defer l.cfg.FwdPkgGCTicker.Stop()

        if err := l.loadAndRemove(); err != nil {
                l.log.Warnf("unable to run initial fwd pkgs gc: %v", err)
        }

        for {
                select {
                case <-l.cfg.FwdPkgGCTicker.Ticks():
                        if err := l.loadAndRemove(); err != nil {
                                l.log.Warnf("unable to remove fwd pkgs: %v",
                                        err)
                                continue
                        }
                case <-l.cg.Done():
                        return
                }
        }
}

// loadAndRemove loads all the channels forwarding packages and determines if
// they can be removed. It is called once before the FwdPkgGCTicker ticks so that
// a longer tick interval can be used.
func (l *channelLink) loadAndRemove() error {
        fwdPkgs, err := l.channel.LoadFwdPkgs()
        if err != nil {
                return err
        }

        var removeHeights []uint64
        for _, fwdPkg := range fwdPkgs {
                if fwdPkg.State != channeldb.FwdStateCompleted {
                        continue
                }

                removeHeights = append(removeHeights, fwdPkg.Height)
        }

        // If removeHeights is empty, return early so we don't use a db
        // transaction.
        if len(removeHeights) == 0 {
                return nil
        }

        return l.channel.RemoveFwdPkgs(removeHeights...)
}

// handleChanSyncErr performs the error handling logic in the case where we
// could not successfully syncChanStates with our channel peer.
func (l *channelLink) handleChanSyncErr(err error) {
        l.log.Warnf("error when syncing channel states: %v", err)

        var errDataLoss *lnwallet.ErrCommitSyncLocalDataLoss

        switch {
        case errors.Is(err, ErrLinkShuttingDown):
                l.log.Debugf("unable to sync channel states, link is " +
                        "shutting down")
                return

        // We failed syncing the commit chains, probably because the remote has
        // lost state. We should force close the channel.
        case errors.Is(err, lnwallet.ErrCommitSyncRemoteDataLoss):
                fallthrough

        // The remote sent us an invalid last commit secret, we should force
        // close the channel.
        // TODO(halseth): and permanently ban the peer?
        case errors.Is(err, lnwallet.ErrInvalidLastCommitSecret):
                fallthrough

        // The remote sent us a commit point different from what they sent us
        // before.
        // TODO(halseth): ban peer?
        case errors.Is(err, lnwallet.ErrInvalidLocalUnrevokedCommitPoint):
                // We'll fail the link and tell the peer to force close the
                // channel. Note that the database state is not updated here,
                // but will be updated when the close transaction is ready to
                // avoid that we go down before storing the transaction in the
                // db.
                l.failf(
                        LinkFailureError{
                                code:          ErrSyncError,
                                FailureAction: LinkFailureForceClose,
                        },
                        "unable to synchronize channel states: %v", err,
                )

        // We have lost state and cannot safely force close the channel. Fail
        // the channel and wait for the remote to hopefully force close it. The
        // remote has sent us its latest unrevoked commitment point, and we'll
        // store it in the database, such that we can attempt to recover the
        // funds if the remote force closes the channel.
        case errors.As(err, &errDataLoss):
                err := l.channel.MarkDataLoss(
                        errDataLoss.CommitPoint,
                )
                if err != nil {
                        l.log.Errorf("unable to mark channel data loss: %v",
                                err)
                }

        // We determined the commit chains were not possible to sync. We
        // cautiously fail the channel, but don't force close.
        // TODO(halseth): can we safely force close in any cases where this
        // error is returned?
        case errors.Is(err, lnwallet.ErrCannotSyncCommitChains):
                if err := l.channel.MarkBorked(); err != nil {
                        l.log.Errorf("unable to mark channel borked: %v", err)
                }

        // Other, unspecified error.
        default:
        }

        l.failf(
                LinkFailureError{
                        code:          ErrRecoveryError,
                        FailureAction: LinkFailureForceNone,
                },
                "unable to synchronize channel states: %v", err,
        )
}

// htlcManager is the primary goroutine which drives a channel's commitment
// update state-machine in response to messages received via several channels.
// This goroutine reads messages from the upstream (remote) peer, and also from
// downstream channel managed by the channel link. In the event that an htlc
// needs to be forwarded, then send-only forward handler is used which sends
// htlc packets to the switch. Additionally, this goroutine handles acting upon
// all timeouts for any active HTLCs, manages the channel's revocation window,
// and also the htlc trickle queue+timer for this active channels.
//
// NOTE: This MUST be run as a goroutine.
//
//nolint:funlen
func (l *channelLink) htlcManager(ctx context.Context) {
        defer func() {
                l.cfg.BatchTicker.Stop()
                l.cg.WgDone()
                l.log.Infof("exited")
        }()

        l.log.Infof("HTLC manager started, bandwidth=%v", l.Bandwidth())

        // Notify any clients that the link is now in the switch via an
        // ActiveLinkEvent. We'll also defer an inactive link notification for
        // when the link exits to ensure that every active notification is
        // matched by an inactive one.
        l.cfg.NotifyActiveLink(l.ChannelPoint())
        defer l.cfg.NotifyInactiveLinkEvent(l.ChannelPoint())

        // TODO(roasbeef): need to call wipe chan whenever D/C?

        // If this isn't the first time that this channel link has been
        // created, then we'll need to check to see if we need to
        // re-synchronize state with the remote peer. settledHtlcs is a map of
        // HTLC's that we re-settled as part of the channel state sync.
        if l.cfg.SyncStates {
                err := l.syncChanStates(ctx)
                if err != nil {
                        l.handleChanSyncErr(err)
                        return
                }
        }

        // If a shutdown message has previously been sent on this link, then we
        // need to make sure that we have disabled any HTLC adds on the outgoing
        // direction of the link and that we re-resend the same shutdown message
        // that we previously sent.
        l.cfg.PreviouslySentShutdown.WhenSome(func(shutdown lnwire.Shutdown) {
                // Immediately disallow any new outgoing HTLCs.
                if !l.DisableAdds(Outgoing) {
                        l.log.Warnf("Outgoing link adds already disabled")
                }

                // Re-send the shutdown message the peer. Since syncChanStates
                // would have sent any outstanding CommitSig, it is fine for us
                // to immediately queue the shutdown message now.
                err := l.cfg.Peer.SendMessage(false, &shutdown)
                if err != nil {
                        l.log.Warnf("Error sending shutdown message: %v", err)
                }
        })

        // We've successfully reestablished the channel, mark it as such to
        // allow the switch to forward HTLCs in the outbound direction.
        l.markReestablished()

        // Now that we've received both channel_ready and channel reestablish,
        // we can go ahead and send the active channel notification. We'll also
        // defer the inactive notification for when the link exits to ensure
        // that every active notification is matched by an inactive one.
        l.cfg.NotifyActiveChannel(l.ChannelPoint())
        defer l.cfg.NotifyInactiveChannel(l.ChannelPoint())

        // With the channel states synced, we now reset the mailbox to ensure
        // we start processing all unacked packets in order. This is done here
        // to ensure that all acknowledgments that occur during channel
        // resynchronization have taken affect, causing us only to pull unacked
        // packets after starting to read from the downstream mailbox.
        l.mailBox.ResetPackets()

        // After cleaning up any memory pertaining to incoming packets, we now
        // replay our forwarding packages to handle any htlcs that can be
        // processed locally, or need to be forwarded out to the switch. We will
        // only attempt to resolve packages if our short chan id indicates that
        // the channel is not pending, otherwise we should have no htlcs to
        // reforward.
        if l.ShortChanID() != hop.Source {
                err := l.resolveFwdPkgs(ctx)
                switch err {
                // No error was encountered, success.
                case nil:

                // If the duplicate keystone error was encountered, we'll fail
                // without sending an Error message to the peer.
                case ErrDuplicateKeystone:
                        l.failf(LinkFailureError{code: ErrCircuitError},
                                "temporary circuit error: %v", err)
                        return

                // A non-nil error was encountered, send an Error message to
                // the peer.
                default:
                        l.failf(LinkFailureError{code: ErrInternalError},
                                "unable to resolve fwd pkgs: %v", err)
                        return
                }

                // With our link's in-memory state fully reconstructed, spawn a
                // goroutine to manage the reclamation of disk space occupied by
                // completed forwarding packages.
                l.cg.WgAdd(1)
                go l.fwdPkgGarbager()
        }

        for {
                // We must always check if we failed at some point processing
                // the last update before processing the next.
                if l.failed {
                        l.log.Errorf("link failed, exiting htlcManager")
                        return
                }

                // If the previous event resulted in a non-empty batch, resume
                // the batch ticker so that it can be cleared. Otherwise pause
                // the ticker to prevent waking up the htlcManager while the
                // batch is empty.
                numUpdates := l.channel.NumPendingUpdates(
                        lntypes.Local, lntypes.Remote,
                )
                if numUpdates > 0 {
                        l.cfg.BatchTicker.Resume()
                        l.log.Tracef("BatchTicker resumed, "+
                                "NumPendingUpdates(Local, Remote)=%d",
                                numUpdates,
                        )
                } else {
                        l.cfg.BatchTicker.Pause()
                        l.log.Trace("BatchTicker paused due to zero " +
                                "NumPendingUpdates(Local, Remote)")
                }

                select {
                // We have a new hook that needs to be run when we reach a clean
                // channel state.
                case hook := <-l.flushHooks.newTransients:
                        if l.channel.IsChannelClean() {
                                hook()
                        } else {
                                l.flushHooks.alloc(hook)
                        }

                // We have a new hook that needs to be run when we have
                // committed all of our updates.
                case hook := <-l.outgoingCommitHooks.newTransients:
                        if !l.channel.OweCommitment() {
                                hook()
                        } else {
                                l.outgoingCommitHooks.alloc(hook)
                        }

                // We have a new hook that needs to be run when our peer has
                // committed all of their updates.
                case hook := <-l.incomingCommitHooks.newTransients:
                        if !l.channel.NeedCommitment() {
                                hook()
                        } else {
                                l.incomingCommitHooks.alloc(hook)
                        }

                // Our update fee timer has fired, so we'll check the network
                // fee to see if we should adjust our commitment fee.
                case <-l.updateFeeTimer.C:
                        l.updateFeeTimer.Reset(l.randomFeeUpdateTimeout())

                        // If we're not the initiator of the channel, don't we
                        // don't control the fees, so we can ignore this.
                        if !l.channel.IsInitiator() {
                                continue
                        }

                        // If we are the initiator, then we'll sample the
                        // current fee rate to get into the chain within 3
                        // blocks.
                        netFee, err := l.sampleNetworkFee()
                        if err != nil {
                                l.log.Errorf("unable to sample network fee: %v",
                                        err)
                                continue
                        }

                        minRelayFee := l.cfg.FeeEstimator.RelayFeePerKW()

                        newCommitFee := l.channel.IdealCommitFeeRate(
                                netFee, minRelayFee,
                                l.cfg.MaxAnchorsCommitFeeRate,
                                l.cfg.MaxFeeAllocation,
                        )

                        // We determine if we should adjust the commitment fee
                        // based on the current commitment fee, the suggested
                        // new commitment fee and the current minimum relay fee
                        // rate.
                        commitFee := l.channel.CommitFeeRate()
                        if !shouldAdjustCommitFee(
                                newCommitFee, commitFee, minRelayFee,
                        ) {

                                continue
                        }

                        // If we do, then we'll send a new UpdateFee message to
                        // the remote party, to be locked in with a new update.
                        err = l.updateChannelFee(ctx, newCommitFee)
                        if err != nil {
                                l.log.Errorf("unable to update fee rate: %v",
                                        err)
                                continue
                        }

                // The underlying channel has notified us of a unilateral close
                // carried out by the remote peer. In the case of such an
                // event, we'll wipe the channel state from the peer, and mark
                // the contract as fully settled. Afterwards we can exit.
                //
                // TODO(roasbeef): add force closure? also breach?
                case <-l.cfg.ChainEvents.RemoteUnilateralClosure:
                        l.log.Warnf("remote peer has closed on-chain")

                        // TODO(roasbeef): remove all together
                        go func() {
                                chanPoint := l.channel.ChannelPoint()
                                l.cfg.Peer.WipeChannel(&chanPoint)
                        }()

                        return

                case <-l.cfg.BatchTicker.Ticks():
                        // Attempt to extend the remote commitment chain
                        // including all the currently pending entries. If the
                        // send was unsuccessful, then abandon the update,
                        // waiting for the revocation window to open up.
                        if !l.updateCommitTxOrFail(ctx) {
                                return
                        }

                case <-l.cfg.PendingCommitTicker.Ticks():
                        l.failf(
                                LinkFailureError{
                                        code:          ErrRemoteUnresponsive,
                                        FailureAction: LinkFailureDisconnect,
                                },
                                "unable to complete dance",
                        )
                        return

                // A message from the switch was just received. This indicates
                // that the link is an intermediate hop in a multi-hop HTLC
                // circuit.
                case pkt := <-l.downstream:
                        l.handleDownstreamPkt(ctx, pkt)

                // A message from the connected peer was just received. This
                // indicates that we have a new incoming HTLC, either directly
                // for us, or part of a multi-hop HTLC circuit.
                case msg := <-l.upstream:
                        l.handleUpstreamMsg(ctx, msg)

                // A htlc resolution is received. This means that we now have a
                // resolution for a previously accepted htlc.
                case hodlItem := <-l.hodlQueue.ChanOut():
                        htlcResolution := hodlItem.(invoices.HtlcResolution)
                        err := l.processHodlQueue(ctx, htlcResolution)
                        switch err {
                        // No error, success.
                        case nil:

                        // If the duplicate keystone error was encountered,
                        // fail back gracefully.
                        case ErrDuplicateKeystone:
                                l.failf(LinkFailureError{
                                        code: ErrCircuitError,
                                }, "process hodl queue: "+
                                        "temporary circuit error: %v",
                                        err,
                                )

                        // Send an Error message to the peer.
                        default:
                                l.failf(LinkFailureError{
                                        code: ErrInternalError,
                                }, "process hodl queue: unable to update "+
                                        "commitment: %v", err,
                                )
                        }

                case qReq := <-l.quiescenceReqs:
                        l.quiescer.InitStfu(qReq)

                        if l.noDanglingUpdates(lntypes.Local) {
                                err := l.quiescer.SendOwedStfu()
                                if err != nil {
                                        l.stfuFailf(
                                                "SendOwedStfu: %s", err.Error(),
                                        )
                                        res := fn.Err[lntypes.ChannelParty](err)
                                        qReq.Resolve(res)
                                }
                        }

                case <-l.cg.Done():
                        return
                }
        }
}

// processHodlQueue processes a received htlc resolution and continues reading
// from the hodl queue until no more resolutions remain. When this function
// returns without an error, the commit tx should be updated.
func (l *channelLink) processHodlQueue(ctx context.Context,
        firstResolution invoices.HtlcResolution) error {

        // Try to read all waiting resolution messages, so that they can all be
        // processed in a single commitment tx update.
        htlcResolution := firstResolution
loop:
        for {
                // Lookup all hodl htlcs that can be failed or settled with this event.
                // The hodl htlc must be present in the map.
                circuitKey := htlcResolution.CircuitKey()
                hodlHtlc, ok := l.hodlMap[circuitKey]
                if !ok {
                        return fmt.Errorf("hodl htlc not found: %v", circuitKey)
                }

                if err := l.processHtlcResolution(htlcResolution, hodlHtlc); err != nil {
                        return err
                }

                // Clean up hodl map.
                delete(l.hodlMap, circuitKey)

                select {
                case item := <-l.hodlQueue.ChanOut():
                        htlcResolution = item.(invoices.HtlcResolution)
                default:
                        break loop
                }
        }

        // Update the commitment tx.
        if err := l.updateCommitTx(ctx); err != nil {
                return err
        }

        return nil
}

// processHtlcResolution applies a received htlc resolution to the provided
// htlc. When this function returns without an error, the commit tx should be
// updated.
func (l *channelLink) processHtlcResolution(resolution invoices.HtlcResolution,
        htlc hodlHtlc) error {

        circuitKey := resolution.CircuitKey()

        // Determine required action for the resolution based on the type of
        // resolution we have received.
        switch res := resolution.(type) {
        // Settle htlcs that returned a settle resolution using the preimage
        // in the resolution.
        case *invoices.HtlcSettleResolution:
                l.log.Debugf("received settle resolution for %v "+
                        "with outcome: %v", circuitKey, res.Outcome)

                return l.settleHTLC(
                        res.Preimage, htlc.add.ID, htlc.sourceRef,
                )

        // For htlc failures, we get the relevant failure message based
        // on the failure resolution and then fail the htlc.
        case *invoices.HtlcFailResolution:
                l.log.Debugf("received cancel resolution for "+
                        "%v with outcome: %v", circuitKey, res.Outcome)

                // Get the lnwire failure message based on the resolution
                // result.
                failure := getResolutionFailure(res, htlc.add.Amount)

                l.sendHTLCError(
                        htlc.add, htlc.sourceRef, failure, htlc.obfuscator,
                        true,
                )
                return nil

        // Fail if we do not get a settle of fail resolution, since we
        // are only expecting to handle settles and fails.
        default:
                return fmt.Errorf("unknown htlc resolution type: %T",
                        resolution)
        }
}

// getResolutionFailure returns the wire message that a htlc resolution should
// be failed with.
func getResolutionFailure(resolution *invoices.HtlcFailResolution,
        amount lnwire.MilliSatoshi) *LinkError {

        // If the resolution has been resolved as part of a MPP timeout,
        // we need to fail the htlc with lnwire.FailMppTimeout.
        if resolution.Outcome == invoices.ResultMppTimeout {
                return NewDetailedLinkError(
                        &lnwire.FailMPPTimeout{}, resolution.Outcome,
                )
        }

        // If the htlc is not a MPP timeout, we fail it with
        // FailIncorrectDetails. This error is sent for invoice payment
        // failures such as underpayment/ expiry too soon and hodl invoices
        // (which return FailIncorrectDetails to avoid leaking information).
        incorrectDetails := lnwire.NewFailIncorrectDetails(
                amount, uint32(resolution.AcceptHeight),
        )

        return NewDetailedLinkError(incorrectDetails, resolution.Outcome)
}

// randomFeeUpdateTimeout returns a random timeout between the bounds defined
// within the link's configuration that will be used to determine when the link
// should propose an update to its commitment fee rate.
func (l *channelLink) randomFeeUpdateTimeout() time.Duration {
        lower := int64(l.cfg.MinUpdateTimeout)
        upper := int64(l.cfg.MaxUpdateTimeout)
        return time.Duration(prand.Int63n(upper-lower) + lower)
}

// handleDownstreamUpdateAdd processes an UpdateAddHTLC packet sent from the
// downstream HTLC Switch.
func (l *channelLink) handleDownstreamUpdateAdd(ctx context.Context,
        pkt *htlcPacket) error {

        htlc, ok := pkt.htlc.(*lnwire.UpdateAddHTLC)
        if !ok {
                return errors.New("not an UpdateAddHTLC packet")
        }

        // If we are flushing the link in the outgoing direction or we have
        // already sent Stfu, then we can't add new htlcs to the link and we
        // need to bounce it.
        if l.IsFlushing(Outgoing) || !l.quiescer.CanSendUpdates() {
                l.mailBox.FailAdd(pkt)

                return NewDetailedLinkError(
                        &lnwire.FailTemporaryChannelFailure{},
                        OutgoingFailureLinkNotEligible,
                )
        }

        // If hodl.AddOutgoing mode is active, we exit early to simulate
        // arbitrary delays between the switch adding an ADD to the
        // mailbox, and the HTLC being added to the commitment state.
        if l.cfg.HodlMask.Active(hodl.AddOutgoing) {
                l.log.Warnf(hodl.AddOutgoing.Warning())
                l.mailBox.AckPacket(pkt.inKey())
                return nil
        }

        // Check if we can add the HTLC here without exceededing the max fee
        // exposure threshold.
        if l.isOverexposedWithHtlc(htlc, false) {
                l.log.Debugf("Unable to handle downstream HTLC - max fee " +
                        "exposure exceeded")

                l.mailBox.FailAdd(pkt)

                return NewDetailedLinkError(
                        lnwire.NewTemporaryChannelFailure(nil),
                        OutgoingFailureDownstreamHtlcAdd,
                )
        }

        // A new payment has been initiated via the downstream channel,
        // so we add the new HTLC to our local log, then update the
        // commitment chains.
        htlc.ChanID = l.ChanID()
        openCircuitRef := pkt.inKey()

        // We enforce the fee buffer for the commitment transaction because
        // we are in control of adding this htlc. Nothing has locked-in yet so
        // we can securely enforce the fee buffer which is only relevant if we
        // are the initiator of the channel.
        index, err := l.channel.AddHTLC(htlc, &openCircuitRef)
        if err != nil {
                // The HTLC was unable to be added to the state machine,
                // as a result, we'll signal the switch to cancel the
                // pending payment.
                l.log.Warnf("Unable to handle downstream add HTLC: %v",
                        err)

                // Remove this packet from the link's mailbox, this
                // prevents it from being reprocessed if the link
                // restarts and resets it mailbox. If this response
                // doesn't make it back to the originating link, it will
                // be rejected upon attempting to reforward the Add to
                // the switch, since the circuit was never fully opened,
                // and the forwarding package shows it as
                // unacknowledged.
                l.mailBox.FailAdd(pkt)

                return NewDetailedLinkError(
                        lnwire.NewTemporaryChannelFailure(nil),
                        OutgoingFailureDownstreamHtlcAdd,
                )
        }

        l.log.Tracef("received downstream htlc: payment_hash=%x, "+
                "local_log_index=%v, pend_updates=%v",
                htlc.PaymentHash[:], index,
                l.channel.NumPendingUpdates(lntypes.Local, lntypes.Remote))

        pkt.outgoingChanID = l.ShortChanID()
        pkt.outgoingHTLCID = index
        htlc.ID = index

        l.log.Debugf("queueing keystone of ADD open circuit: %s->%s",
                pkt.inKey(), pkt.outKey())

        l.openedCircuits = append(l.openedCircuits, pkt.inKey())
        l.keystoneBatch = append(l.keystoneBatch, pkt.keystone())

        _ = l.cfg.Peer.SendMessage(false, htlc)

        // Send a forward event notification to htlcNotifier.
        l.cfg.HtlcNotifier.NotifyForwardingEvent(
                newHtlcKey(pkt),
                HtlcInfo{
                        IncomingTimeLock: pkt.incomingTimeout,
                        IncomingAmt:      pkt.incomingAmount,
                        OutgoingTimeLock: htlc.Expiry,
                        OutgoingAmt:      htlc.Amount,
                },
                getEventType(pkt),
        )

        l.tryBatchUpdateCommitTx(ctx)

        return nil
}

// handleDownstreamPkt processes an HTLC packet sent from the downstream HTLC
// Switch. Possible messages sent by the switch include requests to forward new
// HTLCs, timeout previously cleared HTLCs, and finally to settle currently
// cleared HTLCs with the upstream peer.
//
// TODO(roasbeef): add sync ntfn to ensure switch always has consistent view?
func (l *channelLink) handleDownstreamPkt(ctx context.Context,
        pkt *htlcPacket) {

        if pkt.htlc.MsgType().IsChannelUpdate() &&
                !l.quiescer.CanSendUpdates() {

                l.log.Warnf("unable to process channel update. "+
                        "ChannelID=%v is quiescent.", l.ChanID)

                return
        }

        switch htlc := pkt.htlc.(type) {
        case *lnwire.UpdateAddHTLC:
                // Handle add message. The returned error can be ignored,
                // because it is also sent through the mailbox.
                _ = l.handleDownstreamUpdateAdd(ctx, pkt)

        case *lnwire.UpdateFulfillHTLC:
                // If hodl.SettleOutgoing mode is active, we exit early to
                // simulate arbitrary delays between the switch adding the
                // SETTLE to the mailbox, and the HTLC being added to the
                // commitment state.
                if l.cfg.HodlMask.Active(hodl.SettleOutgoing) {
                        l.log.Warnf(hodl.SettleOutgoing.Warning())
                        l.mailBox.AckPacket(pkt.inKey())
                        return
                }

                // An HTLC we forward to the switch has just settled somewhere
                // upstream. Therefore we settle the HTLC within the our local
                // state machine.
                inKey := pkt.inKey()
                err := l.channel.SettleHTLC(
                        htlc.PaymentPreimage,
                        pkt.incomingHTLCID,
                        pkt.sourceRef,
                        pkt.destRef,
                        &inKey,
                )
                if err != nil {
                        l.log.Errorf("unable to settle incoming HTLC for "+
                                "circuit-key=%v: %v", inKey, err)

                        // If the HTLC index for Settle response was not known
                        // to our commitment state, it has already been
                        // cleaned up by a prior response. We'll thus try to
                        // clean up any lingering state to ensure we don't
                        // continue reforwarding.
                        if _, ok := err.(lnwallet.ErrUnknownHtlcIndex); ok {
                                l.cleanupSpuriousResponse(pkt)
                        }

                        // Remove the packet from the link's mailbox to ensure
                        // it doesn't get replayed after a reconnection.
                        l.mailBox.AckPacket(inKey)

                        return
                }

                l.log.Debugf("queueing removal of SETTLE closed circuit: "+
                        "%s->%s", pkt.inKey(), pkt.outKey())

                l.closedCircuits = append(l.closedCircuits, pkt.inKey())

                // With the HTLC settled, we'll need to populate the wire
                // message to target the specific channel and HTLC to be
                // canceled.
                htlc.ChanID = l.ChanID()
                htlc.ID = pkt.incomingHTLCID

                // Then we send the HTLC settle message to the connected peer
                // so we can continue the propagation of the settle message.
                l.cfg.Peer.SendMessage(false, htlc)

                // Send a settle event notification to htlcNotifier.
                l.cfg.HtlcNotifier.NotifySettleEvent(
                        newHtlcKey(pkt),
                        htlc.PaymentPreimage,
                        getEventType(pkt),
                )

                // Immediately update the commitment tx to minimize latency.
                l.updateCommitTxOrFail(ctx)

        case *lnwire.UpdateFailHTLC:
                // If hodl.FailOutgoing mode is active, we exit early to
                // simulate arbitrary delays between the switch adding a FAIL to
                // the mailbox, and the HTLC being added to the commitment
                // state.
                if l.cfg.HodlMask.Active(hodl.FailOutgoing) {
                        l.log.Warnf(hodl.FailOutgoing.Warning())
                        l.mailBox.AckPacket(pkt.inKey())
                        return
                }

                // An HTLC cancellation has been triggered somewhere upstream,
                // we'll remove then HTLC from our local state machine.
                inKey := pkt.inKey()
                err := l.channel.FailHTLC(
                        pkt.incomingHTLCID,
                        htlc.Reason,
                        pkt.sourceRef,
                        pkt.destRef,
                        &inKey,
                )
                if err != nil {
                        l.log.Errorf("unable to cancel incoming HTLC for "+
                                "circuit-key=%v: %v", inKey, err)

                        // If the HTLC index for Fail response was not known to
                        // our commitment state, it has already been cleaned up
                        // by a prior response. We'll thus try to clean up any
                        // lingering state to ensure we don't continue
                        // reforwarding.
                        if _, ok := err.(lnwallet.ErrUnknownHtlcIndex); ok {
                                l.cleanupSpuriousResponse(pkt)
                        }

                        // Remove the packet from the link's mailbox to ensure
                        // it doesn't get replayed after a reconnection.
                        l.mailBox.AckPacket(inKey)

                        return
                }

                l.log.Debugf("queueing removal of FAIL closed circuit: %s->%s",
                        pkt.inKey(), pkt.outKey())

                l.closedCircuits = append(l.closedCircuits, pkt.inKey())

                // With the HTLC removed, we'll need to populate the wire
                // message to target the specific channel and HTLC to be
                // canceled. The "Reason" field will have already been set
                // within the switch.
                htlc.ChanID = l.ChanID()
                htlc.ID = pkt.incomingHTLCID

                // We send the HTLC message to the peer which initially created
                // the HTLC. If the incoming blinding point is non-nil, we
                // know that we are a relaying node in a blinded path.
                // Otherwise, we're either an introduction node or not part of
                // a blinded path at all.
                if err := l.sendIncomingHTLCFailureMsg(
                        htlc.ID,
                        pkt.obfuscator,
                        htlc.Reason,
                ); err != nil {
                        l.log.Errorf("unable to send HTLC failure: %v",
                                err)

                        return
                }

                // If the packet does not have a link failure set, it failed
                // further down the route so we notify a forwarding failure.
                // Otherwise, we notify a link failure because it failed at our
                // node.
                if pkt.linkFailure != nil {
                        l.cfg.HtlcNotifier.NotifyLinkFailEvent(
                                newHtlcKey(pkt),
                                newHtlcInfo(pkt),
                                getEventType(pkt),
                                pkt.linkFailure,
                                false,
                        )
                } else {
                        l.cfg.HtlcNotifier.NotifyForwardingFailEvent(
                                newHtlcKey(pkt), getEventType(pkt),
                        )
                }

                // Immediately update the commitment tx to minimize latency.
                l.updateCommitTxOrFail(ctx)
        }
}

// tryBatchUpdateCommitTx updates the commitment transaction if the batch is
// full.
func (l *channelLink) tryBatchUpdateCommitTx(ctx context.Context) {
        pending := l.channel.NumPendingUpdates(lntypes.Local, lntypes.Remote)
        if pending < uint64(l.cfg.BatchSize) {
                return
        }

        l.updateCommitTxOrFail(ctx)
}

// cleanupSpuriousResponse attempts to ack any AddRef or SettleFailRef
// associated with this packet. If successful in doing so, it will also purge
// the open circuit from the circuit map and remove the packet from the link's
// mailbox.
func (l *channelLink) cleanupSpuriousResponse(pkt *htlcPacket) {
        inKey := pkt.inKey()

        l.log.Debugf("cleaning up spurious response for incoming "+
                "circuit-key=%v", inKey)

        // If the htlc packet doesn't have a source reference, it is unsafe to
        // proceed, as skipping this ack may cause the htlc to be reforwarded.
        if pkt.sourceRef == nil {
                l.log.Errorf("unable to cleanup response for incoming "+
                        "circuit-key=%v, does not contain source reference",
                        inKey)
                return
        }

        // If the source reference is present,  we will try to prevent this link
        // from resending the packet to the switch. To do so, we ack the AddRef
        // of the incoming HTLC belonging to this link.
        err := l.channel.AckAddHtlcs(*pkt.sourceRef)
        if err != nil {
                l.log.Errorf("unable to ack AddRef for incoming "+
                        "circuit-key=%v: %v", inKey, err)

                // If this operation failed, it is unsafe to attempt removal of
                // the destination reference or circuit, so we exit early. The
                // cleanup may proceed with a different packet in the future
                // that succeeds on this step.
                return
        }

        // Now that we know this link will stop retransmitting Adds to the
        // switch, we can begin to teardown the response reference and circuit
        // map.
        //
        // If the packet includes a destination reference, then a response for
        // this HTLC was locked into the outgoing channel. Attempt to remove
        // this reference, so we stop retransmitting the response internally.
        // Even if this fails, we will proceed in trying to delete the circuit.
        // When retransmitting responses, the destination references will be
        // cleaned up if an open circuit is not found in the circuit map.
        if pkt.destRef != nil {
                err := l.channel.AckSettleFails(*pkt.destRef)
                if err != nil {
                        l.log.Errorf("unable to ack SettleFailRef "+
                                "for incoming circuit-key=%v: %v",
                                inKey, err)
                }
        }

        l.log.Debugf("deleting circuit for incoming circuit-key=%x", inKey)

        // With all known references acked, we can now safely delete the circuit
        // from the switch's circuit map, as the state is no longer needed.
        err = l.cfg.Circuits.DeleteCircuits(inKey)
        if err != nil {
                l.log.Errorf("unable to delete circuit for "+
                        "circuit-key=%v: %v", inKey, err)
        }
}

// handleUpstreamMsg processes wire messages related to commitment state
// updates from the upstream peer. The upstream peer is the peer whom we have a
// direct channel with, updating our respective commitment chains.
//
//nolint:funlen
func (l *channelLink) handleUpstreamMsg(ctx context.Context,
        msg lnwire.Message) {

        // First check if the message is an update and we are capable of
        // receiving updates right now.
        if msg.MsgType().IsChannelUpdate() && !l.quiescer.CanRecvUpdates() {
                l.stfuFailf("update received after stfu: %T", msg)
                return
        }

        switch msg := msg.(type) {
        case *lnwire.UpdateAddHTLC:
                if l.IsFlushing(Incoming) {
                        // This is forbidden by the protocol specification.
                        // The best chance we have to deal with this is to drop
                        // the connection. This should roll back the channel
                        // state to the last CommitSig. If the remote has
                        // already sent a CommitSig we haven't received yet,
                        // channel state will be re-synchronized with a
                        // ChannelReestablish message upon reconnection and the
                        // protocol state that caused us to flush the link will
                        // be rolled back. In the event that there was some
                        // non-deterministic behavior in the remote that caused
                        // them to violate the protocol, we have a decent shot
                        // at correcting it this way, since reconnecting will
                        // put us in the cleanest possible state to try again.
                        //
                        // In addition to the above, it is possible for us to
                        // hit this case in situations where we improperly
                        // handle message ordering due to concurrency choices.
                        // An issue has been filed to address this here:
                        // https://github.com/lightningnetwork/lnd/issues/8393
                        l.failf(
                                LinkFailureError{
                                        code:             ErrInvalidUpdate,
                                        FailureAction:    LinkFailureDisconnect,
                                        PermanentFailure: false,
                                        Warning:          true,
                                },
                                "received add while link is flushing",
                        )

                        return
                }

                // Disallow htlcs with blinding points set if we haven't
                // enabled the feature. This saves us from having to process
                // the onion at all, but will only catch blinded payments
                // where we are a relaying node (as the blinding point will
                // be in the payload when we're the introduction node).
                if msg.BlindingPoint.IsSome() && l.cfg.DisallowRouteBlinding {
                        l.failf(LinkFailureError{code: ErrInvalidUpdate},
                                "blinding point included when route blinding "+
                                        "is disabled")

                        return
                }

                // We have to check the limit here rather than later in the
                // switch because the counterparty can keep sending HTLC's
                // without sending a revoke. This would mean that the switch
                // check would only occur later.
                if l.isOverexposedWithHtlc(msg, true) {
                        l.failf(LinkFailureError{code: ErrInternalError},
                                "peer sent us an HTLC that exceeded our max "+
                                        "fee exposure")

                        return
                }

                // We just received an add request from an upstream peer, so we
                // add it to our state machine, then add the HTLC to our
                // "settle" list in the event that we know the preimage.
                index, err := l.channel.ReceiveHTLC(msg)
                if err != nil {
                        l.failf(LinkFailureError{code: ErrInvalidUpdate},
                                "unable to handle upstream add HTLC: %v", err)
                        return
                }

                l.log.Tracef("receive upstream htlc with payment hash(%x), "+
                        "assigning index: %v", msg.PaymentHash[:], index)

        case *lnwire.UpdateFulfillHTLC:
                pre := msg.PaymentPreimage
                idx := msg.ID

                // Before we pipeline the settle, we'll check the set of active
                // htlc's to see if the related UpdateAddHTLC has been fully
                // locked-in.
                var lockedin bool
                htlcs := l.channel.ActiveHtlcs()
                for _, add := range htlcs {
                        // The HTLC will be outgoing and match idx.
                        if !add.Incoming && add.HtlcIndex == idx {
                                lockedin = true
                                break
                        }
                }

                if !lockedin {
                        l.failf(
                                LinkFailureError{code: ErrInvalidUpdate},
                                "unable to handle upstream settle",
                        )
                        return
                }

                if err := l.channel.ReceiveHTLCSettle(pre, idx); err != nil {
                        l.failf(
                                LinkFailureError{
                                        code:          ErrInvalidUpdate,
                                        FailureAction: LinkFailureForceClose,
                                },
                                "unable to handle upstream settle HTLC: %v", err,
                        )
                        return
                }

                settlePacket := &htlcPacket{
                        outgoingChanID: l.ShortChanID(),
                        outgoingHTLCID: idx,
                        htlc: &lnwire.UpdateFulfillHTLC{
                                PaymentPreimage: pre,
                        },
                }

                // Add the newly discovered preimage to our growing list of
                // uncommitted preimage. These will be written to the witness
                // cache just before accepting the next commitment signature
                // from the remote peer.
                l.uncommittedPreimages = append(l.uncommittedPreimages, pre)

                // Pipeline this settle, send it to the switch.
                go l.forwardBatch(false, settlePacket)

        case *lnwire.UpdateFailMalformedHTLC:
                // Convert the failure type encoded within the HTLC fail
                // message to the proper generic lnwire error code.
                var failure lnwire.FailureMessage
                switch msg.FailureCode {
                case lnwire.CodeInvalidOnionVersion:
                        failure = &lnwire.FailInvalidOnionVersion{
                                OnionSHA256: msg.ShaOnionBlob,
                        }
                case lnwire.CodeInvalidOnionHmac:
                        failure = &lnwire.FailInvalidOnionHmac{
                                OnionSHA256: msg.ShaOnionBlob,
                        }

                case lnwire.CodeInvalidOnionKey:
                        failure = &lnwire.FailInvalidOnionKey{
                                OnionSHA256: msg.ShaOnionBlob,
                        }

                // Handle malformed errors that are part of a blinded route.
                // This case is slightly different, because we expect every
                // relaying node in the blinded portion of the route to send
                // malformed errors. If we're also a relaying node, we're
                // likely going to switch this error out anyway for our own
                // malformed error, but we handle the case here for
                // completeness.
                case lnwire.CodeInvalidBlinding:
                        failure = &lnwire.FailInvalidBlinding{
                                OnionSHA256: msg.ShaOnionBlob,
                        }

                default:
                        l.log.Warnf("unexpected failure code received in "+
                                "UpdateFailMailformedHTLC: %v", msg.FailureCode)

                        // We don't just pass back the error we received from
                        // our successor. Otherwise we might report a failure
                        // that penalizes us more than needed. If the onion that
                        // we forwarded was correct, the node should have been
                        // able to send back its own failure. The node did not
                        // send back its own failure, so we assume there was a
                        // problem with the onion and report that back. We reuse
                        // the invalid onion key failure because there is no
                        // specific error for this case.
                        failure = &lnwire.FailInvalidOnionKey{
                                OnionSHA256: msg.ShaOnionBlob,
                        }
                }

                // With the error parsed, we'll convert the into it's opaque
                // form.
                var b bytes.Buffer
                if err := lnwire.EncodeFailure(&b, failure, 0); err != nil {
                        l.log.Errorf("unable to encode malformed error: %v", err)
                        return
                }

                // If remote side have been unable to parse the onion blob we
                // have sent to it, than we should transform the malformed HTLC
                // message to the usual HTLC fail message.
                err := l.channel.ReceiveFailHTLC(msg.ID, b.Bytes())
                if err != nil {
                        l.failf(LinkFailureError{code: ErrInvalidUpdate},
                                "unable to handle upstream fail HTLC: %v", err)
                        return
                }

        case *lnwire.UpdateFailHTLC:
                // Verify that the failure reason is at least 256 bytes plus
                // overhead.
                const minimumFailReasonLength = lnwire.FailureMessageLength +
                        2 + 2 + 32

                if len(msg.Reason) < minimumFailReasonLength {
                        // We've received a reason with a non-compliant length.
                        // Older nodes happily relay back these failures that
                        // may originate from a node further downstream.
                        // Therefore we can't just fail the channel.
                        //
                        // We want to be compliant ourselves, so we also can't
                        // pass back the reason unmodified. And we must make
                        // sure that we don't hit the magic length check of 260
                        // bytes in processRemoteSettleFails either.
                        //
                        // Because the reason is unreadable for the payer
                        // anyway, we just replace it by a compliant-length
                        // series of random bytes.
                        msg.Reason = make([]byte, minimumFailReasonLength)
                        _, err := crand.Read(msg.Reason[:])
                        if err != nil {
                                l.log.Errorf("Random generation error: %v", err)

                                return
                        }
                }

                // Add fail to the update log.
                idx := msg.ID
                err := l.channel.ReceiveFailHTLC(idx, msg.Reason[:])
                if err != nil {
                        l.failf(LinkFailureError{code: ErrInvalidUpdate},
                                "unable to handle upstream fail HTLC: %v", err)
                        return
                }

        case *lnwire.CommitSig:
                // Since we may have learned new preimages for the first time,
                // we'll add them to our preimage cache. By doing this, we
                // ensure any contested contracts watched by any on-chain
                // arbitrators can now sweep this HTLC on-chain. We delay
                // committing the preimages until just before accepting the new
                // remote commitment, as afterwards the peer won't resend the
                // Settle messages on the next channel reestablishment. Doing so
                // allows us to more effectively batch this operation, instead
                // of doing a single write per preimage.
                err := l.cfg.PreimageCache.AddPreimages(
                        l.uncommittedPreimages...,
                )
                if err != nil {
                        l.failf(
                                LinkFailureError{code: ErrInternalError},
                                "unable to add preimages=%v to cache: %v",
                                l.uncommittedPreimages, err,
                        )
                        return
                }

                // Instead of truncating the slice to conserve memory
                // allocations, we simply set the uncommitted preimage slice to
                // nil so that a new one will be initialized if any more
                // witnesses are discovered. We do this because the maximum size
                // that the slice can occupy is 15KB, and we want to ensure we
                // release that memory back to the runtime.
                l.uncommittedPreimages = nil

                // We just received a new updates to our local commitment
                // chain, validate this new commitment, closing the link if
                // invalid.
                auxSigBlob, err := msg.CustomRecords.Serialize()
                if err != nil {
                        l.failf(
                                LinkFailureError{code: ErrInvalidCommitment},
                                "unable to serialize custom records: %v", err,
                        )

                        return
                }
                err = l.channel.ReceiveNewCommitment(&lnwallet.CommitSigs{
                        CommitSig:  msg.CommitSig,
                        HtlcSigs:   msg.HtlcSigs,
                        PartialSig: msg.PartialSig,
                        AuxSigBlob: auxSigBlob,
                })
                if err != nil {
                        // If we were unable to reconstruct their proposed
                        // commitment, then we'll examine the type of error. If
                        // it's an InvalidCommitSigError, then we'll send a
                        // direct error.
                        var sendData []byte
                        switch err.(type) {
                        case *lnwallet.InvalidCommitSigError:
                                sendData = []byte(err.Error())
                        case *lnwallet.InvalidHtlcSigError:
                                sendData = []byte(err.Error())
                        }
                        l.failf(
                                LinkFailureError{
                                        code:          ErrInvalidCommitment,
                                        FailureAction: LinkFailureForceClose,
                                        SendData:      sendData,
                                },
                                "ChannelPoint(%v): unable to accept new "+
                                        "commitment: %v",
                                l.channel.ChannelPoint(), err,
                        )
                        return
                }

                // As we've just accepted a new state, we'll now
                // immediately send the remote peer a revocation for our prior
                // state.
                nextRevocation, currentHtlcs, finalHTLCs, err :=
                        l.channel.RevokeCurrentCommitment()
                if err != nil {
                        l.log.Errorf("unable to revoke commitment: %v", err)

                        // We need to fail the channel in case revoking our
                        // local commitment does not succeed. We might have
                        // already advanced our channel state which would lead
                        // us to proceed with an unclean state.
                        //
                        // NOTE: We do not trigger a force close because this
                        // could resolve itself in case our db was just busy
                        // not accepting new transactions.
                        l.failf(
                                LinkFailureError{
                                        code:          ErrInternalError,
                                        Warning:       true,
                                        FailureAction: LinkFailureDisconnect,
                                },
                                "ChannelPoint(%v): unable to accept new "+
                                        "commitment: %v",
                                l.channel.ChannelPoint(), err,
                        )
                        return
                }

                // As soon as we are ready to send our next revocation, we can
                // invoke the incoming commit hooks.
                l.RWMutex.Lock()
                l.incomingCommitHooks.invoke()
                l.RWMutex.Unlock()

                l.cfg.Peer.SendMessage(false, nextRevocation)

                // Notify the incoming htlcs of which the resolutions were
                // locked in.
                for id, settled := range finalHTLCs {
                        l.cfg.HtlcNotifier.NotifyFinalHtlcEvent(
                                models.CircuitKey{
                                        ChanID: l.ShortChanID(),
                                        HtlcID: id,
                                },
                                channeldb.FinalHtlcInfo{
                                        Settled:  settled,
                                        Offchain: true,
                                },
                        )
                }

                // Since we just revoked our commitment, we may have a new set
                // of HTLC's on our commitment, so we'll send them using our
                // function closure NotifyContractUpdate.
                newUpdate := &contractcourt.ContractUpdate{
                        HtlcKey: contractcourt.LocalHtlcSet,
                        Htlcs:   currentHtlcs,
                }
                err = l.cfg.NotifyContractUpdate(newUpdate)
                if err != nil {
                        l.log.Errorf("unable to notify contract update: %v",
                                err)
                        return
                }

                select {
                case <-l.cg.Done():
                        return
                default:
                }

                // If the remote party initiated the state transition,
                // we'll reply with a signature to provide them with their
                // version of the latest commitment. Otherwise, both commitment
                // chains are fully synced from our PoV, then we don't need to
                // reply with a signature as both sides already have a
                // commitment with the latest accepted.
                if l.channel.OweCommitment() {
                        if !l.updateCommitTxOrFail(ctx) {
                                return
                        }
                }

                // If we need to send out an Stfu, this would be the time to do
                // so.
                if l.noDanglingUpdates(lntypes.Local) {
                        err = l.quiescer.SendOwedStfu()
                        if err != nil {
                                l.stfuFailf("sendOwedStfu: %v", err.Error())
                        }
                }

                // Now that we have finished processing the incoming CommitSig
                // and sent out our RevokeAndAck, we invoke the flushHooks if
                // the channel state is clean.
                l.RWMutex.Lock()
                if l.channel.IsChannelClean() {
                        l.flushHooks.invoke()
                }
                l.RWMutex.Unlock()

        case *lnwire.RevokeAndAck:
                // We've received a revocation from the remote chain, if valid,
                // this moves the remote chain forward, and expands our
                // revocation window.

                // We now process the message and advance our remote commit
                // chain.
                fwdPkg, remoteHTLCs, err := l.channel.ReceiveRevocation(msg)
                if err != nil {
                        // TODO(halseth): force close?
                        l.failf(
                                LinkFailureError{
                                        code:          ErrInvalidRevocation,
                                        FailureAction: LinkFailureDisconnect,
                                },
                                "unable to accept revocation: %v", err,
                        )
                        return
                }

                // The remote party now has a new primary commitment, so we'll
                // update the contract court to be aware of this new set (the
                // prior old remote pending).
                newUpdate := &contractcourt.ContractUpdate{
                        HtlcKey: contractcourt.RemoteHtlcSet,
                        Htlcs:   remoteHTLCs,
                }
                err = l.cfg.NotifyContractUpdate(newUpdate)
                if err != nil {
                        l.log.Errorf("unable to notify contract update: %v",
                                err)
                        return
                }

                select {
                case <-l.cg.Done():
                        return
                default:
                }

                // If we have a tower client for this channel type, we'll
                // create a backup for the current state.
                if l.cfg.TowerClient != nil {
                        state := l.channel.State()
                        chanID := l.ChanID()

                        err = l.cfg.TowerClient.BackupState(
                                &chanID, state.RemoteCommitment.CommitHeight-1,
                        )
                        if err != nil {
                                l.failf(LinkFailureError{
                                        code: ErrInternalError,
                                }, "unable to queue breach backup: %v", err)
                                return
                        }
                }

                // If we can send updates then we can process adds in case we
                // are the exit hop and need to send back resolutions, or in
                // case there are validity issues with the packets. Otherwise
                // we defer the action until resume.
                //
                // We are free to process the settles and fails without this
                // check since processing those can't result in further updates
                // to this channel link.
                if l.quiescer.CanSendUpdates() {
                        l.processRemoteAdds(fwdPkg)
                } else {
                        l.quiescer.OnResume(func() {
                                l.processRemoteAdds(fwdPkg)
                        })
                }
                l.processRemoteSettleFails(fwdPkg)

                // If the link failed during processing the adds, we must
                // return to ensure we won't attempted to update the state
                // further.
                if l.failed {
                        return
                }

                // The revocation window opened up. If there are pending local
                // updates, try to update the commit tx. Pending updates could
                // already have been present because of a previously failed
                // update to the commit tx or freshly added in by
                // processRemoteAdds. Also in case there are no local updates,
                // but there are still remote updates that are not in the remote
                // commit tx yet, send out an update.
                if l.channel.OweCommitment() {
                        if !l.updateCommitTxOrFail(ctx) {
                                return
                        }
                }

                // Now that we have finished processing the RevokeAndAck, we
                // can invoke the flushHooks if the channel state is clean.
                l.RWMutex.Lock()
                if l.channel.IsChannelClean() {
                        l.flushHooks.invoke()
                }
                l.RWMutex.Unlock()

        case *lnwire.UpdateFee:
                // Check and see if their proposed fee-rate would make us
                // exceed the fee threshold.
                fee := chainfee.SatPerKWeight(msg.FeePerKw)

                isDust, err := l.exceedsFeeExposureLimit(fee)
                if err != nil {
                        // This shouldn't typically happen. If it does, it
                        // indicates something is wrong with our channel state.
                        l.log.Errorf("Unable to determine if fee threshold " +
                                "exceeded")
                        l.failf(LinkFailureError{code: ErrInternalError},
                                "error calculating fee exposure: %v", err)

                        return
                }

                if isDust {
                        // The proposed fee-rate makes us exceed the fee
                        // threshold.
                        l.failf(LinkFailureError{code: ErrInternalError},
                                "fee threshold exceeded: %v", err)
                        return
                }

                // We received fee update from peer. If we are the initiator we
                // will fail the channel, if not we will apply the update.
                if err := l.channel.ReceiveUpdateFee(fee); err != nil {
                        l.failf(LinkFailureError{code: ErrInvalidUpdate},
                                "error receiving fee update: %v", err)
                        return
                }

                // Update the mailbox's feerate as well.
                l.mailBox.SetFeeRate(fee)

        case *lnwire.Stfu:
                err := l.handleStfu(msg)
                if err != nil {
                        l.stfuFailf("handleStfu: %v", err.Error())
                }

        // In the case where we receive a warning message from our peer, just
        // log it and move on. We choose not to disconnect from our peer,
        // although we "MAY" do so according to the specification.
        case *lnwire.Warning:
                l.log.Warnf("received warning message from peer: %v",
                        msg.Warning())

        case *lnwire.Error:
                // Error received from remote, MUST fail channel, but should
                // only print the contents of the error message if all
                // characters are printable ASCII.
                l.failf(
                        LinkFailureError{
                                code: ErrRemoteError,

                                // TODO(halseth): we currently don't fail the
                                // channel permanently, as there are some sync
                                // issues with other implementations that will
                                // lead to them sending an error message, but
                                // we can recover from on next connection. See
                                // https://github.com/ElementsProject/lightning/issues/4212
                                PermanentFailure: false,
                        },
                        "ChannelPoint(%v): received error from peer: %v",
                        l.channel.ChannelPoint(), msg.Error(),
                )
        default:
                l.log.Warnf("received unknown message of type %T", msg)
        }

}

// handleStfu implements the top-level logic for handling the Stfu message from
// our peer.
func (l *channelLink) handleStfu(stfu *lnwire.Stfu) error {
        if !l.noDanglingUpdates(lntypes.Remote) {
                return ErrPendingRemoteUpdates
        }
        err := l.quiescer.RecvStfu(*stfu)
        if err != nil {
                return err
        }

        // If we can immediately send an Stfu response back, we will.
        if l.noDanglingUpdates(lntypes.Local) {
                return l.quiescer.SendOwedStfu()
        }

        return nil
}

// stfuFailf fails the link in the case where the requirements of the quiescence
// protocol are violated. In all cases we opt to drop the connection as only
// link state (as opposed to channel state) is affected.
func (l *channelLink) stfuFailf(format string, args ...interface{}) {
        l.failf(LinkFailureError{
                code:             ErrStfuViolation,
                FailureAction:    LinkFailureDisconnect,
                PermanentFailure: false,
                Warning:          true,
        }, format, args...)
}

// noDanglingUpdates returns true when there are 0 updates that were originally
// issued by whose on either the Local or Remote commitment transaction.
func (l *channelLink) noDanglingUpdates(whose lntypes.ChannelParty) bool {
        pendingOnLocal := l.channel.NumPendingUpdates(
                whose, lntypes.Local,
        )
        pendingOnRemote := l.channel.NumPendingUpdates(
                whose, lntypes.Remote,
        )

        return pendingOnLocal == 0 && pendingOnRemote == 0
}

// ackDownStreamPackets is responsible for removing htlcs from a link's mailbox
// for packets delivered from server, and cleaning up any circuits closed by
// signing a previous commitment txn. This method ensures that the circuits are
// removed from the circuit map before removing them from the link's mailbox,
// otherwise it could be possible for some circuit to be missed if this link
// flaps.
func (l *channelLink) ackDownStreamPackets() error {
        // First, remove the downstream Add packets that were included in the
        // previous commitment signature. This will prevent the Adds from being
        // replayed if this link disconnects.
        for _, inKey := range l.openedCircuits {
                // In order to test the sphinx replay logic of the remote
                // party, unsafe replay does not acknowledge the packets from
                // the mailbox. We can then force a replay of any Add packets
                // held in memory by disconnecting and reconnecting the link.
                if l.cfg.UnsafeReplay {
                        continue
                }

                l.log.Debugf("removing Add packet %s from mailbox", inKey)
                l.mailBox.AckPacket(inKey)
        }

        // Now, we will delete all circuits closed by the previous commitment
        // signature, which is the result of downstream Settle/Fail packets. We
        // batch them here to ensure circuits are closed atomically and for
        // performance.
        err := l.cfg.Circuits.DeleteCircuits(l.closedCircuits...)
        switch err {
        case nil:
                // Successful deletion.

        default:
                l.log.Errorf("unable to delete %d circuits: %v",
                        len(l.closedCircuits), err)
                return err
        }

        // With the circuits removed from memory and disk, we now ack any
        // Settle/Fails in the mailbox to ensure they do not get redelivered
        // after startup. If forgive is enabled and we've reached this point,
        // the circuits must have been removed at some point, so it is now safe
        // to un-queue the corresponding Settle/Fails.
        for _, inKey := range l.closedCircuits {
                l.log.Debugf("removing Fail/Settle packet %s from mailbox",
                        inKey)
                l.mailBox.AckPacket(inKey)
        }

        // Lastly, reset our buffers to be empty while keeping any acquired
        // growth in the backing array.
        l.openedCircuits = l.openedCircuits[:0]
        l.closedCircuits = l.closedCircuits[:0]

        return nil
}

// updateCommitTxOrFail updates the commitment tx and if that fails, it fails
// the link.
func (l *channelLink) updateCommitTxOrFail(ctx context.Context) bool {
        err := l.updateCommitTx(ctx)
        switch err {
        // No error encountered, success.
        case nil:

        // A duplicate keystone error should be resolved and is not fatal, so
        // we won't send an Error message to the peer.
        case ErrDuplicateKeystone:
                l.failf(LinkFailureError{code: ErrCircuitError},
                        "temporary circuit error: %v", err)
                return false

        // Any other error is treated results in an Error message being sent to
        // the peer.
        default:
                l.failf(LinkFailureError{code: ErrInternalError},
                        "unable to update commitment: %v", err)
                return false
        }

        return true
}

// updateCommitTx signs, then sends an update to the remote peer adding a new
// commitment to their commitment chain which includes all the latest updates
// we've received+processed up to this point.
func (l *channelLink) updateCommitTx(ctx context.Context) error {
        // Preemptively write all pending keystones to disk, just in case the
        // HTLCs we have in memory are included in the subsequent attempt to
        // sign a commitment state.
        err := l.cfg.Circuits.OpenCircuits(l.keystoneBatch...)
        if err != nil {
                // If ErrDuplicateKeystone is returned, the caller will catch
                // it.
                return err
        }

        // Reset the batch, but keep the backing buffer to avoid reallocating.
        l.keystoneBatch = l.keystoneBatch[:0]

        // If hodl.Commit mode is active, we will refrain from attempting to
        // commit any in-memory modifications to the channel state. Exiting here
        // permits testing of either the switch or link's ability to trim
        // circuits that have been opened, but unsuccessfully committed.
        if l.cfg.HodlMask.Active(hodl.Commit) {
                l.log.Warnf(hodl.Commit.Warning())
                return nil
        }

        ctx, done := l.cg.Create(ctx)
        defer done()

        newCommit, err := l.channel.SignNextCommitment(ctx)
        if err == lnwallet.ErrNoWindow {
                l.cfg.PendingCommitTicker.Resume()
                l.log.Trace("PendingCommitTicker resumed")

                n := l.channel.NumPendingUpdates(lntypes.Local, lntypes.Remote)
                l.log.Tracef("revocation window exhausted, unable to send: "+
                        "%v, pend_updates=%v, dangling_closes%v", n,
                        lnutils.SpewLogClosure(l.openedCircuits),
                        lnutils.SpewLogClosure(l.closedCircuits))

                return nil
        } else if err != nil {
                return err
        }

        if err := l.ackDownStreamPackets(); err != nil {
                return err
        }

        l.cfg.PendingCommitTicker.Pause()
        l.log.Trace("PendingCommitTicker paused after ackDownStreamPackets")

        // The remote party now has a new pending commitment, so we'll update
        // the contract court to be aware of this new set (the prior old remote
        // pending).
        newUpdate := &contractcourt.ContractUpdate{
                HtlcKey: contractcourt.RemotePendingHtlcSet,
                Htlcs:   newCommit.PendingHTLCs,
        }
        err = l.cfg.NotifyContractUpdate(newUpdate)
        if err != nil {
                l.log.Errorf("unable to notify contract update: %v", err)
                return err
        }

        select {
        case <-l.cg.Done():
                return ErrLinkShuttingDown
        default:
        }

        auxBlobRecords, err := lnwire.ParseCustomRecords(newCommit.AuxSigBlob)
        if err != nil {
                return fmt.Errorf("error parsing aux sigs: %w", err)
        }

        commitSig := &lnwire.CommitSig{
                ChanID:        l.ChanID(),
                CommitSig:     newCommit.CommitSig,
                HtlcSigs:      newCommit.HtlcSigs,
                PartialSig:    newCommit.PartialSig,
                CustomRecords: auxBlobRecords,
        }
        l.cfg.Peer.SendMessage(false, commitSig)

        // Now that we have sent out a new CommitSig, we invoke the outgoing set
        // of commit hooks.
        l.RWMutex.Lock()
        l.outgoingCommitHooks.invoke()
        l.RWMutex.Unlock()

        return nil
}

// Peer returns the representation of remote peer with which we have the
// channel link opened.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) PeerPubKey() [33]byte {
        return l.cfg.Peer.PubKey()
}

// ChannelPoint returns the channel outpoint for the channel link.
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) ChannelPoint() wire.OutPoint {
        return l.channel.ChannelPoint()
}

// ShortChanID returns the short channel ID for the channel link. The short
// channel ID encodes the exact location in the main chain that the original
// funding output can be found.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) ShortChanID() lnwire.ShortChannelID {
        l.RLock()
        defer l.RUnlock()

        return l.channel.ShortChanID()
}

// UpdateShortChanID updates the short channel ID for a link. This may be
// required in the event that a link is created before the short chan ID for it
// is known, or a re-org occurs, and the funding transaction changes location
// within the chain.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) UpdateShortChanID() (lnwire.ShortChannelID, error) {
        chanID := l.ChanID()

        // Refresh the channel state's short channel ID by loading it from disk.
        // This ensures that the channel state accurately reflects the updated
        // short channel ID.
        err := l.channel.State().Refresh()
        if err != nil {
                l.log.Errorf("unable to refresh short_chan_id for chan_id=%v: "+
                        "%v", chanID, err)
                return hop.Source, err
        }

        return hop.Source, nil
}

// ChanID returns the channel ID for the channel link. The channel ID is a more
// compact representation of a channel's full outpoint.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) ChanID() lnwire.ChannelID {
        return lnwire.NewChanIDFromOutPoint(l.channel.ChannelPoint())
}

// Bandwidth returns the total amount that can flow through the channel link at
// this given instance. The value returned is expressed in millisatoshi and can
// be used by callers when making forwarding decisions to determine if a link
// can accept an HTLC.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) Bandwidth() lnwire.MilliSatoshi {
        // Get the balance available on the channel for new HTLCs. This takes
        // the channel reserve into account so HTLCs up to this value won't
        // violate it.
        return l.channel.AvailableBalance()
}

// MayAddOutgoingHtlc indicates whether we can add an outgoing htlc with the
// amount provided to the link. This check does not reserve a space, since
// forwards or other payments may use the available slot, so it should be
// considered best-effort.
func (l *channelLink) MayAddOutgoingHtlc(amt lnwire.MilliSatoshi) error {
        return l.channel.MayAddOutgoingHtlc(amt)
}

// getDustSum is a wrapper method that calls the underlying channel's dust sum
// method.
//
// NOTE: Part of the dustHandler interface.
func (l *channelLink) getDustSum(whoseCommit lntypes.ChannelParty,
        dryRunFee fn.Option[chainfee.SatPerKWeight]) lnwire.MilliSatoshi {

        return l.channel.GetDustSum(whoseCommit, dryRunFee)
}

// getFeeRate is a wrapper method that retrieves the underlying channel's
// feerate.
//
// NOTE: Part of the dustHandler interface.
func (l *channelLink) getFeeRate() chainfee.SatPerKWeight {
        return l.channel.CommitFeeRate()
}

// getDustClosure returns a closure that can be used by the switch or mailbox
// to evaluate whether a given HTLC is dust.
//
// NOTE: Part of the dustHandler interface.
func (l *channelLink) getDustClosure() dustClosure {
        localDustLimit := l.channel.State().LocalChanCfg.DustLimit
        remoteDustLimit := l.channel.State().RemoteChanCfg.DustLimit
        chanType := l.channel.State().ChanType

        return dustHelper(chanType, localDustLimit, remoteDustLimit)
}

// getCommitFee returns either the local or remote CommitFee in satoshis. This
// is used so that the Switch can have access to the commitment fee without
// needing to have a *LightningChannel. This doesn't include dust.
//
// NOTE: Part of the dustHandler interface.
func (l *channelLink) getCommitFee(remote bool) btcutil.Amount {
        if remote {
                return l.channel.State().RemoteCommitment.CommitFee
        }

        return l.channel.State().LocalCommitment.CommitFee
}

// exceedsFeeExposureLimit returns whether or not the new proposed fee-rate
// increases the total dust and fees within the channel past the configured
// fee threshold. It first calculates the dust sum over every update in the
// update log with the proposed fee-rate and taking into account both the local
// and remote dust limits. It uses every update in the update log instead of
// what is actually on the local and remote commitments because it is assumed
// that in a worst-case scenario, every update in the update log could
// theoretically be on either commitment transaction and this needs to be
// accounted for with this fee-rate. It then calculates the local and remote
// commitment fees given the proposed fee-rate. Finally, it tallies the results
// and determines if the fee threshold has been exceeded.
func (l *channelLink) exceedsFeeExposureLimit(
        feePerKw chainfee.SatPerKWeight) (bool, error) {

        dryRunFee := fn.Some[chainfee.SatPerKWeight](feePerKw)

        // Get the sum of dust for both the local and remote commitments using
        // this "dry-run" fee.
        localDustSum := l.getDustSum(lntypes.Local, dryRunFee)
        remoteDustSum := l.getDustSum(lntypes.Remote, dryRunFee)

        // Calculate the local and remote commitment fees using this dry-run
        // fee.
        localFee, remoteFee, err := l.channel.CommitFeeTotalAt(feePerKw)
        if err != nil {
                return false, err
        }

        // Finally, check whether the max fee exposure was exceeded on either
        // future commitment transaction with the fee-rate.
        totalLocalDust := localDustSum + lnwire.NewMSatFromSatoshis(localFee)
        if totalLocalDust > l.cfg.MaxFeeExposure {
                l.log.Debugf("ChannelLink(%v): exceeds fee exposure limit: "+
                        "local dust: %v, local fee: %v", l.ShortChanID(),
                        totalLocalDust, localFee)

                return true, nil
        }

        totalRemoteDust := remoteDustSum + lnwire.NewMSatFromSatoshis(
                remoteFee,
        )

        if totalRemoteDust > l.cfg.MaxFeeExposure {
                l.log.Debugf("ChannelLink(%v): exceeds fee exposure limit: "+
                        "remote dust: %v, remote fee: %v", l.ShortChanID(),
                        totalRemoteDust, remoteFee)

                return true, nil
        }

        return false, nil
}

// isOverexposedWithHtlc calculates whether the proposed HTLC will make the
// channel exceed the fee threshold. It first fetches the largest fee-rate that
// may be on any unrevoked commitment transaction. Then, using this fee-rate,
// determines if the to-be-added HTLC is dust. If the HTLC is dust, it adds to
// the overall dust sum. If it is not dust, it contributes to weight, which
// also adds to the overall dust sum by an increase in fees. If the dust sum on
// either commitment exceeds the configured fee threshold, this function
// returns true.
func (l *channelLink) isOverexposedWithHtlc(htlc *lnwire.UpdateAddHTLC,
        incoming bool) bool {

        dustClosure := l.getDustClosure()

        feeRate := l.channel.WorstCaseFeeRate()

        amount := htlc.Amount.ToSatoshis()

        // See if this HTLC is dust on both the local and remote commitments.
        isLocalDust := dustClosure(feeRate, incoming, lntypes.Local, amount)
        isRemoteDust := dustClosure(feeRate, incoming, lntypes.Remote, amount)

        // Calculate the dust sum for the local and remote commitments.
        localDustSum := l.getDustSum(
                lntypes.Local, fn.None[chainfee.SatPerKWeight](),
        )
        remoteDustSum := l.getDustSum(
                lntypes.Remote, fn.None[chainfee.SatPerKWeight](),
        )

        // Grab the larger of the local and remote commitment fees w/o dust.
        commitFee := l.getCommitFee(false)

        if l.getCommitFee(true) > commitFee {
                commitFee = l.getCommitFee(true)
        }

        commitFeeMSat := lnwire.NewMSatFromSatoshis(commitFee)

        localDustSum += commitFeeMSat
        remoteDustSum += commitFeeMSat

        // Calculate the additional fee increase if this is a non-dust HTLC.
        weight := lntypes.WeightUnit(input.HTLCWeight)
        additional := lnwire.NewMSatFromSatoshis(
                feeRate.FeeForWeight(weight),
        )

        if isLocalDust {
                // If this is dust, it doesn't contribute to weight but does
                // contribute to the overall dust sum.
                localDustSum += lnwire.NewMSatFromSatoshis(amount)
        } else {
                // Account for the fee increase that comes with an increase in
                // weight.
                localDustSum += additional
        }

        if localDustSum > l.cfg.MaxFeeExposure {
                // The max fee exposure was exceeded.
                l.log.Debugf("ChannelLink(%v): HTLC %v makes the channel "+
                        "overexposed, total local dust: %v (current commit "+
                        "fee: %v)", l.ShortChanID(), htlc, localDustSum)

                return true
        }

        if isRemoteDust {
                // If this is dust, it doesn't contribute to weight but does
                // contribute to the overall dust sum.
                remoteDustSum += lnwire.NewMSatFromSatoshis(amount)
        } else {
                // Account for the fee increase that comes with an increase in
                // weight.
                remoteDustSum += additional
        }

        if remoteDustSum > l.cfg.MaxFeeExposure {
                // The max fee exposure was exceeded.
                l.log.Debugf("ChannelLink(%v): HTLC %v makes the channel "+
                        "overexposed, total remote dust: %v (current commit "+
                        "fee: %v)", l.ShortChanID(), htlc, remoteDustSum)

                return true
        }

        return false
}

// dustClosure is a function that evaluates whether an HTLC is dust. It returns
// true if the HTLC is dust. It takes in a feerate, a boolean denoting whether
// the HTLC is incoming (i.e. one that the remote sent), a boolean denoting
// whether to evaluate on the local or remote commit, and finally an HTLC
// amount to test.
type dustClosure func(feerate chainfee.SatPerKWeight, incoming bool,
        whoseCommit lntypes.ChannelParty, amt btcutil.Amount) bool

// dustHelper is used to construct the dustClosure.
func dustHelper(chantype channeldb.ChannelType, localDustLimit,
        remoteDustLimit btcutil.Amount) dustClosure {

        isDust := func(feerate chainfee.SatPerKWeight, incoming bool,
                whoseCommit lntypes.ChannelParty, amt btcutil.Amount) bool {

                var dustLimit btcutil.Amount
                if whoseCommit.IsLocal() {
                        dustLimit = localDustLimit
                } else {
                        dustLimit = remoteDustLimit
                }

                return lnwallet.HtlcIsDust(
                        chantype, incoming, whoseCommit, feerate, amt,
                        dustLimit,
                )
        }

        return isDust
}

// zeroConfConfirmed returns whether or not the zero-conf channel has
// confirmed on-chain.
//
// Part of the scidAliasHandler interface.
func (l *channelLink) zeroConfConfirmed() bool {
        return l.channel.State().ZeroConfConfirmed()
}

// confirmedScid returns the confirmed SCID for a zero-conf channel. This
// should not be called for non-zero-conf channels.
//
// Part of the scidAliasHandler interface.
func (l *channelLink) confirmedScid() lnwire.ShortChannelID {
        return l.channel.State().ZeroConfRealScid()
}

// isZeroConf returns whether or not the underlying channel is a zero-conf
// channel.
//
// Part of the scidAliasHandler interface.
func (l *channelLink) isZeroConf() bool {
        return l.channel.State().IsZeroConf()
}

// negotiatedAliasFeature returns whether or not the underlying channel has
// negotiated the option-scid-alias feature bit. This will be true for both
// option-scid-alias and zero-conf channel-types. It will also be true for
// channels with the feature bit but without the above channel-types.
//
// Part of the scidAliasFeature interface.
func (l *channelLink) negotiatedAliasFeature() bool {
        return l.channel.State().NegotiatedAliasFeature()
}

// getAliases returns the set of aliases for the underlying channel.
//
// Part of the scidAliasHandler interface.
func (l *channelLink) getAliases() []lnwire.ShortChannelID {
        return l.cfg.GetAliases(l.ShortChanID())
}

// attachFailAliasUpdate sets the link's FailAliasUpdate function.
//
// Part of the scidAliasHandler interface.
func (l *channelLink) attachFailAliasUpdate(closure func(
        sid lnwire.ShortChannelID, incoming bool) *lnwire.ChannelUpdate1) {

        l.Lock()
        l.cfg.FailAliasUpdate = closure
        l.Unlock()
}

// AttachMailBox updates the current mailbox used by this link, and hooks up
// the mailbox's message and packet outboxes to the link's upstream and
// downstream chans, respectively.
func (l *channelLink) AttachMailBox(mailbox MailBox) {
        l.Lock()
        l.mailBox = mailbox
        l.upstream = mailbox.MessageOutBox()
        l.downstream = mailbox.PacketOutBox()
        l.Unlock()

        // Set the mailbox's fee rate. This may be refreshing a feerate that was
        // never committed.
        l.mailBox.SetFeeRate(l.getFeeRate())

        // Also set the mailbox's dust closure so that it can query whether HTLC's
        // are dust given the current feerate.
        l.mailBox.SetDustClosure(l.getDustClosure())
}

// UpdateForwardingPolicy updates the forwarding policy for the target
// ChannelLink. Once updated, the link will use the new forwarding policy to
// govern if it an incoming HTLC should be forwarded or not. We assume that
// fields that are zero are intentionally set to zero, so we'll use newPolicy to
// update all of the link's FwrdingPolicy's values.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) UpdateForwardingPolicy(
        newPolicy models.ForwardingPolicy) {

        l.Lock()
        defer l.Unlock()

        l.cfg.FwrdingPolicy = newPolicy
}

// CheckHtlcForward should return a nil error if the passed HTLC details
// satisfy the current forwarding policy fo the target link. Otherwise,
// a LinkError with a valid protocol failure message should be returned
// in order to signal to the source of the HTLC, the policy consistency
// issue.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) CheckHtlcForward(payHash [32]byte, incomingHtlcAmt,
        amtToForward lnwire.MilliSatoshi, incomingTimeout,
        outgoingTimeout uint32, inboundFee models.InboundFee,
        heightNow uint32, originalScid lnwire.ShortChannelID,
        customRecords lnwire.CustomRecords) *LinkError {

        l.RLock()
        policy := l.cfg.FwrdingPolicy
        l.RUnlock()

        // Using the outgoing HTLC amount, we'll calculate the outgoing
        // fee this incoming HTLC must carry in order to satisfy the constraints
        // of the outgoing link.
        outFee := ExpectedFee(policy, amtToForward)

        // Then calculate the inbound fee that we charge based on the sum of
        // outgoing HTLC amount and outgoing fee.
        inFee := inboundFee.CalcFee(amtToForward + outFee)

        // Add up both fee components. It is important to calculate both fees
        // separately. An alternative way of calculating is to first determine
        // an aggregate fee and apply that to the outgoing HTLC amount. However,
        // rounding may cause the result to be slightly higher than in the case
        // of separately rounded fee components. This potentially causes failed
        // forwards for senders and is something to be avoided.
        expectedFee := inFee + int64(outFee)

        // If the actual fee is less than our expected fee, then we'll reject
        // this HTLC as it didn't provide a sufficient amount of fees, or the
        // values have been tampered with, or the send used incorrect/dated
        // information to construct the forwarding information for this hop. In
        // any case, we'll cancel this HTLC.
        actualFee := int64(incomingHtlcAmt) - int64(amtToForward)
        if incomingHtlcAmt < amtToForward || actualFee < expectedFee {
                l.log.Warnf("outgoing htlc(%x) has insufficient fee: "+
                        "expected %v, got %v: incoming=%v, outgoing=%v, "+
                        "inboundFee=%v",
                        payHash[:], expectedFee, actualFee,
                        incomingHtlcAmt, amtToForward, inboundFee,
                )

                // As part of the returned error, we'll send our latest routing
                // policy so the sending node obtains the most up to date data.
                cb := func(upd *lnwire.ChannelUpdate1) lnwire.FailureMessage {
                        return lnwire.NewFeeInsufficient(amtToForward, *upd)
                }
                failure := l.createFailureWithUpdate(false, originalScid, cb)
                return NewLinkError(failure)
        }

        // Check whether the outgoing htlc satisfies the channel policy.
        err := l.canSendHtlc(
                policy, payHash, amtToForward, outgoingTimeout, heightNow,
                originalScid, customRecords,
        )
        if err != nil {
                return err
        }

        // Finally, we'll ensure that the time-lock on the outgoing HTLC meets
        // the following constraint: the incoming time-lock minus our time-lock
        // delta should equal the outgoing time lock. Otherwise, whether the
        // sender messed up, or an intermediate node tampered with the HTLC.
        timeDelta := policy.TimeLockDelta
        if incomingTimeout < outgoingTimeout+timeDelta {
                l.log.Warnf("incoming htlc(%x) has incorrect time-lock value: "+
                        "expected at least %v block delta, got %v block delta",
                        payHash[:], timeDelta, incomingTimeout-outgoingTimeout)

                // Grab the latest routing policy so the sending node is up to
                // date with our current policy.
                cb := func(upd *lnwire.ChannelUpdate1) lnwire.FailureMessage {
                        return lnwire.NewIncorrectCltvExpiry(
                                incomingTimeout, *upd,
                        )
                }
                failure := l.createFailureWithUpdate(false, originalScid, cb)
                return NewLinkError(failure)
        }

        return nil
}

// CheckHtlcTransit should return a nil error if the passed HTLC details
// satisfy the current channel policy.  Otherwise, a LinkError with a
// valid protocol failure message should be returned in order to signal
// the violation. This call is intended to be used for locally initiated
// payments for which there is no corresponding incoming htlc.
func (l *channelLink) CheckHtlcTransit(payHash [32]byte,
        amt lnwire.MilliSatoshi, timeout uint32, heightNow uint32,
        customRecords lnwire.CustomRecords) *LinkError {

        l.RLock()
        policy := l.cfg.FwrdingPolicy
        l.RUnlock()

        // We pass in hop.Source here as this is only used in the Switch when
        // trying to send over a local link. This causes the fallback mechanism
        // to occur.
        return l.canSendHtlc(
                policy, payHash, amt, timeout, heightNow, hop.Source,
                customRecords,
        )
}

// canSendHtlc checks whether the given htlc parameters satisfy
// the channel's amount and time lock constraints.
func (l *channelLink) canSendHtlc(policy models.ForwardingPolicy,
        payHash [32]byte, amt lnwire.MilliSatoshi, timeout uint32,
        heightNow uint32, originalScid lnwire.ShortChannelID,
        customRecords lnwire.CustomRecords) *LinkError {

        // As our first sanity check, we'll ensure that the passed HTLC isn't
        // too small for the next hop. If so, then we'll cancel the HTLC
        // directly.
        if amt < policy.MinHTLCOut {
                l.log.Warnf("outgoing htlc(%x) is too small: min_htlc=%v, "+
                        "htlc_value=%v", payHash[:], policy.MinHTLCOut,
                        amt)

                // As part of the returned error, we'll send our latest routing
                // policy so the sending node obtains the most up to date data.
                cb := func(upd *lnwire.ChannelUpdate1) lnwire.FailureMessage {
                        return lnwire.NewAmountBelowMinimum(amt, *upd)
                }
                failure := l.createFailureWithUpdate(false, originalScid, cb)
                return NewLinkError(failure)
        }

        // Next, ensure that the passed HTLC isn't too large. If so, we'll
        // cancel the HTLC directly.
        if policy.MaxHTLC != 0 && amt > policy.MaxHTLC {
                l.log.Warnf("outgoing htlc(%x) is too large: max_htlc=%v, "+
                        "htlc_value=%v", payHash[:], policy.MaxHTLC, amt)

                // As part of the returned error, we'll send our latest routing
                // policy so the sending node obtains the most up-to-date data.
                cb := func(upd *lnwire.ChannelUpdate1) lnwire.FailureMessage {
                        return lnwire.NewTemporaryChannelFailure(upd)
                }
                failure := l.createFailureWithUpdate(false, originalScid, cb)
                return NewDetailedLinkError(failure, OutgoingFailureHTLCExceedsMax)
        }

        // We want to avoid offering an HTLC which will expire in the near
        // future, so we'll reject an HTLC if the outgoing expiration time is
        // too close to the current height.
        if timeout <= heightNow+l.cfg.OutgoingCltvRejectDelta {
                l.log.Warnf("htlc(%x) has an expiry that's too soon: "+
                        "outgoing_expiry=%v, best_height=%v", payHash[:],
                        timeout, heightNow)

                cb := func(upd *lnwire.ChannelUpdate1) lnwire.FailureMessage {
                        return lnwire.NewExpiryTooSoon(*upd)
                }
                failure := l.createFailureWithUpdate(false, originalScid, cb)
                return NewLinkError(failure)
        }

        // Check absolute max delta.
        if timeout > l.cfg.MaxOutgoingCltvExpiry+heightNow {
                l.log.Warnf("outgoing htlc(%x) has a time lock too far in "+
                        "the future: got %v, but maximum is %v", payHash[:],
                        timeout-heightNow, l.cfg.MaxOutgoingCltvExpiry)

                return NewLinkError(&lnwire.FailExpiryTooFar{})
        }

        // We now check the available bandwidth to see if this HTLC can be
        // forwarded.
        availableBandwidth := l.Bandwidth()
        auxBandwidth, err := fn.MapOptionZ(
                l.cfg.AuxTrafficShaper,
                func(ts AuxTrafficShaper) fn.Result[OptionalBandwidth] {
                        var htlcBlob fn.Option[tlv.Blob]
                        blob, err := customRecords.Serialize()
                        if err != nil {
                                return fn.Err[OptionalBandwidth](
                                        fmt.Errorf("unable to serialize "+
                                                "custom records: %w", err))
                        }

                        if len(blob) > 0 {
                                htlcBlob = fn.Some(blob)
                        }

                        return l.AuxBandwidth(amt, originalScid, htlcBlob, ts)
                },
        ).Unpack()
        if err != nil {
                l.log.Errorf("Unable to determine aux bandwidth: %v", err)
                return NewLinkError(&lnwire.FailTemporaryNodeFailure{})
        }

        auxBandwidth.WhenSome(func(bandwidth lnwire.MilliSatoshi) {
                availableBandwidth = bandwidth
        })

        // Check to see if there is enough balance in this channel.
        if amt > availableBandwidth {
                l.log.Warnf("insufficient bandwidth to route htlc: %v is "+
                        "larger than %v", amt, availableBandwidth)
                cb := func(upd *lnwire.ChannelUpdate1) lnwire.FailureMessage {
                        return lnwire.NewTemporaryChannelFailure(upd)
                }
                failure := l.createFailureWithUpdate(false, originalScid, cb)
                return NewDetailedLinkError(
                        failure, OutgoingFailureInsufficientBalance,
                )
        }

        return nil
}

// AuxBandwidth returns the bandwidth that can be used for a channel, expressed
// in milli-satoshi. This might be different from the regular BTC bandwidth for
// custom channels. This will always return fn.None() for a regular (non-custom)
// channel.
func (l *channelLink) AuxBandwidth(amount lnwire.MilliSatoshi,
        cid lnwire.ShortChannelID, htlcBlob fn.Option[tlv.Blob],
        ts AuxTrafficShaper) fn.Result[OptionalBandwidth] {

        unknownBandwidth := fn.None[lnwire.MilliSatoshi]()

        fundingBlob := l.FundingCustomBlob()
        shouldHandle, err := ts.ShouldHandleTraffic(cid, fundingBlob)
        if err != nil {
                return fn.Err[OptionalBandwidth](fmt.Errorf("traffic shaper "+
                        "failed to decide whether to handle traffic: %w", err))
        }

        log.Debugf("ShortChannelID=%v: aux traffic shaper is handling "+
                "traffic: %v", cid, shouldHandle)

        // If this channel isn't handled by the aux traffic shaper, we'll return
        // early.
        if !shouldHandle {
                return fn.Ok(unknownBandwidth)
        }

        // Ask for a specific bandwidth to be used for the channel.
        commitmentBlob := l.CommitmentCustomBlob()
        auxBandwidth, err := ts.PaymentBandwidth(
                htlcBlob, commitmentBlob, l.Bandwidth(), amount,
        )
        if err != nil {
                return fn.Err[OptionalBandwidth](fmt.Errorf("failed to get "+
                        "bandwidth from external traffic shaper: %w", err))
        }

        log.Debugf("ShortChannelID=%v: aux traffic shaper reported available "+
                "bandwidth: %v", cid, auxBandwidth)

        return fn.Ok(fn.Some(auxBandwidth))
}

// Stats returns the statistics of channel link.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) Stats() (uint64, lnwire.MilliSatoshi, lnwire.MilliSatoshi) {
        snapshot := l.channel.StateSnapshot()

        return snapshot.ChannelCommitment.CommitHeight,
                snapshot.TotalMSatSent,
                snapshot.TotalMSatReceived
}

// String returns the string representation of channel link.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) String() string {
        return l.channel.ChannelPoint().String()
}

// handleSwitchPacket handles the switch packets. This packets which might be
// forwarded to us from another channel link in case the htlc update came from
// another peer or if the update was created by user
//
// NOTE: Part of the packetHandler interface.
func (l *channelLink) handleSwitchPacket(pkt *htlcPacket) error {
        l.log.Tracef("received switch packet inkey=%v, outkey=%v",
                pkt.inKey(), pkt.outKey())

        return l.mailBox.AddPacket(pkt)
}

// HandleChannelUpdate handles the htlc requests as settle/add/fail which sent
// to us from remote peer we have a channel with.
//
// NOTE: Part of the ChannelLink interface.
func (l *channelLink) HandleChannelUpdate(message lnwire.Message) {
        select {
        case <-l.cg.Done():
                // Return early if the link is already in the process of
                // quitting. It doesn't make sense to hand the message to the
                // mailbox here.
                return
        default:
        }

        err := l.mailBox.AddMessage(message)
        if err != nil {
                l.log.Errorf("failed to add Message to mailbox: %v", err)
        }
}

// updateChannelFee updates the commitment fee-per-kw on this channel by
// committing to an update_fee message.
func (l *channelLink) updateChannelFee(ctx context.Context,
        feePerKw chainfee.SatPerKWeight) error {

        l.log.Infof("updating commit fee to %v", feePerKw)

        // We skip sending the UpdateFee message if the channel is not
        // currently eligible to forward messages.
        if !l.eligibleToUpdate() {
                l.log.Debugf("skipping fee update for inactive channel")
                return nil
        }

        // Check and see if our proposed fee-rate would make us exceed the fee
        // threshold.
        thresholdExceeded, err := l.exceedsFeeExposureLimit(feePerKw)
        if err != nil {
                // This shouldn't typically happen. If it does, it indicates
                // something is wrong with our channel state.
                return err
        }

        if thresholdExceeded {
                return fmt.Errorf("link fee threshold exceeded")
        }

        // First, we'll update the local fee on our commitment.
        if err := l.channel.UpdateFee(feePerKw); err != nil {
                return err
        }

        // The fee passed the channel's validation checks, so we update the
        // mailbox feerate.
        l.mailBox.SetFeeRate(feePerKw)

        // We'll then attempt to send a new UpdateFee message, and also lock it
        // in immediately by triggering a commitment update.
        msg := lnwire.NewUpdateFee(l.ChanID(), uint32(feePerKw))
        if err := l.cfg.Peer.SendMessage(false, msg); err != nil {
                return err
        }

        return l.updateCommitTx(ctx)
}

// processRemoteSettleFails accepts a batch of settle/fail payment descriptors
// after receiving a revocation from the remote party, and reprocesses them in
// the context of the provided forwarding package. Any settles or fails that
// have already been acknowledged in the forwarding package will not be sent to
// the switch.
func (l *channelLink) processRemoteSettleFails(fwdPkg *channeldb.FwdPkg) {
        if len(fwdPkg.SettleFails) == 0 {
                return
        }

        l.log.Debugf("settle-fail-filter: %v", fwdPkg.SettleFailFilter)

        var switchPackets []*htlcPacket
        for i, update := range fwdPkg.SettleFails {
                destRef := fwdPkg.DestRef(uint16(i))

                // Skip any settles or fails that have already been
                // acknowledged by the incoming link that originated the
                // forwarded Add.
                if fwdPkg.SettleFailFilter.Contains(uint16(i)) {
                        continue
                }

                // TODO(roasbeef): rework log entries to a shared
                // interface.

                switch msg := update.UpdateMsg.(type) {
                // A settle for an HTLC we previously forwarded HTLC has been
                // received. So we'll forward the HTLC to the switch which will
                // handle propagating the settle to the prior hop.
                case *lnwire.UpdateFulfillHTLC:
                        // If hodl.SettleIncoming is requested, we will not
                        // forward the SETTLE to the switch and will not signal
                        // a free slot on the commitment transaction.
                        if l.cfg.HodlMask.Active(hodl.SettleIncoming) {
                                l.log.Warnf(hodl.SettleIncoming.Warning())
                                continue
                        }

                        settlePacket := &htlcPacket{
                                outgoingChanID: l.ShortChanID(),
                                outgoingHTLCID: msg.ID,
                                destRef:        &destRef,
                                htlc:           msg,
                        }

                        // Add the packet to the batch to be forwarded, and
                        // notify the overflow queue that a spare spot has been
                        // freed up within the commitment state.
                        switchPackets = append(switchPackets, settlePacket)

                // A failureCode message for a previously forwarded HTLC has
                // been received. As a result a new slot will be freed up in
                // our commitment state, so we'll forward this to the switch so
                // the backwards undo can continue.
                case *lnwire.UpdateFailHTLC:
                        // If hodl.SettleIncoming is requested, we will not
                        // forward the FAIL to the switch and will not signal a
                        // free slot on the commitment transaction.
                        if l.cfg.HodlMask.Active(hodl.FailIncoming) {
                                l.log.Warnf(hodl.FailIncoming.Warning())
                                continue
                        }

                        // Fetch the reason the HTLC was canceled so we can
                        // continue to propagate it. This failure originated
                        // from another node, so the linkFailure field is not
                        // set on the packet.
                        failPacket := &htlcPacket{
                                outgoingChanID: l.ShortChanID(),
                                outgoingHTLCID: msg.ID,
                                destRef:        &destRef,
                                htlc:           msg,
                        }

                        l.log.Debugf("Failed to send HTLC with ID=%d", msg.ID)

                        // If the failure message lacks an HMAC (but includes
                        // the 4 bytes for encoding the message and padding
                        // lengths, then this means that we received it as an
                        // UpdateFailMalformedHTLC. As a result, we'll signal
                        // that we need to convert this error within the switch
                        // to an actual error, by encrypting it as if we were
                        // the originating hop.
                        convertedErrorSize := lnwire.FailureMessageLength + 4
                        if len(msg.Reason) == convertedErrorSize {
                                failPacket.convertedError = true
                        }

                        // Add the packet to the batch to be forwarded, and
                        // notify the overflow queue that a spare spot has been
                        // freed up within the commitment state.
                        switchPackets = append(switchPackets, failPacket)
                }
        }

        // Only spawn the task forward packets we have a non-zero number.
        if len(switchPackets) > 0 {
                go l.forwardBatch(false, switchPackets...)
        }
}

// processRemoteAdds serially processes each of the Add payment descriptors
// which have been "locked-in" by receiving a revocation from the remote party.
// The forwarding package provided instructs how to process this batch,
// indicating whether this is the first time these Adds are being processed, or
// whether we are reprocessing as a result of a failure or restart. Adds that
// have already been acknowledged in the forwarding package will be ignored.
//
//nolint:funlen
func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg) {
        l.log.Tracef("processing %d remote adds for height %d",
                len(fwdPkg.Adds), fwdPkg.Height)

        decodeReqs := make(
                []hop.DecodeHopIteratorRequest, 0, len(fwdPkg.Adds),
        )
        for _, update := range fwdPkg.Adds {
                if msg, ok := update.UpdateMsg.(*lnwire.UpdateAddHTLC); ok {
                        // Before adding the new htlc to the state machine,
                        // parse the onion object in order to obtain the
                        // routing information with DecodeHopIterator function
                        // which process the Sphinx packet.
                        onionReader := bytes.NewReader(msg.OnionBlob[:])

                        req := hop.DecodeHopIteratorRequest{
                                OnionReader:    onionReader,
                                RHash:          msg.PaymentHash[:],
                                IncomingCltv:   msg.Expiry,
                                IncomingAmount: msg.Amount,
                                BlindingPoint:  msg.BlindingPoint,
                        }

                        decodeReqs = append(decodeReqs, req)
                }
        }

        // Atomically decode the incoming htlcs, simultaneously checking for
        // replay attempts. A particular index in the returned, spare list of
        // channel iterators should only be used if the failure code at the
        // same index is lnwire.FailCodeNone.
        decodeResps, sphinxErr := l.cfg.DecodeHopIterators(
                fwdPkg.ID(), decodeReqs,
        )
        if sphinxErr != nil {
                l.failf(LinkFailureError{code: ErrInternalError},
                        "unable to decode hop iterators: %v", sphinxErr)
                return
        }

        var switchPackets []*htlcPacket

        for i, update := range fwdPkg.Adds {
                idx := uint16(i)

                //nolint:forcetypeassert
                add := *update.UpdateMsg.(*lnwire.UpdateAddHTLC)
                sourceRef := fwdPkg.SourceRef(idx)

                if fwdPkg.State == channeldb.FwdStateProcessed &&
                        fwdPkg.AckFilter.Contains(idx) {

                        // If this index is already found in the ack filter,
                        // the response to this forwarding decision has already
                        // been committed by one of our commitment txns. ADDs
                        // in this state are waiting for the rest of the fwding
                        // package to get acked before being garbage collected.
                        continue
                }

                // An incoming HTLC add has been full-locked in. As a result we
                // can now examine the forwarding details of the HTLC, and the
                // HTLC itself to decide if: we should forward it, cancel it,
                // or are able to settle it (and it adheres to our fee related
                // constraints).

                // Before adding the new htlc to the state machine, parse the
                // onion object in order to obtain the routing information with
                // DecodeHopIterator function which process the Sphinx packet.
                chanIterator, failureCode := decodeResps[i].Result()
                if failureCode != lnwire.CodeNone {
                        // If we're unable to process the onion blob then we
                        // should send the malformed htlc error to payment
                        // sender.
                        l.sendMalformedHTLCError(
                                add.ID, failureCode, add.OnionBlob, &sourceRef,
                        )

                        l.log.Errorf("unable to decode onion hop "+
                                "iterator: %v", failureCode)
                        continue
                }

                heightNow := l.cfg.BestHeight()

                pld, routeRole, pldErr := chanIterator.HopPayload()
                if pldErr != nil {
                        // If we're unable to process the onion payload, or we
                        // received invalid onion payload failure, then we
                        // should send an error back to the caller so the HTLC
                        // can be canceled.
                        var failedType uint64

                        // We need to get the underlying error value, so we
                        // can't use errors.As as suggested by the linter.
                        //nolint:errorlint
                        if e, ok := pldErr.(hop.ErrInvalidPayload); ok {
                                failedType = uint64(e.Type)
                        }

                        // If we couldn't parse the payload, make our best
                        // effort at creating an error encrypter that knows
                        // what blinding type we were, but if we couldn't
                        // parse the payload we have no way of knowing whether
                        // we were the introduction node or not.
                        //
                        //nolint:ll
                        obfuscator, failCode := chanIterator.ExtractErrorEncrypter(
                                l.cfg.ExtractErrorEncrypter,
                                // We need our route role here because we
                                // couldn't parse or validate the payload.
                                routeRole == hop.RouteRoleIntroduction,
                        )
                        if failCode != lnwire.CodeNone {
                                l.log.Errorf("could not extract error "+
                                        "encrypter: %v", pldErr)

                                // We can't process this htlc, send back
                                // malformed.
                                l.sendMalformedHTLCError(
                                        add.ID, failureCode, add.OnionBlob,
                                        &sourceRef,
                                )

                                continue
                        }

                        // TODO: currently none of the test unit infrastructure
                        // is setup to handle TLV payloads, so testing this
                        // would require implementing a separate mock iterator
                        // for TLV payloads that also supports injecting invalid
                        // payloads. Deferring this non-trival effort till a
                        // later date
                        failure := lnwire.NewInvalidOnionPayload(failedType, 0)

                        l.sendHTLCError(
                                add, sourceRef, NewLinkError(failure),
                                obfuscator, false,
                        )

                        l.log.Errorf("unable to decode forwarding "+
                                "instructions: %v", pldErr)

                        continue
                }

                // Retrieve onion obfuscator from onion blob in order to
                // produce initial obfuscation of the onion failureCode.
                obfuscator, failureCode := chanIterator.ExtractErrorEncrypter(
                        l.cfg.ExtractErrorEncrypter,
                        routeRole == hop.RouteRoleIntroduction,
                )
                if failureCode != lnwire.CodeNone {
                        // If we're unable to process the onion blob than we
                        // should send the malformed htlc error to payment
                        // sender.
                        l.sendMalformedHTLCError(
                                add.ID, failureCode, add.OnionBlob,
                                &sourceRef,
                        )

                        l.log.Errorf("unable to decode onion "+
                                "obfuscator: %v", failureCode)

                        continue
                }

                fwdInfo := pld.ForwardingInfo()

                // Check whether the payload we've just processed uses our
                // node as the introduction point (gave us a blinding key in
                // the payload itself) and fail it back if we don't support
                // route blinding.
                if fwdInfo.NextBlinding.IsSome() &&
                        l.cfg.DisallowRouteBlinding {

                        failure := lnwire.NewInvalidBlinding(
                                fn.Some(add.OnionBlob),
                        )

                        l.sendHTLCError(
                                add, sourceRef, NewLinkError(failure),
                                obfuscator, false,
                        )

                        l.log.Error("rejected htlc that uses use as an " +
                                "introduction point when we do not support " +
                                "route blinding")

                        continue
                }

                switch fwdInfo.NextHop {
                case hop.Exit:
                        err := l.processExitHop(
                                add, sourceRef, obfuscator, fwdInfo,
                                heightNow, pld,
                        )
                        if err != nil {
                                l.failf(LinkFailureError{
                                        code: ErrInternalError,
                                }, err.Error()) //nolint

                                return
                        }

                // There are additional channels left within this route. So
                // we'll simply do some forwarding package book-keeping.
                default:
                        // If hodl.AddIncoming is requested, we will not
                        // validate the forwarded ADD, nor will we send the
                        // packet to the htlc switch.
                        if l.cfg.HodlMask.Active(hodl.AddIncoming) {
                                l.log.Warnf(hodl.AddIncoming.Warning())
                                continue
                        }

                        endorseValue := l.experimentalEndorsement(
                                record.CustomSet(add.CustomRecords),
                        )
                        endorseType := uint64(
                                lnwire.ExperimentalEndorsementType,
                        )

                        switch fwdPkg.State {
                        case channeldb.FwdStateProcessed:
                                // This add was not forwarded on the previous
                                // processing phase, run it through our
                                // validation pipeline to reproduce an error.
                                // This may trigger a different error due to
                                // expiring timelocks, but we expect that an
                                // error will be reproduced.
                                if !fwdPkg.FwdFilter.Contains(idx) {
                                        break
                                }

                                // Otherwise, it was already processed, we can
                                // can collect it and continue.
                                outgoingAdd := &lnwire.UpdateAddHTLC{
                                        Expiry:        fwdInfo.OutgoingCTLV,
                                        Amount:        fwdInfo.AmountToForward,
                                        PaymentHash:   add.PaymentHash,
                                        BlindingPoint: fwdInfo.NextBlinding,
                                }

                                endorseValue.WhenSome(func(e byte) {
                                        custRecords := map[uint64][]byte{
                                                endorseType: {e},
                                        }

                                        outgoingAdd.CustomRecords = custRecords
                                })

                                // Finally, we'll encode the onion packet for
                                // the _next_ hop using the hop iterator
                                // decoded for the current hop.
                                buf := bytes.NewBuffer(
                                        outgoingAdd.OnionBlob[0:0],
                                )

                                // We know this cannot fail, as this ADD
                                // was marked forwarded in a previous
                                // round of processing.
                                chanIterator.EncodeNextHop(buf)

                                inboundFee := l.cfg.FwrdingPolicy.InboundFee

                                //nolint:ll
                                updatePacket := &htlcPacket{
                                        incomingChanID:       l.ShortChanID(),
                                        incomingHTLCID:       add.ID,
                                        outgoingChanID:       fwdInfo.NextHop,
                                        sourceRef:            &sourceRef,
                                        incomingAmount:       add.Amount,
                                        amount:               outgoingAdd.Amount,
                                        htlc:                 outgoingAdd,
                                        obfuscator:           obfuscator,
                                        incomingTimeout:      add.Expiry,
                                        outgoingTimeout:      fwdInfo.OutgoingCTLV,
                                        inOnionCustomRecords: pld.CustomRecords(),
                                        inboundFee:           inboundFee,
                                        inWireCustomRecords:  add.CustomRecords.Copy(),
                                }
                                switchPackets = append(
                                        switchPackets, updatePacket,
                                )

                                continue
                        }

                        // TODO(roasbeef): ensure don't accept outrageous
                        // timeout for htlc

                        // With all our forwarding constraints met, we'll
                        // create the outgoing HTLC using the parameters as
                        // specified in the forwarding info.
                        addMsg := &lnwire.UpdateAddHTLC{
                                Expiry:        fwdInfo.OutgoingCTLV,
                                Amount:        fwdInfo.AmountToForward,
                                PaymentHash:   add.PaymentHash,
                                BlindingPoint: fwdInfo.NextBlinding,
                        }

                        endorseValue.WhenSome(func(e byte) {
                                addMsg.CustomRecords = map[uint64][]byte{
                                        endorseType: {e},
                                }
                        })

                        // Finally, we'll encode the onion packet for the
                        // _next_ hop using the hop iterator decoded for the
                        // current hop.
                        buf := bytes.NewBuffer(addMsg.OnionBlob[0:0])
                        err := chanIterator.EncodeNextHop(buf)
                        if err != nil {
                                l.log.Errorf("unable to encode the "+
                                        "remaining route %v", err)

                                cb := func(upd *lnwire.ChannelUpdate1) lnwire.FailureMessage { //nolint:ll
                                        return lnwire.NewTemporaryChannelFailure(upd)
                                }

                                failure := l.createFailureWithUpdate(
                                        true, hop.Source, cb,
                                )

                                l.sendHTLCError(
                                        add, sourceRef, NewLinkError(failure),
                                        obfuscator, false,
                                )
                                continue
                        }

                        // Now that this add has been reprocessed, only append
                        // it to our list of packets to forward to the switch
                        // this is the first time processing the add. If the
                        // fwd pkg has already been processed, then we entered
                        // the above section to recreate a previous error.  If
                        // the packet had previously been forwarded, it would
                        // have been added to switchPackets at the top of this
                        // section.
                        if fwdPkg.State == channeldb.FwdStateLockedIn {
                                inboundFee := l.cfg.FwrdingPolicy.InboundFee

                                //nolint:ll
                                updatePacket := &htlcPacket{
                                        incomingChanID:       l.ShortChanID(),
                                        incomingHTLCID:       add.ID,
                                        outgoingChanID:       fwdInfo.NextHop,
                                        sourceRef:            &sourceRef,
                                        incomingAmount:       add.Amount,
                                        amount:               addMsg.Amount,
                                        htlc:                 addMsg,
                                        obfuscator:           obfuscator,
                                        incomingTimeout:      add.Expiry,
                                        outgoingTimeout:      fwdInfo.OutgoingCTLV,
                                        inOnionCustomRecords: pld.CustomRecords(),
                                        inboundFee:           inboundFee,
                                        inWireCustomRecords:  add.CustomRecords.Copy(),
                                }

                                fwdPkg.FwdFilter.Set(idx)
                                switchPackets = append(switchPackets,
                                        updatePacket)
                        }
                }
        }

        // Commit the htlcs we are intending to forward if this package has not
        // been fully processed.
        if fwdPkg.State == channeldb.FwdStateLockedIn {
                err := l.channel.SetFwdFilter(fwdPkg.Height, fwdPkg.FwdFilter)
                if err != nil {
                        l.failf(LinkFailureError{code: ErrInternalError},
                                "unable to set fwd filter: %v", err)
                        return
                }
        }

        if len(switchPackets) == 0 {
                return
        }

        replay := fwdPkg.State != channeldb.FwdStateLockedIn

        l.log.Debugf("forwarding %d packets to switch: replay=%v",
                len(switchPackets), replay)

        // NOTE: This call is made synchronous so that we ensure all circuits
        // are committed in the exact order that they are processed in the link.
        // Failing to do this could cause reorderings/gaps in the range of
        // opened circuits, which violates assumptions made by the circuit
        // trimming.
        l.forwardBatch(replay, switchPackets...)
}

// experimentalEndorsement returns the value to set for our outgoing
// experimental endorsement field, and a boolean indicating whether it should
// be populated on the outgoing htlc.
func (l *channelLink) experimentalEndorsement(
        customUpdateAdd record.CustomSet) fn.Option[byte] {

        // Only relay experimental signal if we are within the experiment
        // period.
        if !l.cfg.ShouldFwdExpEndorsement() {
                return fn.None[byte]()
        }

        // If we don't have any custom records or the experimental field is
        // not set, just forward a zero value.
        if len(customUpdateAdd) == 0 {
                return fn.Some[byte](lnwire.ExperimentalUnendorsed)
        }

        t := uint64(lnwire.ExperimentalEndorsementType)
        value, set := customUpdateAdd[t]
        if !set {
                return fn.Some[byte](lnwire.ExperimentalUnendorsed)
        }

        // We expect at least one byte for this field, consider it invalid if
        // it has no data and just forward a zero value.
        if len(value) == 0 {
                return fn.Some[byte](lnwire.ExperimentalUnendorsed)
        }

        // Only forward endorsed if the incoming link is endorsed.
        if value[0] == lnwire.ExperimentalEndorsed {
                return fn.Some[byte](lnwire.ExperimentalEndorsed)
        }

        // Forward as unendorsed otherwise, including cases where we've
        // received an invalid value that uses more than 3 bits of information.
        return fn.Some[byte](lnwire.ExperimentalUnendorsed)
}

// processExitHop handles an htlc for which this link is the exit hop. It
// returns a boolean indicating whether the commitment tx needs an update.
func (l *channelLink) processExitHop(add lnwire.UpdateAddHTLC,
        sourceRef channeldb.AddRef, obfuscator hop.ErrorEncrypter,
        fwdInfo hop.ForwardingInfo, heightNow uint32,
        payload invoices.Payload) error {

        // If hodl.ExitSettle is requested, we will not validate the final hop's
        // ADD, nor will we settle the corresponding invoice or respond with the
        // preimage.
        if l.cfg.HodlMask.Active(hodl.ExitSettle) {
                l.log.Warnf("%s for htlc(rhash=%x,htlcIndex=%v)",
                        hodl.ExitSettle.Warning(), add.PaymentHash, add.ID)

                return nil
        }

        // In case the traffic shaper is active, we'll check if the HTLC has
        // custom records and skip the amount check in the onion payload below.
        isCustomHTLC := fn.MapOptionZ(
                l.cfg.AuxTrafficShaper,
                func(ts AuxTrafficShaper) bool {
                        return ts.IsCustomHTLC(add.CustomRecords)
                },
        )

        // As we're the exit hop, we'll double check the hop-payload included in
        // the HTLC to ensure that it was crafted correctly by the sender and
        // is compatible with the HTLC we were extended. If an external
        // validator is active we might bypass the amount check.
        if !isCustomHTLC && add.Amount < fwdInfo.AmountToForward {
                l.log.Errorf("onion payload of incoming htlc(%x) has "+
                        "incompatible value: expected <=%v, got %v",
                        add.PaymentHash, add.Amount, fwdInfo.AmountToForward)

                failure := NewLinkError(
                        lnwire.NewFinalIncorrectHtlcAmount(add.Amount),
                )
                l.sendHTLCError(add, sourceRef, failure, obfuscator, true)

                return nil
        }

        // We'll also ensure that our time-lock value has been computed
        // correctly.
        if add.Expiry < fwdInfo.OutgoingCTLV {
                l.log.Errorf("onion payload of incoming htlc(%x) has "+
                        "incompatible time-lock: expected <=%v, got %v",
                        add.PaymentHash, add.Expiry, fwdInfo.OutgoingCTLV)

                failure := NewLinkError(
                        lnwire.NewFinalIncorrectCltvExpiry(add.Expiry),
                )

                l.sendHTLCError(add, sourceRef, failure, obfuscator, true)

                return nil
        }

        // Notify the invoiceRegistry of the exit hop htlc. If we crash right
        // after this, this code will be re-executed after restart. We will
        // receive back a resolution event.
        invoiceHash := lntypes.Hash(add.PaymentHash)

        circuitKey := models.CircuitKey{
                ChanID: l.ShortChanID(),
                HtlcID: add.ID,
        }

        event, err := l.cfg.Registry.NotifyExitHopHtlc(
                invoiceHash, add.Amount, add.Expiry, int32(heightNow),
                circuitKey, l.hodlQueue.ChanIn(), add.CustomRecords, payload,
        )
        if err != nil {
                return err
        }

        // Create a hodlHtlc struct and decide either resolved now or later.
        htlc := hodlHtlc{
                add:        add,
                sourceRef:  sourceRef,
                obfuscator: obfuscator,
        }

        // If the event is nil, the invoice is being held, so we save payment
        // descriptor for future reference.
        if event == nil {
                l.hodlMap[circuitKey] = htlc
                return nil
        }

        // Process the received resolution.
        return l.processHtlcResolution(event, htlc)
}

// settleHTLC settles the HTLC on the channel.
func (l *channelLink) settleHTLC(preimage lntypes.Preimage,
        htlcIndex uint64, sourceRef channeldb.AddRef) error {

        hash := preimage.Hash()

        l.log.Infof("settling htlc %v as exit hop", hash)

        err := l.channel.SettleHTLC(
                preimage, htlcIndex, &sourceRef, nil, nil,
        )
        if err != nil {
                return fmt.Errorf("unable to settle htlc: %w", err)
        }

        // If the link is in hodl.BogusSettle mode, replace the preimage with a
        // fake one before sending it to the peer.
        if l.cfg.HodlMask.Active(hodl.BogusSettle) {
                l.log.Warnf(hodl.BogusSettle.Warning())
                preimage = [32]byte{}
                copy(preimage[:], bytes.Repeat([]byte{2}, 32))
        }

        // HTLC was successfully settled locally send notification about it
        // remote peer.
        l.cfg.Peer.SendMessage(false, &lnwire.UpdateFulfillHTLC{
                ChanID:          l.ChanID(),
                ID:              htlcIndex,
                PaymentPreimage: preimage,
        })

        // Once we have successfully settled the htlc, notify a settle event.
        l.cfg.HtlcNotifier.NotifySettleEvent(
                HtlcKey{
                        IncomingCircuit: models.CircuitKey{
                                ChanID: l.ShortChanID(),
                                HtlcID: htlcIndex,
                        },
                },
                preimage,
                HtlcEventTypeReceive,
        )

        return nil
}

// forwardBatch forwards the given htlcPackets to the switch, and waits on the
// err chan for the individual responses. This method is intended to be spawned
// as a goroutine so the responses can be handled in the background.
func (l *channelLink) forwardBatch(replay bool, packets ...*htlcPacket) {
        // Don't forward packets for which we already have a response in our
        // mailbox. This could happen if a packet fails and is buffered in the
        // mailbox, and the incoming link flaps.
        var filteredPkts = make([]*htlcPacket, 0, len(packets))
        for _, pkt := range packets {
                if l.mailBox.HasPacket(pkt.inKey()) {
                        continue
                }

                filteredPkts = append(filteredPkts, pkt)
        }

        err := l.cfg.ForwardPackets(l.cg.Done(), replay, filteredPkts...)
        if err != nil {
                log.Errorf("Unhandled error while reforwarding htlc "+
                        "settle/fail over htlcswitch: %v", err)
        }
}

// sendHTLCError functions cancels HTLC and send cancel message back to the
// peer from which HTLC was received.
func (l *channelLink) sendHTLCError(add lnwire.UpdateAddHTLC,
        sourceRef channeldb.AddRef, failure *LinkError,
        e hop.ErrorEncrypter, isReceive bool) {

        reason, err := e.EncryptFirstHop(failure.WireMessage())
        if err != nil {
                l.log.Errorf("unable to obfuscate error: %v", err)
                return
        }

        err = l.channel.FailHTLC(add.ID, reason, &sourceRef, nil, nil)
        if err != nil {
                l.log.Errorf("unable cancel htlc: %v", err)
                return
        }

        // Send the appropriate failure message depending on whether we're
        // in a blinded route or not.
        if err := l.sendIncomingHTLCFailureMsg(
                add.ID, e, reason,
        ); err != nil {
                l.log.Errorf("unable to send HTLC failure: %v", err)
                return
        }

        // Notify a link failure on our incoming link. Outgoing htlc information
        // is not available at this point, because we have not decrypted the
        // onion, so it is excluded.
        var eventType HtlcEventType
        if isReceive {
                eventType = HtlcEventTypeReceive
        } else {
                eventType = HtlcEventTypeForward
        }

        l.cfg.HtlcNotifier.NotifyLinkFailEvent(
                HtlcKey{
                        IncomingCircuit: models.CircuitKey{
                                ChanID: l.ShortChanID(),
                                HtlcID: add.ID,
                        },
                },
                HtlcInfo{
                        IncomingTimeLock: add.Expiry,
                        IncomingAmt:      add.Amount,
                },
                eventType,
                failure,
                true,
        )
}

// sendPeerHTLCFailure handles sending a HTLC failure message back to the
// peer from which the HTLC was received. This function is primarily used to
// handle the special requirements of route blinding, specifically:
// - Forwarding nodes must switch out any errors with MalformedFailHTLC
// - Introduction nodes should return regular HTLC failure messages.
//
// It accepts the original opaque failure, which will be used in the case
// that we're not part of a blinded route and an error encrypter that'll be
// used if we are the introduction node and need to present an error as if
// we're the failing party.
func (l *channelLink) sendIncomingHTLCFailureMsg(htlcIndex uint64,
        e hop.ErrorEncrypter,
        originalFailure lnwire.OpaqueReason) error {

        var msg lnwire.Message
        switch {
        // Our circuit's error encrypter will be nil if this was a locally
        // initiated payment. We can only hit a blinded error for a locally
        // initiated payment if we allow ourselves to be picked as the
        // introduction node for our own payments and in that case we
        // shouldn't reach this code. To prevent the HTLC getting stuck,
        // we fail it back and log an error.
        // code.
        case e == nil:
                msg = &lnwire.UpdateFailHTLC{
                        ChanID: l.ChanID(),
                        ID:     htlcIndex,
                        Reason: originalFailure,
                }

                l.log.Errorf("Unexpected blinded failure when "+
                        "we are the sending node, incoming htlc: %v(%v)",
                        l.ShortChanID(), htlcIndex)

        // For cleartext hops (ie, non-blinded/normal) we don't need any
        // transformation on the error message and can just send the original.
        case !e.Type().IsBlinded():
                msg = &lnwire.UpdateFailHTLC{
                        ChanID: l.ChanID(),
                        ID:     htlcIndex,
                        Reason: originalFailure,
                }

        // When we're the introduction node, we need to convert the error to
        // a UpdateFailHTLC.
        case e.Type() == hop.EncrypterTypeIntroduction:
                l.log.Debugf("Introduction blinded node switching out failure "+
                        "error: %v", htlcIndex)

                // The specification does not require that we set the onion
                // blob.
                failureMsg := lnwire.NewInvalidBlinding(
                        fn.None[[lnwire.OnionPacketSize]byte](),
                )
                reason, err := e.EncryptFirstHop(failureMsg)
                if err != nil {
                        return err
                }

                msg = &lnwire.UpdateFailHTLC{
                        ChanID: l.ChanID(),
                        ID:     htlcIndex,
                        Reason: reason,
                }

        // If we are a relaying node, we need to switch out any error that
        // we've received to a malformed HTLC error.
        case e.Type() == hop.EncrypterTypeRelaying:
                l.log.Debugf("Relaying blinded node switching out malformed "+
                        "error: %v", htlcIndex)

                msg = &lnwire.UpdateFailMalformedHTLC{
                        ChanID:      l.ChanID(),
                        ID:          htlcIndex,
                        FailureCode: lnwire.CodeInvalidBlinding,
                }

        default:
                return fmt.Errorf("unexpected encrypter: %d", e)
        }

        if err := l.cfg.Peer.SendMessage(false, msg); err != nil {
                l.log.Warnf("Send update fail failed: %v", err)
        }

        return nil
}

// sendMalformedHTLCError helper function which sends the malformed HTLC update
// to the payment sender.
func (l *channelLink) sendMalformedHTLCError(htlcIndex uint64,
        code lnwire.FailCode, onionBlob [lnwire.OnionPacketSize]byte,
        sourceRef *channeldb.AddRef) {

        shaOnionBlob := sha256.Sum256(onionBlob[:])
        err := l.channel.MalformedFailHTLC(htlcIndex, code, shaOnionBlob, sourceRef)
        if err != nil {
                l.log.Errorf("unable cancel htlc: %v", err)
                return
        }

        l.cfg.Peer.SendMessage(false, &lnwire.UpdateFailMalformedHTLC{
                ChanID:       l.ChanID(),
                ID:           htlcIndex,
                ShaOnionBlob: shaOnionBlob,
                FailureCode:  code,
        })
}

// failf is a function which is used to encapsulate the action necessary for
// properly failing the link. It takes a LinkFailureError, which will be passed
// to the OnChannelFailure closure, in order for it to determine if we should
// force close the channel, and if we should send an error message to the
// remote peer.
func (l *channelLink) failf(linkErr LinkFailureError, format string,
        a ...interface{}) {

        reason := fmt.Errorf(format, a...)

        // Return if we have already notified about a failure.
        if l.failed {
                l.log.Warnf("ignoring link failure (%v), as link already "+
                        "failed", reason)
                return
        }

        l.log.Errorf("failing link: %s with error: %v", reason, linkErr)

        // Set failed, such that we won't process any more updates, and notify
        // the peer about the failure.
        l.failed = true
        l.cfg.OnChannelFailure(l.ChanID(), l.ShortChanID(), linkErr)
}

// FundingCustomBlob returns the custom funding blob of the channel that this
// link is associated with. The funding blob represents static information about
// the channel that was created at channel funding time.
func (l *channelLink) FundingCustomBlob() fn.Option[tlv.Blob] {
        if l.channel == nil {
                return fn.None[tlv.Blob]()
        }

        if l.channel.State() == nil {
                return fn.None[tlv.Blob]()
        }

        return l.channel.State().CustomBlob
}

// CommitmentCustomBlob returns the custom blob of the current local commitment
// of the channel that this link is associated with.
func (l *channelLink) CommitmentCustomBlob() fn.Option[tlv.Blob] {
        if l.channel == nil {
                return fn.None[tlv.Blob]()
        }

        return l.channel.LocalCommitmentBlob()
}

lightningnetwork / lnd / 13587276956

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