9915780197

Committed 13 Jul 2024 12:30AM UTC coverage: 49.268% (-9.1%) from 58.413%

Build # 9915780197

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #8653 from ProofOfKeags/fn-prim

DynComms [0/n]: `fn` package additions

Run Details

92837 of 188433 relevant lines covered (49.27%)

1.55 hits per line

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

64.75

/routing/router.go

package routing

import (
        "bytes"
        "context"
        "fmt"
        "math"
        "runtime"
        "strings"
        "sync"
        "sync/atomic"
        "time"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/wire"
        "github.com/davecgh/go-spew/spew"
        "github.com/go-errors/errors"
        sphinx "github.com/lightningnetwork/lightning-onion"
        "github.com/lightningnetwork/lnd/amp"
        "github.com/lightningnetwork/lnd/batch"
        "github.com/lightningnetwork/lnd/chainntnfs"
        "github.com/lightningnetwork/lnd/channeldb"
        "github.com/lightningnetwork/lnd/channeldb/models"
        "github.com/lightningnetwork/lnd/clock"
        "github.com/lightningnetwork/lnd/fn"
        "github.com/lightningnetwork/lnd/htlcswitch"
        "github.com/lightningnetwork/lnd/input"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lntypes"
        "github.com/lightningnetwork/lnd/lnutils"
        "github.com/lightningnetwork/lnd/lnwallet"
        "github.com/lightningnetwork/lnd/lnwallet/btcwallet"
        "github.com/lightningnetwork/lnd/lnwallet/chanvalidate"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/multimutex"
        "github.com/lightningnetwork/lnd/record"
        "github.com/lightningnetwork/lnd/routing/chainview"
        "github.com/lightningnetwork/lnd/routing/route"
        "github.com/lightningnetwork/lnd/routing/shards"
        "github.com/lightningnetwork/lnd/ticker"
        "github.com/lightningnetwork/lnd/zpay32"
)

const (
        // DefaultPayAttemptTimeout is the default payment attempt timeout. The
        // payment attempt timeout defines the duration after which we stop
        // trying more routes for a payment.
        DefaultPayAttemptTimeout = time.Second * 60

        // DefaultChannelPruneExpiry is the default duration used to determine
        // if a channel should be pruned or not.
        DefaultChannelPruneExpiry = time.Hour * 24 * 14

        // DefaultFirstTimePruneDelay is the time we'll wait after startup
        // before attempting to prune the graph for zombie channels. We don't
        // do it immediately after startup to allow lnd to start up without
        // getting blocked by this job.
        DefaultFirstTimePruneDelay = 30 * time.Second

        // defaultStatInterval governs how often the router will log non-empty
        // stats related to processing new channels, updates, or node
        // announcements.
        defaultStatInterval = time.Minute

        // MinCLTVDelta is the minimum CLTV value accepted by LND for all
        // timelock deltas. This includes both forwarding CLTV deltas set on
        // channel updates, as well as final CLTV deltas used to create BOLT 11
        // payment requests.
        //
        // NOTE: For payment requests, BOLT 11 stipulates that a final CLTV
        // delta of 9 should be used when no value is decoded. This however
        // leads to inflexibility in upgrading this default parameter, since it
        // can create inconsistencies around the assumed value between sender
        // and receiver. Specifically, if the receiver assumes a higher value
        // than the sender, the receiver will always see the received HTLCs as
        // invalid due to their timelock not meeting the required delta.
        //
        // We skirt this by always setting an explicit CLTV delta when creating
        // invoices. This allows LND nodes to freely update the minimum without
        // creating incompatibilities during the upgrade process. For some time
        // LND has used an explicit default final CLTV delta of 40 blocks for
        // bitcoin, though we now clamp the lower end of this
        // range for user-chosen deltas to 18 blocks to be conservative.
        MinCLTVDelta = 18

        // MaxCLTVDelta is the maximum CLTV value accepted by LND for all
        // timelock deltas.
        MaxCLTVDelta = math.MaxUint16
)

var (
        // ErrRouterShuttingDown is returned if the router is in the process of
        // shutting down.
        ErrRouterShuttingDown = fmt.Errorf("router shutting down")

        // ErrSelfIntro is a failure returned when the source node of a
        // route request is also the introduction node. This is not yet
        // supported because LND does not support blinded forwardingg.
        ErrSelfIntro = errors.New("introduction point as own node not " +
                "supported")

        // ErrHintsAndBlinded is returned if a route request has both
        // bolt 11 route hints and a blinded path set.
        ErrHintsAndBlinded = errors.New("bolt 11 route hints and blinded " +
                "paths are mutually exclusive")

        // ErrExpiryAndBlinded is returned if a final cltv and a blinded path
        // are provided, as the cltv should be provided within the blinded
        // path.
        ErrExpiryAndBlinded = errors.New("final cltv delta and blinded " +
                "paths are mutually exclusive")

        // ErrTargetAndBlinded is returned is a target destination and a
        // blinded path are both set (as the target is inferred from the
        // blinded path).
        ErrTargetAndBlinded = errors.New("target node and blinded paths " +
                "are mutually exclusive")

        // ErrNoTarget is returned when the target node for a route is not
        // provided by either a blinded route or a cleartext pubkey.
        ErrNoTarget = errors.New("destination not set in target or blinded " +
                "path")

        // ErrSkipTempErr is returned when a non-MPP is made yet the
        // skipTempErr flag is set.
        ErrSkipTempErr = errors.New("cannot skip temp error for non-MPP")
)

// ChannelGraphSource represents the source of information about the topology
// of the lightning network. It's responsible for the addition of nodes, edges,
// applying edge updates, and returning the current block height with which the
// topology is synchronized.
type ChannelGraphSource interface {
        // AddNode is used to add information about a node to the router
        // database. If the node with this pubkey is not present in an existing
        // channel, it will be ignored.
        AddNode(node *channeldb.LightningNode,
                op ...batch.SchedulerOption) error

        // AddEdge is used to add edge/channel to the topology of the router,
        // after all information about channel will be gathered this
        // edge/channel might be used in construction of payment path.
        AddEdge(edge *models.ChannelEdgeInfo,
                op ...batch.SchedulerOption) error

        // AddProof updates the channel edge info with proof which is needed to
        // properly announce the edge to the rest of the network.
        AddProof(chanID lnwire.ShortChannelID,
                proof *models.ChannelAuthProof) error

        // UpdateEdge is used to update edge information, without this message
        // edge considered as not fully constructed.
        UpdateEdge(policy *models.ChannelEdgePolicy,
                op ...batch.SchedulerOption) error

        // IsStaleNode returns true if the graph source has a node announcement
        // for the target node with a more recent timestamp. This method will
        // also return true if we don't have an active channel announcement for
        // the target node.
        IsStaleNode(node route.Vertex, timestamp time.Time) bool

        // IsPublicNode determines whether the given vertex is seen as a public
        // node in the graph from the graph's source node's point of view.
        IsPublicNode(node route.Vertex) (bool, error)

        // IsKnownEdge returns true if the graph source already knows of the
        // passed channel ID either as a live or zombie edge.
        IsKnownEdge(chanID lnwire.ShortChannelID) bool

        // IsStaleEdgePolicy returns true if the graph source has a channel
        // edge for the passed channel ID (and flags) that have a more recent
        // timestamp.
        IsStaleEdgePolicy(chanID lnwire.ShortChannelID, timestamp time.Time,
                flags lnwire.ChanUpdateChanFlags) bool

        // MarkEdgeLive clears an edge from our zombie index, deeming it as
        // live.
        MarkEdgeLive(chanID lnwire.ShortChannelID) error

        // ForAllOutgoingChannels is used to iterate over all channels
        // emanating from the "source" node which is the center of the
        // star-graph.
        ForAllOutgoingChannels(cb func(tx kvdb.RTx,
                c *models.ChannelEdgeInfo,
                e *models.ChannelEdgePolicy) error) error

        // CurrentBlockHeight returns the block height from POV of the router
        // subsystem.
        CurrentBlockHeight() (uint32, error)

        // GetChannelByID return the channel by the channel id.
        GetChannelByID(chanID lnwire.ShortChannelID) (
                *models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
                *models.ChannelEdgePolicy, error)

        // FetchLightningNode attempts to look up a target node by its identity
        // public key. channeldb.ErrGraphNodeNotFound is returned if the node
        // doesn't exist within the graph.
        FetchLightningNode(route.Vertex) (*channeldb.LightningNode, error)

        // ForEachNode is used to iterate over every node in the known graph.
        ForEachNode(func(node *channeldb.LightningNode) error) error
}

// PaymentAttemptDispatcher is used by the router to send payment attempts onto
// the network, and receive their results.
type PaymentAttemptDispatcher interface {
        // SendHTLC is a function that directs a link-layer switch to
        // forward a fully encoded payment to the first hop in the route
        // denoted by its public key. A non-nil error is to be returned if the
        // payment was unsuccessful.
        SendHTLC(firstHop lnwire.ShortChannelID,
                attemptID uint64,
                htlcAdd *lnwire.UpdateAddHTLC) error

        // GetAttemptResult returns the result of the payment attempt with
        // the given attemptID. The paymentHash should be set to the payment's
        // overall hash, or in case of AMP payments the payment's unique
        // identifier.
        //
        // The method returns a channel where the payment result will be sent
        // when available, or an error is encountered during forwarding. When a
        // result is received on the channel, the HTLC is guaranteed to no
        // longer be in flight.  The switch shutting down is signaled by
        // closing the channel. If the attemptID is unknown,
        // ErrPaymentIDNotFound will be returned.
        GetAttemptResult(attemptID uint64, paymentHash lntypes.Hash,
                deobfuscator htlcswitch.ErrorDecrypter) (
                <-chan *htlcswitch.PaymentResult, error)

        // CleanStore calls the underlying result store, telling it is safe to
        // delete all entries except the ones in the keepPids map. This should
        // be called periodically to let the switch clean up payment results
        // that we have handled.
        // NOTE: New payment attempts MUST NOT be made after the keepPids map
        // has been created and this method has returned.
        CleanStore(keepPids map[uint64]struct{}) error
}

// PaymentSessionSource is an interface that defines a source for the router to
// retrieve new payment sessions.
type PaymentSessionSource interface {
        // NewPaymentSession creates a new payment session that will produce
        // routes to the given target. An optional set of routing hints can be
        // provided in order to populate additional edges to explore when
        // finding a path to the payment's destination.
        NewPaymentSession(p *LightningPayment) (PaymentSession, error)

        // NewPaymentSessionEmpty creates a new paymentSession instance that is
        // empty, and will be exhausted immediately. Used for failure reporting
        // to missioncontrol for resumed payment we don't want to make more
        // attempts for.
        NewPaymentSessionEmpty() PaymentSession
}

// MissionController is an interface that exposes failure reporting and
// probability estimation.
type MissionController interface {
        // ReportPaymentFail reports a failed payment to mission control as
        // input for future probability estimates. It returns a bool indicating
        // whether this error is a final error and no further payment attempts
        // need to be made.
        ReportPaymentFail(attemptID uint64, rt *route.Route,
                failureSourceIdx *int, failure lnwire.FailureMessage) (
                *channeldb.FailureReason, error)

        // ReportPaymentSuccess reports a successful payment to mission control
        // as input for future probability estimates.
        ReportPaymentSuccess(attemptID uint64, rt *route.Route) error

        // GetProbability is expected to return the success probability of a
        // payment from fromNode along edge.
        GetProbability(fromNode, toNode route.Vertex,
                amt lnwire.MilliSatoshi, capacity btcutil.Amount) float64
}

// FeeSchema is the set fee configuration for a Lightning Node on the network.
// Using the coefficients described within the schema, the required fee to
// forward outgoing payments can be derived.
type FeeSchema struct {
        // BaseFee is the base amount of milli-satoshis that will be chained
        // for ANY payment forwarded.
        BaseFee lnwire.MilliSatoshi

        // FeeRate is the rate that will be charged for forwarding payments.
        // This value should be interpreted as the numerator for a fraction
        // (fixed point arithmetic) whose denominator is 1 million. As a result
        // the effective fee rate charged per mSAT will be: (amount *
        // FeeRate/1,000,000).
        FeeRate uint32

        // InboundFee is the inbound fee schedule that applies to forwards
        // coming in through a channel to which this FeeSchema pertains.
        InboundFee fn.Option[models.InboundFee]
}

// ChannelPolicy holds the parameters that determine the policy we enforce
// when forwarding payments on a channel. These parameters are communicated
// to the rest of the network in ChannelUpdate messages.
type ChannelPolicy struct {
        // FeeSchema holds the fee configuration for a channel.
        FeeSchema

        // TimeLockDelta is the required HTLC timelock delta to be used
        // when forwarding payments.
        TimeLockDelta uint32

        // MaxHTLC is the maximum HTLC size including fees we are allowed to
        // forward over this channel.
        MaxHTLC lnwire.MilliSatoshi

        // MinHTLC is the minimum HTLC size including fees we are allowed to
        // forward over this channel.
        MinHTLC *lnwire.MilliSatoshi
}

// Config defines the configuration for the ChannelRouter. ALL elements within
// the configuration MUST be non-nil for the ChannelRouter to carry out its
// duties.
type Config struct {
        // Graph is the channel graph that the ChannelRouter will use to gather
        // metrics from and also to carry out path finding queries.
        // TODO(roasbeef): make into an interface
        Graph *channeldb.ChannelGraph

        // Chain is the router's source to the most up-to-date blockchain data.
        // All incoming advertised channels will be checked against the chain
        // to ensure that the channels advertised are still open.
        Chain lnwallet.BlockChainIO

        // ChainView is an instance of a FilteredChainView which is used to
        // watch the sub-set of the UTXO set (the set of active channels) that
        // we need in order to properly maintain the channel graph.
        ChainView chainview.FilteredChainView

        // Notifier is a reference to the ChainNotifier, used to grab
        // the latest blocks if the router is missing any.
        Notifier chainntnfs.ChainNotifier

        // Payer is an instance of a PaymentAttemptDispatcher and is used by
        // the router to send payment attempts onto the network, and receive
        // their results.
        Payer PaymentAttemptDispatcher

        // Control keeps track of the status of ongoing payments, ensuring we
        // can properly resume them across restarts.
        Control ControlTower

        // MissionControl is a shared memory of sorts that executions of
        // payment path finding use in order to remember which vertexes/edges
        // were pruned from prior attempts. During SendPayment execution,
        // errors sent by nodes are mapped into a vertex or edge to be pruned.
        // Each run will then take into account this set of pruned
        // vertexes/edges to reduce route failure and pass on graph information
        // gained to the next execution.
        MissionControl MissionController

        // SessionSource defines a source for the router to retrieve new payment
        // sessions.
        SessionSource PaymentSessionSource

        // ChannelPruneExpiry is the duration used to determine if a channel
        // should be pruned or not. If the delta between now and when the
        // channel was last updated is greater than ChannelPruneExpiry, then
        // the channel is marked as a zombie channel eligible for pruning.
        ChannelPruneExpiry time.Duration

        // GraphPruneInterval is used as an interval to determine how often we
        // should examine the channel graph to garbage collect zombie channels.
        GraphPruneInterval time.Duration

        // FirstTimePruneDelay is the time we'll wait after startup before
        // attempting to prune the graph for zombie channels. We don't do it
        // immediately after startup to allow lnd to start up without getting
        // blocked by this job.
        FirstTimePruneDelay time.Duration

        // QueryBandwidth is a method that allows the router to query the lower
        // link layer to determine the up-to-date available bandwidth at a
        // prospective link to be traversed. If the  link isn't available, then
        // a value of zero should be returned. Otherwise, the current up-to-
        // date knowledge of the available bandwidth of the link should be
        // returned.
        GetLink getLinkQuery

        // NextPaymentID is a method that guarantees to return a new, unique ID
        // each time it is called. This is used by the router to generate a
        // unique payment ID for each payment it attempts to send, such that
        // the switch can properly handle the HTLC.
        NextPaymentID func() (uint64, error)

        // AssumeChannelValid toggles whether the router will check for
        // spentness of channel outpoints. For neutrino, this saves long rescans
        // from blocking initial usage of the daemon.
        AssumeChannelValid bool

        // PathFindingConfig defines global path finding parameters.
        PathFindingConfig PathFindingConfig

        // Clock is mockable time provider.
        Clock clock.Clock

        // StrictZombiePruning determines if we attempt to prune zombie
        // channels according to a stricter criteria. If true, then we'll prune
        // a channel if only *one* of the edges is considered a zombie.
        // Otherwise, we'll only prune the channel when both edges have a very
        // dated last update.
        StrictZombiePruning bool

        // IsAlias returns whether a passed ShortChannelID is an alias. This is
        // only used for our local channels.
        IsAlias func(scid lnwire.ShortChannelID) bool
}

// EdgeLocator is a struct used to identify a specific edge.
type EdgeLocator struct {
        // ChannelID is the channel of this edge.
        ChannelID uint64

        // Direction takes the value of 0 or 1 and is identical in definition to
        // the channel direction flag. A value of 0 means the direction from the
        // lower node pubkey to the higher.
        Direction uint8
}

// String returns a human-readable version of the edgeLocator values.
func (e *EdgeLocator) String() string {
        return fmt.Sprintf("%v:%v", e.ChannelID, e.Direction)
}

// ChannelRouter is the layer 3 router within the Lightning stack. Below the
// ChannelRouter is the HtlcSwitch, and below that is the Bitcoin blockchain
// itself. The primary role of the ChannelRouter is to respond to queries for
// potential routes that can support a payment amount, and also general graph
// reachability questions. The router will prune the channel graph
// automatically as new blocks are discovered which spend certain known funding
// outpoints, thereby closing their respective channels.
type ChannelRouter struct {
        ntfnClientCounter uint64 // To be used atomically.

        started uint32 // To be used atomically.
        stopped uint32 // To be used atomically.

        bestHeight uint32 // To be used atomically.

        // cfg is a copy of the configuration struct that the ChannelRouter was
        // initialized with.
        cfg *Config

        // selfNode is the center of the star-graph centered around the
        // ChannelRouter. The ChannelRouter uses this node as a starting point
        // when doing any path finding.
        selfNode *channeldb.LightningNode

        // cachedGraph is an instance of routingGraph that caches the source
        // node as well as the channel graph itself in memory.
        cachedGraph routingGraph

        // newBlocks is a channel in which new blocks connected to the end of
        // the main chain are sent over, and blocks updated after a call to
        // UpdateFilter.
        newBlocks <-chan *chainview.FilteredBlock

        // staleBlocks is a channel in which blocks disconnected from the end
        // of our currently known best chain are sent over.
        staleBlocks <-chan *chainview.FilteredBlock

        // networkUpdates is a channel that carries new topology updates
        // messages from outside the ChannelRouter to be processed by the
        // networkHandler.
        networkUpdates chan *routingMsg

        // topologyClients maps a client's unique notification ID to a
        // topologyClient client that contains its notification dispatch
        // channel.
        topologyClients *lnutils.SyncMap[uint64, *topologyClient]

        // ntfnClientUpdates is a channel that's used to send new updates to
        // topology notification clients to the ChannelRouter. Updates either
        // add a new notification client, or cancel notifications for an
        // existing client.
        ntfnClientUpdates chan *topologyClientUpdate

        // channelEdgeMtx is a mutex we use to make sure we process only one
        // ChannelEdgePolicy at a time for a given channelID, to ensure
        // consistency between the various database accesses.
        channelEdgeMtx *multimutex.Mutex[uint64]

        // statTicker is a resumable ticker that logs the router's progress as
        // it discovers channels or receives updates.
        statTicker ticker.Ticker

        // stats tracks newly processed channels, updates, and node
        // announcements over a window of defaultStatInterval.
        stats *routerStats

        quit chan struct{}
        wg   sync.WaitGroup
}

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

// New creates a new instance of the ChannelRouter with the specified
// configuration parameters. As part of initialization, if the router detects
// that the channel graph isn't fully in sync with the latest UTXO (since the
// channel graph is a subset of the UTXO set) set, then the router will proceed
// to fully sync to the latest state of the UTXO set.
func New(cfg Config) (*ChannelRouter, error) {
        selfNode, err := cfg.Graph.SourceNode()
        if err != nil {
                return nil, err
        }

        r := &ChannelRouter{
                cfg: &cfg,
                cachedGraph: &CachedGraph{
                        graph:  cfg.Graph,
                        source: selfNode.PubKeyBytes,
                },
                networkUpdates:    make(chan *routingMsg),
                topologyClients:   &lnutils.SyncMap[uint64, *topologyClient]{},
                ntfnClientUpdates: make(chan *topologyClientUpdate),
                channelEdgeMtx:    multimutex.NewMutex[uint64](),
                selfNode:          selfNode,
                statTicker:        ticker.New(defaultStatInterval),
                stats:             new(routerStats),
                quit:              make(chan struct{}),
        }

        return r, nil
}

// Start launches all the goroutines the ChannelRouter requires to carry out
// its duties. If the router has already been started, then this method is a
// noop.
func (r *ChannelRouter) Start() error {
        if !atomic.CompareAndSwapUint32(&r.started, 0, 1) {
                return nil
        }

        log.Info("Channel Router starting")

        bestHash, bestHeight, err := r.cfg.Chain.GetBestBlock()
        if err != nil {
                return err
        }

        // If the graph has never been pruned, or hasn't fully been created yet,
        // then we don't treat this as an explicit error.
        if _, _, err := r.cfg.Graph.PruneTip(); err != nil {
                switch {
                case errors.Is(err, channeldb.ErrGraphNeverPruned):
                        fallthrough

                case errors.Is(err, channeldb.ErrGraphNotFound):
                        // If the graph has never been pruned, then we'll set
                        // the prune height to the current best height of the
                        // chain backend.
                        _, err = r.cfg.Graph.PruneGraph(
                                nil, bestHash, uint32(bestHeight),
                        )
                        if err != nil {
                                return err
                        }

                default:
                        return err
                }
        }

        // If AssumeChannelValid is present, then we won't rely on pruning
        // channels from the graph based on their spentness, but whether they
        // are considered zombies or not. We will start zombie pruning after a
        // small delay, to avoid slowing down startup of lnd.
        if r.cfg.AssumeChannelValid {
                time.AfterFunc(r.cfg.FirstTimePruneDelay, func() {
                        select {
                        case <-r.quit:
                                return
                        default:
                        }

                        log.Info("Initial zombie prune starting")
                        if err := r.pruneZombieChans(); err != nil {
                                log.Errorf("Unable to prune zombies: %v", err)
                        }
                })
        } else {
                // Otherwise, we'll use our filtered chain view to prune
                // channels as soon as they are detected as spent on-chain.
                if err := r.cfg.ChainView.Start(); err != nil {
                        return err
                }

                // Once the instance is active, we'll fetch the channel we'll
                // receive notifications over.
                r.newBlocks = r.cfg.ChainView.FilteredBlocks()
                r.staleBlocks = r.cfg.ChainView.DisconnectedBlocks()

                // Before we perform our manual block pruning, we'll construct
                // and apply a fresh chain filter to the active
                // FilteredChainView instance.  We do this before, as otherwise
                // we may miss on-chain events as the filter hasn't properly
                // been applied.
                channelView, err := r.cfg.Graph.ChannelView()
                if err != nil && !errors.Is(
                        err, channeldb.ErrGraphNoEdgesFound,
                ) {

                        return err
                }

                log.Infof("Filtering chain using %v channels active",
                        len(channelView))

                if len(channelView) != 0 {
                        err = r.cfg.ChainView.UpdateFilter(
                                channelView, uint32(bestHeight),
                        )
                        if err != nil {
                                return err
                        }
                }

                // The graph pruning might have taken a while and there could be
                // new blocks available.
                _, bestHeight, err = r.cfg.Chain.GetBestBlock()
                if err != nil {
                        return err
                }
                r.bestHeight = uint32(bestHeight)

                // Before we begin normal operation of the router, we first need
                // to synchronize the channel graph to the latest state of the
                // UTXO set.
                if err := r.syncGraphWithChain(); err != nil {
                        return err
                }

                // Finally, before we proceed, we'll prune any unconnected nodes
                // from the graph in order to ensure we maintain a tight graph
                // of "useful" nodes.
                err = r.cfg.Graph.PruneGraphNodes()
                if err != nil && !errors.Is(
                        err, channeldb.ErrGraphNodesNotFound,
                ) {

                        return err
                }
        }

        // If any payments are still in flight, we resume, to make sure their
        // results are properly handled.
        payments, err := r.cfg.Control.FetchInFlightPayments()
        if err != nil {
                return err
        }

        // Before we restart existing payments and start accepting more
        // payments to be made, we clean the network result store of the
        // Switch. We do this here at startup to ensure no more payments can be
        // made concurrently, so we know the toKeep map will be up-to-date
        // until the cleaning has finished.
        toKeep := make(map[uint64]struct{})
        for _, p := range payments {
                for _, a := range p.HTLCs {
                        toKeep[a.AttemptID] = struct{}{}
                }
        }

        log.Debugf("Cleaning network result store.")
        if err := r.cfg.Payer.CleanStore(toKeep); err != nil {
                return err
        }

        for _, payment := range payments {
                log.Infof("Resuming payment %v", payment.Info.PaymentIdentifier)
                r.wg.Add(1)
                go func(payment *channeldb.MPPayment) {
                        defer r.wg.Done()

                        // Get the hashes used for the outstanding HTLCs.
                        htlcs := make(map[uint64]lntypes.Hash)
                        for _, a := range payment.HTLCs {
                                a := a

                                // We check whether the individual attempts
                                // have their HTLC hash set, if not we'll fall
                                // back to the overall payment hash.
                                hash := payment.Info.PaymentIdentifier
                                if a.Hash != nil {
                                        hash = *a.Hash
                                }

                                htlcs[a.AttemptID] = hash
                        }

                        // Since we are not supporting creating more shards
                        // after a restart (only receiving the result of the
                        // shards already outstanding), we create a simple
                        // shard tracker that will map the attempt IDs to
                        // hashes used for the HTLCs. This will be enough also
                        // for AMP payments, since we only need the hashes for
                        // the individual HTLCs to regenerate the circuits, and
                        // we don't currently persist the root share necessary
                        // to re-derive them.
                        shardTracker := shards.NewSimpleShardTracker(
                                payment.Info.PaymentIdentifier, htlcs,
                        )

                        // We create a dummy, empty payment session such that
                        // we won't make another payment attempt when the
                        // result for the in-flight attempt is received.
                        paySession := r.cfg.SessionSource.NewPaymentSessionEmpty()

                        // We pass in a non-timeout context, to indicate we
                        // don't need it to timeout. It will stop immediately
                        // after the existing attempt has finished anyway. We
                        // also set a zero fee limit, as no more routes should
                        // be tried.
                        noTimeout := time.Duration(0)
                        _, _, err := r.sendPayment(
                                context.Background(), 0,
                                payment.Info.PaymentIdentifier, noTimeout,
                                paySession, shardTracker,
                        )
                        if err != nil {
                                log.Errorf("Resuming payment %v failed: %v.",
                                        payment.Info.PaymentIdentifier, err)
                                return
                        }

                        log.Infof("Resumed payment %v completed.",
                                payment.Info.PaymentIdentifier)
                }(payment)
        }

        r.wg.Add(1)
        go r.networkHandler()

        return nil
}

// Stop signals the ChannelRouter to gracefully halt all routines. This method
// will *block* until all goroutines have excited. If the channel router has
// already stopped then this method will return immediately.
func (r *ChannelRouter) Stop() error {
        if !atomic.CompareAndSwapUint32(&r.stopped, 0, 1) {
                return nil
        }

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

        // Our filtered chain view could've only been started if
        // AssumeChannelValid isn't present.
        if !r.cfg.AssumeChannelValid {
                if err := r.cfg.ChainView.Stop(); err != nil {
                        return err
                }
        }

        close(r.quit)
        r.wg.Wait()

        return nil
}

// syncGraphWithChain attempts to synchronize the current channel graph with
// the latest UTXO set state. This process involves pruning from the channel
// graph any channels which have been closed by spending their funding output
// since we've been down.
func (r *ChannelRouter) syncGraphWithChain() error {
        // First, we'll need to check to see if we're already in sync with the
        // latest state of the UTXO set.
        bestHash, bestHeight, err := r.cfg.Chain.GetBestBlock()
        if err != nil {
                return err
        }
        r.bestHeight = uint32(bestHeight)

        pruneHash, pruneHeight, err := r.cfg.Graph.PruneTip()
        if err != nil {
                switch {
                // If the graph has never been pruned, or hasn't fully been
                // created yet, then we don't treat this as an explicit error.
                case errors.Is(err, channeldb.ErrGraphNeverPruned):
                case errors.Is(err, channeldb.ErrGraphNotFound):
                default:
                        return err
                }
        }

        log.Infof("Prune tip for Channel Graph: height=%v, hash=%v",
                pruneHeight, pruneHash)

        switch {

        // If the graph has never been pruned, then we can exit early as this
        // entails it's being created for the first time and hasn't seen any
        // block or created channels.
        case pruneHeight == 0 || pruneHash == nil:
                return nil

        // If the block hashes and heights match exactly, then we don't need to
        // prune the channel graph as we're already fully in sync.
        case bestHash.IsEqual(pruneHash) && uint32(bestHeight) == pruneHeight:
                return nil
        }

        // If the main chain blockhash at prune height is different from the
        // prune hash, this might indicate the database is on a stale branch.
        mainBlockHash, err := r.cfg.Chain.GetBlockHash(int64(pruneHeight))
        if err != nil {
                return err
        }

        // While we are on a stale branch of the chain, walk backwards to find
        // first common block.
        for !pruneHash.IsEqual(mainBlockHash) {
                log.Infof("channel graph is stale. Disconnecting block %v "+
                        "(hash=%v)", pruneHeight, pruneHash)
                // Prune the graph for every channel that was opened at height
                // >= pruneHeight.
                _, err := r.cfg.Graph.DisconnectBlockAtHeight(pruneHeight)
                if err != nil {
                        return err
                }

                pruneHash, pruneHeight, err = r.cfg.Graph.PruneTip()
                if err != nil {
                        switch {
                        // If at this point the graph has never been pruned, we
                        // can exit as this entails we are back to the point
                        // where it hasn't seen any block or created channels,
                        // alas there's nothing left to prune.
                        case errors.Is(err, channeldb.ErrGraphNeverPruned):
                                return nil

                        case errors.Is(err, channeldb.ErrGraphNotFound):
                                return nil

                        default:
                                return err
                        }
                }
                mainBlockHash, err = r.cfg.Chain.GetBlockHash(int64(pruneHeight))
                if err != nil {
                        return err
                }
        }

        log.Infof("Syncing channel graph from height=%v (hash=%v) to height=%v "+
                "(hash=%v)", pruneHeight, pruneHash, bestHeight, bestHash)

        // If we're not yet caught up, then we'll walk forward in the chain
        // pruning the channel graph with each new block that hasn't yet been
        // consumed by the channel graph.
        var spentOutputs []*wire.OutPoint
        for nextHeight := pruneHeight + 1; nextHeight <= uint32(bestHeight); nextHeight++ {
                // Break out of the rescan early if a shutdown has been
                // requested, otherwise long rescans will block the daemon from
                // shutting down promptly.
                select {
                case <-r.quit:
                        return ErrRouterShuttingDown
                default:
                }

                // Using the next height, request a manual block pruning from
                // the chainview for the particular block hash.
                log.Infof("Filtering block for closed channels, at height: %v",
                        int64(nextHeight))
                nextHash, err := r.cfg.Chain.GetBlockHash(int64(nextHeight))
                if err != nil {
                        return err
                }
                log.Tracef("Running block filter on block with hash: %v",
                        nextHash)
                filterBlock, err := r.cfg.ChainView.FilterBlock(nextHash)
                if err != nil {
                        return err
                }

                // We're only interested in all prior outputs that have been
                // spent in the block, so collate all the referenced previous
                // outpoints within each tx and input.
                for _, tx := range filterBlock.Transactions {
                        for _, txIn := range tx.TxIn {
                                spentOutputs = append(spentOutputs,
                                        &txIn.PreviousOutPoint)
                        }
                }
        }

        // With the spent outputs gathered, attempt to prune the channel graph,
        // also passing in the best hash+height so the prune tip can be updated.
        closedChans, err := r.cfg.Graph.PruneGraph(
                spentOutputs, bestHash, uint32(bestHeight),
        )
        if err != nil {
                return err
        }

        log.Infof("Graph pruning complete: %v channels were closed since "+
                "height %v", len(closedChans), pruneHeight)
        return nil
}

// isZombieChannel takes two edge policy updates and determines if the
// corresponding channel should be considered a zombie. The first boolean is
// true if the policy update from node 1 is considered a zombie, the second
// boolean is that of node 2, and the final boolean is true if the channel
// is considered a zombie.
func (r *ChannelRouter) isZombieChannel(e1,
        e2 *models.ChannelEdgePolicy) (bool, bool, bool) {

        chanExpiry := r.cfg.ChannelPruneExpiry

        e1Zombie := e1 == nil || time.Since(e1.LastUpdate) >= chanExpiry
        e2Zombie := e2 == nil || time.Since(e2.LastUpdate) >= chanExpiry

        var e1Time, e2Time time.Time
        if e1 != nil {
                e1Time = e1.LastUpdate
        }
        if e2 != nil {
                e2Time = e2.LastUpdate
        }

        return e1Zombie, e2Zombie, r.IsZombieChannel(e1Time, e2Time)
}

// IsZombieChannel takes the timestamps of the latest channel updates for a
// channel and returns true if the channel should be considered a zombie based
// on these timestamps.
func (r *ChannelRouter) IsZombieChannel(updateTime1,
        updateTime2 time.Time) bool {

        chanExpiry := r.cfg.ChannelPruneExpiry

        e1Zombie := updateTime1.IsZero() ||
                time.Since(updateTime1) >= chanExpiry

        e2Zombie := updateTime2.IsZero() ||
                time.Since(updateTime2) >= chanExpiry

        // If we're using strict zombie pruning, then a channel is only
        // considered live if both edges have a recent update we know of.
        if r.cfg.StrictZombiePruning {
                return e1Zombie || e2Zombie
        }

        // Otherwise, if we're using the less strict variant, then a channel is
        // considered live if either of the edges have a recent update.
        return e1Zombie && e2Zombie
}

// pruneZombieChans is a method that will be called periodically to prune out
// any "zombie" channels. We consider channels zombies if *both* edges haven't
// been updated since our zombie horizon. If AssumeChannelValid is present,
// we'll also consider channels zombies if *both* edges are disabled. This
// usually signals that a channel has been closed on-chain. We do this
// periodically to keep a healthy, lively routing table.
func (r *ChannelRouter) pruneZombieChans() error {
        chansToPrune := make(map[uint64]struct{})
        chanExpiry := r.cfg.ChannelPruneExpiry

        log.Infof("Examining channel graph for zombie channels")

        // A helper method to detect if the channel belongs to this node
        isSelfChannelEdge := func(info *models.ChannelEdgeInfo) bool {
                return info.NodeKey1Bytes == r.selfNode.PubKeyBytes ||
                        info.NodeKey2Bytes == r.selfNode.PubKeyBytes
        }

        // First, we'll collect all the channels which are eligible for garbage
        // collection due to being zombies.
        filterPruneChans := func(info *models.ChannelEdgeInfo,
                e1, e2 *models.ChannelEdgePolicy) error {

                // Exit early in case this channel is already marked to be
                // pruned
                _, markedToPrune := chansToPrune[info.ChannelID]
                if markedToPrune {
                        return nil
                }

                // We'll ensure that we don't attempt to prune our *own*
                // channels from the graph, as in any case this should be
                // re-advertised by the sub-system above us.
                if isSelfChannelEdge(info) {
                        return nil
                }

                e1Zombie, e2Zombie, isZombieChan := r.isZombieChannel(e1, e2)

                if e1Zombie {
                        log.Tracef("Node1 pubkey=%x of chan_id=%v is zombie",
                                info.NodeKey1Bytes, info.ChannelID)
                }

                if e2Zombie {
                        log.Tracef("Node2 pubkey=%x of chan_id=%v is zombie",
                                info.NodeKey2Bytes, info.ChannelID)
                }

                // If either edge hasn't been updated for a period of
                // chanExpiry, then we'll mark the channel itself as eligible
                // for graph pruning.
                if !isZombieChan {
                        return nil
                }

                log.Debugf("ChannelID(%v) is a zombie, collecting to prune",
                        info.ChannelID)

                // TODO(roasbeef): add ability to delete single directional edge
                chansToPrune[info.ChannelID] = struct{}{}

                return nil
        }

        // If AssumeChannelValid is present we'll look at the disabled bit for
        // both edges. If they're both disabled, then we can interpret this as
        // the channel being closed and can prune it from our graph.
        if r.cfg.AssumeChannelValid {
                disabledChanIDs, err := r.cfg.Graph.DisabledChannelIDs()
                if err != nil {
                        return fmt.Errorf("unable to get disabled channels "+
                                "ids chans: %v", err)
                }

                disabledEdges, err := r.cfg.Graph.FetchChanInfos(
                        nil, disabledChanIDs,
                )
                if err != nil {
                        return fmt.Errorf("unable to fetch disabled channels "+
                                "edges chans: %v", err)
                }

                // Ensuring we won't prune our own channel from the graph.
                for _, disabledEdge := range disabledEdges {
                        if !isSelfChannelEdge(disabledEdge.Info) {
                                chansToPrune[disabledEdge.Info.ChannelID] =
                                        struct{}{}
                        }
                }
        }

        startTime := time.Unix(0, 0)
        endTime := time.Now().Add(-1 * chanExpiry)
        oldEdges, err := r.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime)
        if err != nil {
                return fmt.Errorf("unable to fetch expired channel updates "+
                        "chans: %v", err)
        }

        for _, u := range oldEdges {
                err = filterPruneChans(u.Info, u.Policy1, u.Policy2)
                if err != nil {
                        log.Warnf("Filter pruning channels: %w\n", err)
                }
        }

        log.Infof("Pruning %v zombie channels", len(chansToPrune))
        if len(chansToPrune) == 0 {
                return nil
        }

        // With the set of zombie-like channels obtained, we'll do another pass
        // to delete them from the channel graph.
        toPrune := make([]uint64, 0, len(chansToPrune))
        for chanID := range chansToPrune {
                toPrune = append(toPrune, chanID)
                log.Tracef("Pruning zombie channel with ChannelID(%v)", chanID)
        }
        err = r.cfg.Graph.DeleteChannelEdges(
                r.cfg.StrictZombiePruning, true, toPrune...,
        )
        if err != nil {
                return fmt.Errorf("unable to delete zombie channels: %w", err)
        }

        // With the channels pruned, we'll also attempt to prune any nodes that
        // were a part of them.
        err = r.cfg.Graph.PruneGraphNodes()
        if err != nil && !errors.Is(err, channeldb.ErrGraphNodesNotFound) {
                return fmt.Errorf("unable to prune graph nodes: %w", err)
        }

        return nil
}

// handleNetworkUpdate is responsible for processing the update message and
// notifies topology changes, if any.
//
// NOTE: must be run inside goroutine.
func (r *ChannelRouter) handleNetworkUpdate(vb *ValidationBarrier,
        update *routingMsg) {

        defer r.wg.Done()
        defer vb.CompleteJob()

        // If this message has an existing dependency, then we'll wait until
        // that has been fully validated before we proceed.
        err := vb.WaitForDependants(update.msg)
        if err != nil {
                switch {
                case IsError(err, ErrVBarrierShuttingDown):
                        update.err <- err

                case IsError(err, ErrParentValidationFailed):
                        update.err <- newErrf(ErrIgnored, err.Error())

                default:
                        log.Warnf("unexpected error during validation "+
                                "barrier shutdown: %v", err)
                        update.err <- err
                }

                return
        }

        // Process the routing update to determine if this is either a new
        // update from our PoV or an update to a prior vertex/edge we
        // previously accepted.
        err = r.processUpdate(update.msg, update.op...)
        update.err <- err

        // If this message had any dependencies, then we can now signal them to
        // continue.
        allowDependents := err == nil || IsError(err, ErrIgnored, ErrOutdated)
        vb.SignalDependants(update.msg, allowDependents)

        // If the error is not nil here, there's no need to send topology
        // change.
        if err != nil {
                // We now decide to log an error or not. If allowDependents is
                // false, it means there is an error and the error is neither
                // ErrIgnored nor ErrOutdated. In this case, we'll log an error.
                // Otherwise, we'll add debug log only.
                if allowDependents {
                        log.Debugf("process network updates got: %v", err)
                } else {
                        log.Errorf("process network updates got: %v", err)
                }

                return
        }

        // Otherwise, we'll send off a new notification for the newly accepted
        // update, if any.
        topChange := &TopologyChange{}
        err = addToTopologyChange(r.cfg.Graph, topChange, update.msg)
        if err != nil {
                log.Errorf("unable to update topology change notification: %v",
                        err)
                return
        }

        if !topChange.isEmpty() {
                r.notifyTopologyChange(topChange)
        }
}

// networkHandler is the primary goroutine for the ChannelRouter. The roles of
// this goroutine include answering queries related to the state of the
// network, pruning the graph on new block notification, applying network
// updates, and registering new topology clients.
//
// NOTE: This MUST be run as a goroutine.
func (r *ChannelRouter) networkHandler() {
        defer r.wg.Done()

        graphPruneTicker := time.NewTicker(r.cfg.GraphPruneInterval)
        defer graphPruneTicker.Stop()

        defer r.statTicker.Stop()

        r.stats.Reset()

        // We'll use this validation barrier to ensure that we process all jobs
        // in the proper order during parallel validation.
        //
        // NOTE: For AssumeChannelValid, we bump up the maximum number of
        // concurrent validation requests since there are no blocks being
        // fetched. This significantly increases the performance of IGD for
        // neutrino nodes.
        //
        // However, we dial back to use multiple of the number of cores when
        // fully validating, to avoid fetching up to 1000 blocks from the
        // backend. On bitcoind, this will empirically cause massive latency
        // spikes when executing this many concurrent RPC calls. Critical
        // subsystems or basic rpc calls that rely on calls such as GetBestBlock
        // will hang due to excessive load.
        //
        // See https://github.com/lightningnetwork/lnd/issues/4892.
        var validationBarrier *ValidationBarrier
        if r.cfg.AssumeChannelValid {
                validationBarrier = NewValidationBarrier(1000, r.quit)
        } else {
                validationBarrier = NewValidationBarrier(
                        4*runtime.NumCPU(), r.quit,
                )
        }

        for {

                // If there are stats, resume the statTicker.
                if !r.stats.Empty() {
                        r.statTicker.Resume()
                }

                select {
                // A new fully validated network update has just arrived. As a
                // result we'll modify the channel graph accordingly depending
                // on the exact type of the message.
                case update := <-r.networkUpdates:
                        // We'll set up any dependants, and wait until a free
                        // slot for this job opens up, this allows us to not
                        // have thousands of goroutines active.
                        validationBarrier.InitJobDependencies(update.msg)

                        r.wg.Add(1)
                        go r.handleNetworkUpdate(validationBarrier, update)

                        // TODO(roasbeef): remove all unconnected vertexes
                        // after N blocks pass with no corresponding
                        // announcements.

                case chainUpdate, ok := <-r.staleBlocks:
                        // If the channel has been closed, then this indicates
                        // the daemon is shutting down, so we exit ourselves.
                        if !ok {
                                return
                        }

                        // Since this block is stale, we update our best height
                        // to the previous block.
                        blockHeight := chainUpdate.Height
                        atomic.StoreUint32(&r.bestHeight, blockHeight-1)

                        // Update the channel graph to reflect that this block
                        // was disconnected.
                        _, err := r.cfg.Graph.DisconnectBlockAtHeight(blockHeight)
                        if err != nil {
                                log.Errorf("unable to prune graph with stale "+
                                        "block: %v", err)
                                continue
                        }

                        // TODO(halseth): notify client about the reorg?

                // A new block has arrived, so we can prune the channel graph
                // of any channels which were closed in the block.
                case chainUpdate, ok := <-r.newBlocks:
                        // If the channel has been closed, then this indicates
                        // the daemon is shutting down, so we exit ourselves.
                        if !ok {
                                return
                        }

                        // We'll ensure that any new blocks received attach
                        // directly to the end of our main chain. If not, then
                        // we've somehow missed some blocks. Here we'll catch
                        // up the chain with the latest blocks.
                        currentHeight := atomic.LoadUint32(&r.bestHeight)
                        switch {
                        case chainUpdate.Height == currentHeight+1:
                                err := r.updateGraphWithClosedChannels(
                                        chainUpdate,
                                )
                                if err != nil {
                                        log.Errorf("unable to prune graph "+
                                                "with closed channels: %v", err)
                                }

                        case chainUpdate.Height > currentHeight+1:
                                log.Errorf("out of order block: expecting "+
                                        "height=%v, got height=%v",
                                        currentHeight+1, chainUpdate.Height)

                                err := r.getMissingBlocks(currentHeight, chainUpdate)
                                if err != nil {
                                        log.Errorf("unable to retrieve missing"+
                                                "blocks: %v", err)
                                }

                        case chainUpdate.Height < currentHeight+1:
                                log.Errorf("out of order block: expecting "+
                                        "height=%v, got height=%v",
                                        currentHeight+1, chainUpdate.Height)

                                log.Infof("Skipping channel pruning since "+
                                        "received block height %v was already"+
                                        " processed.", chainUpdate.Height)
                        }

                // A new notification client update has arrived. We're either
                // gaining a new client, or cancelling notifications for an
                // existing client.
                case ntfnUpdate := <-r.ntfnClientUpdates:
                        clientID := ntfnUpdate.clientID

                        if ntfnUpdate.cancel {
                                client, ok := r.topologyClients.LoadAndDelete(
                                        clientID,
                                )
                                if ok {
                                        close(client.exit)
                                        client.wg.Wait()

                                        close(client.ntfnChan)
                                }

                                continue
                        }

                        r.topologyClients.Store(clientID, &topologyClient{
                                ntfnChan: ntfnUpdate.ntfnChan,
                                exit:     make(chan struct{}),
                        })

                // The graph prune ticker has ticked, so we'll examine the
                // state of the known graph to filter out any zombie channels
                // for pruning.
                case <-graphPruneTicker.C:
                        if err := r.pruneZombieChans(); err != nil {
                                log.Errorf("Unable to prune zombies: %v", err)
                        }

                // Log any stats if we've processed a non-empty number of
                // channels, updates, or nodes. We'll only pause the ticker if
                // the last window contained no updates to avoid resuming and
                // pausing while consecutive windows contain new info.
                case <-r.statTicker.Ticks():
                        if !r.stats.Empty() {
                                log.Infof(r.stats.String())
                        } else {
                                r.statTicker.Pause()
                        }
                        r.stats.Reset()

                // The router has been signalled to exit, to we exit our main
                // loop so the wait group can be decremented.
                case <-r.quit:
                        return
                }
        }
}

// getMissingBlocks walks through all missing blocks and updates the graph
// closed channels accordingly.
func (r *ChannelRouter) getMissingBlocks(currentHeight uint32,
        chainUpdate *chainview.FilteredBlock) error {

        outdatedHash, err := r.cfg.Chain.GetBlockHash(int64(currentHeight))
        if err != nil {
                return err
        }

        outdatedBlock := &chainntnfs.BlockEpoch{
                Height: int32(currentHeight),
                Hash:   outdatedHash,
        }

        epochClient, err := r.cfg.Notifier.RegisterBlockEpochNtfn(
                outdatedBlock,
        )
        if err != nil {
                return err
        }
        defer epochClient.Cancel()

        blockDifference := int(chainUpdate.Height - currentHeight)

        // We'll walk through all the outdated blocks and make sure we're able
        // to update the graph with any closed channels from them.
        for i := 0; i < blockDifference; i++ {
                var (
                        missingBlock *chainntnfs.BlockEpoch
                        ok           bool
                )

                select {
                case missingBlock, ok = <-epochClient.Epochs:
                        if !ok {
                                return nil
                        }

                case <-r.quit:
                        return nil
                }

                filteredBlock, err := r.cfg.ChainView.FilterBlock(
                        missingBlock.Hash,
                )
                if err != nil {
                        return err
                }

                err = r.updateGraphWithClosedChannels(
                        filteredBlock,
                )
                if err != nil {
                        return err
                }
        }

        return nil
}

// updateGraphWithClosedChannels prunes the channel graph of closed channels
// that are no longer needed.
func (r *ChannelRouter) updateGraphWithClosedChannels(
        chainUpdate *chainview.FilteredBlock) error {

        // Once a new block arrives, we update our running track of the height
        // of the chain tip.
        blockHeight := chainUpdate.Height

        atomic.StoreUint32(&r.bestHeight, blockHeight)
        log.Infof("Pruning channel graph using block %v (height=%v)",
                chainUpdate.Hash, blockHeight)

        // We're only interested in all prior outputs that have been spent in
        // the block, so collate all the referenced previous outpoints within
        // each tx and input.
        var spentOutputs []*wire.OutPoint
        for _, tx := range chainUpdate.Transactions {
                for _, txIn := range tx.TxIn {
                        spentOutputs = append(spentOutputs,
                                &txIn.PreviousOutPoint)
                }
        }

        // With the spent outputs gathered, attempt to prune the channel graph,
        // also passing in the hash+height of the block being pruned so the
        // prune tip can be updated.
        chansClosed, err := r.cfg.Graph.PruneGraph(spentOutputs,
                &chainUpdate.Hash, chainUpdate.Height)
        if err != nil {
                log.Errorf("unable to prune routing table: %v", err)
                return err
        }

        log.Infof("Block %v (height=%v) closed %v channels", chainUpdate.Hash,
                blockHeight, len(chansClosed))

        if len(chansClosed) == 0 {
                return err
        }

        // Notify all currently registered clients of the newly closed channels.
        closeSummaries := createCloseSummaries(blockHeight, chansClosed...)
        r.notifyTopologyChange(&TopologyChange{
                ClosedChannels: closeSummaries,
        })

        return nil
}

// assertNodeAnnFreshness returns a non-nil error if we have an announcement in
// the database for the passed node with a timestamp newer than the passed
// timestamp. ErrIgnored will be returned if we already have the node, and
// ErrOutdated will be returned if we have a timestamp that's after the new
// timestamp.
func (r *ChannelRouter) assertNodeAnnFreshness(node route.Vertex,
        msgTimestamp time.Time) error {

        // If we are not already aware of this node, it means that we don't
        // know about any channel using this node. To avoid a DoS attack by
        // node announcements, we will ignore such nodes. If we do know about
        // this node, check that this update brings info newer than what we
        // already have.
        lastUpdate, exists, err := r.cfg.Graph.HasLightningNode(node)
        if err != nil {
                return errors.Errorf("unable to query for the "+
                        "existence of node: %v", err)
        }
        if !exists {
                return newErrf(ErrIgnored, "Ignoring node announcement"+
                        " for node not found in channel graph (%x)",
                        node[:])
        }

        // If we've reached this point then we're aware of the vertex being
        // advertised. So we now check if the new message has a new time stamp,
        // if not then we won't accept the new data as it would override newer
        // data.
        if !lastUpdate.Before(msgTimestamp) {
                return newErrf(ErrOutdated, "Ignoring outdated "+
                        "announcement for %x", node[:])
        }

        return nil
}

// addZombieEdge adds a channel that failed complete validation into the zombie
// index, so we can avoid having to re-validate it in the future.
func (r *ChannelRouter) addZombieEdge(chanID uint64) error {
        // If the edge fails validation we'll mark the edge itself as a zombie,
        // so we don't continue to request it. We use the "zero key" for both
        // node pubkeys so this edge can't be resurrected.
        var zeroKey [33]byte
        err := r.cfg.Graph.MarkEdgeZombie(chanID, zeroKey, zeroKey)
        if err != nil {
                return fmt.Errorf("unable to mark spent chan(id=%v) as a "+
                        "zombie: %w", chanID, err)
        }

        return nil
}

// makeFundingScript is used to make the funding script for both segwit v0 and
// segwit v1 (taproot) channels.
//
// TODO(roasbeef: export and use elsewhere?
func makeFundingScript(bitcoinKey1, bitcoinKey2 []byte,
        chanFeatures []byte) ([]byte, error) {

        legacyFundingScript := func() ([]byte, error) {
                witnessScript, err := input.GenMultiSigScript(
                        bitcoinKey1, bitcoinKey2,
                )
                if err != nil {
                        return nil, err
                }
                pkScript, err := input.WitnessScriptHash(witnessScript)
                if err != nil {
                        return nil, err
                }

                return pkScript, nil
        }

        if len(chanFeatures) == 0 {
                return legacyFundingScript()
        }

        // In order to make the correct funding script, we'll need to parse the
        // chanFeatures bytes into a feature vector we can interact with.
        rawFeatures := lnwire.NewRawFeatureVector()
        err := rawFeatures.Decode(bytes.NewReader(chanFeatures))
        if err != nil {
                return nil, fmt.Errorf("unable to parse chan feature "+
                        "bits: %w", err)
        }

        chanFeatureBits := lnwire.NewFeatureVector(
                rawFeatures, lnwire.Features,
        )
        if chanFeatureBits.HasFeature(
                lnwire.SimpleTaprootChannelsOptionalStaging,
        ) {

                pubKey1, err := btcec.ParsePubKey(bitcoinKey1)
                if err != nil {
                        return nil, err
                }
                pubKey2, err := btcec.ParsePubKey(bitcoinKey2)
                if err != nil {
                        return nil, err
                }

                fundingScript, _, err := input.GenTaprootFundingScript(
                        pubKey1, pubKey2, 0,
                )
                if err != nil {
                        return nil, err
                }

                return fundingScript, nil
        }

        return legacyFundingScript()
}

// processUpdate processes a new relate authenticated channel/edge, node or
// channel/edge update network update. If the update didn't affect the internal
// state of the draft due to either being out of date, invalid, or redundant,
// then error is returned.
func (r *ChannelRouter) processUpdate(msg interface{},
        op ...batch.SchedulerOption) error {

        switch msg := msg.(type) {
        case *channeldb.LightningNode:
                // Before we add the node to the database, we'll check to see
                // if the announcement is "fresh" or not. If it isn't, then
                // we'll return an error.
                err := r.assertNodeAnnFreshness(msg.PubKeyBytes, msg.LastUpdate)
                if err != nil {
                        return err
                }

                if err := r.cfg.Graph.AddLightningNode(msg, op...); err != nil {
                        return errors.Errorf("unable to add node %x to the "+
                                "graph: %v", msg.PubKeyBytes, err)
                }

                log.Tracef("Updated vertex data for node=%x", msg.PubKeyBytes)
                r.stats.incNumNodeUpdates()

        case *models.ChannelEdgeInfo:
                log.Debugf("Received ChannelEdgeInfo for channel %v",
                        msg.ChannelID)

                // Prior to processing the announcement we first check if we
                // already know of this channel, if so, then we can exit early.
                _, _, exists, isZombie, err := r.cfg.Graph.HasChannelEdge(
                        msg.ChannelID,
                )
                if err != nil && !errors.Is(
                        err, channeldb.ErrGraphNoEdgesFound,
                ) {

                        return errors.Errorf("unable to check for edge "+
                                "existence: %v", err)
                }
                if isZombie {
                        return newErrf(ErrIgnored, "ignoring msg for zombie "+
                                "chan_id=%v", msg.ChannelID)
                }
                if exists {
                        return newErrf(ErrIgnored, "ignoring msg for known "+
                                "chan_id=%v", msg.ChannelID)
                }

                // If AssumeChannelValid is present, then we are unable to
                // perform any of the expensive checks below, so we'll
                // short-circuit our path straight to adding the edge to our
                // graph. If the passed ShortChannelID is an alias, then we'll
                // skip validation as it will not map to a legitimate tx. This
                // is not a DoS vector as only we can add an alias
                // ChannelAnnouncement from the gossiper.
                scid := lnwire.NewShortChanIDFromInt(msg.ChannelID)
                if r.cfg.AssumeChannelValid || r.cfg.IsAlias(scid) {
                        if err := r.cfg.Graph.AddChannelEdge(msg, op...); err != nil {
                                return fmt.Errorf("unable to add edge: %w", err)
                        }
                        log.Tracef("New channel discovered! Link "+
                                "connects %x and %x with ChannelID(%v)",
                                msg.NodeKey1Bytes, msg.NodeKey2Bytes,
                                msg.ChannelID)
                        r.stats.incNumEdgesDiscovered()

                        break
                }

                // Before we can add the channel to the channel graph, we need
                // to obtain the full funding outpoint that's encoded within
                // the channel ID.
                channelID := lnwire.NewShortChanIDFromInt(msg.ChannelID)
                fundingTx, err := r.fetchFundingTxWrapper(&channelID)
                if err != nil {
                        // In order to ensure we don't erroneously mark a
                        // channel as a zombie due to an RPC failure, we'll
                        // attempt to string match for the relevant errors.
                        //
                        // * btcd:
                        //    * https://github.com/btcsuite/btcd/blob/master/rpcserver.go#L1316
                        //    * https://github.com/btcsuite/btcd/blob/master/rpcserver.go#L1086
                        // * bitcoind:
                        //    * https://github.com/bitcoin/bitcoin/blob/7fcf53f7b4524572d1d0c9a5fdc388e87eb02416/src/rpc/blockchain.cpp#L770
                        //     * https://github.com/bitcoin/bitcoin/blob/7fcf53f7b4524572d1d0c9a5fdc388e87eb02416/src/rpc/blockchain.cpp#L954
                        switch {
                        case strings.Contains(err.Error(), "not found"):
                                fallthrough

                        case strings.Contains(err.Error(), "out of range"):
                                // If the funding transaction isn't found at
                                // all, then we'll mark the edge itself as a
                                // zombie, so we don't continue to request it.
                                // We use the "zero key" for both node pubkeys
                                // so this edge can't be resurrected.
                                zErr := r.addZombieEdge(msg.ChannelID)
                                if zErr != nil {
                                        return zErr
                                }

                        default:
                        }

                        return newErrf(ErrNoFundingTransaction, "unable to "+
                                "locate funding tx: %v", err)
                }

                // Recreate witness output to be sure that declared in channel
                // edge bitcoin keys and channel value corresponds to the
                // reality.
                fundingPkScript, err := makeFundingScript(
                        msg.BitcoinKey1Bytes[:], msg.BitcoinKey2Bytes[:],
                        msg.Features,
                )
                if err != nil {
                        return err
                }

                // Next we'll validate that this channel is actually
                // well-formed. If this check fails, then this channel either
                // doesn't exist, or isn't the one that was meant to be created
                // according to the passed channel proofs.
                fundingPoint, err := chanvalidate.Validate(&chanvalidate.Context{
                        Locator: &chanvalidate.ShortChanIDChanLocator{
                                ID: channelID,
                        },
                        MultiSigPkScript: fundingPkScript,
                        FundingTx:        fundingTx,
                })
                if err != nil {
                        // Mark the edge as a zombie, so we won't try to
                        // re-validate it on start up.
                        if err := r.addZombieEdge(msg.ChannelID); err != nil {
                                return err
                        }

                        return newErrf(ErrInvalidFundingOutput, "output "+
                                "failed validation: %w", err)
                }

                // Now that we have the funding outpoint of the channel, ensure
                // that it hasn't yet been spent. If so, then this channel has
                // been closed, so we'll ignore it.
                chanUtxo, err := r.cfg.Chain.GetUtxo(
                        fundingPoint, fundingPkScript, channelID.BlockHeight,
                        r.quit,
                )
                if err != nil {
                        if errors.Is(err, btcwallet.ErrOutputSpent) {
                                zErr := r.addZombieEdge(msg.ChannelID)
                                if zErr != nil {
                                        return zErr
                                }
                        }

                        return newErrf(ErrChannelSpent, "unable to fetch utxo "+
                                "for chan_id=%v, chan_point=%v: %v",
                                msg.ChannelID, fundingPoint, err)
                }

                // TODO(roasbeef): this is a hack, needs to be removed
                // after commitment fees are dynamic.
                msg.Capacity = btcutil.Amount(chanUtxo.Value)
                msg.ChannelPoint = *fundingPoint
                if err := r.cfg.Graph.AddChannelEdge(msg, op...); err != nil {
                        return errors.Errorf("unable to add edge: %v", err)
                }

                log.Debugf("New channel discovered! Link "+
                        "connects %x and %x with ChannelPoint(%v): "+
                        "chan_id=%v, capacity=%v",
                        msg.NodeKey1Bytes, msg.NodeKey2Bytes,
                        fundingPoint, msg.ChannelID, msg.Capacity)
                r.stats.incNumEdgesDiscovered()

                // As a new edge has been added to the channel graph, we'll
                // update the current UTXO filter within our active
                // FilteredChainView, so we are notified if/when this channel is
                // closed.
                filterUpdate := []channeldb.EdgePoint{
                        {
                                FundingPkScript: fundingPkScript,
                                OutPoint:        *fundingPoint,
                        },
                }
                err = r.cfg.ChainView.UpdateFilter(
                        filterUpdate, atomic.LoadUint32(&r.bestHeight),
                )
                if err != nil {
                        return errors.Errorf("unable to update chain "+
                                "view: %v", err)
                }

        case *models.ChannelEdgePolicy:
                log.Debugf("Received ChannelEdgePolicy for channel %v",
                        msg.ChannelID)

                // We make sure to hold the mutex for this channel ID,
                // such that no other goroutine is concurrently doing
                // database accesses for the same channel ID.
                r.channelEdgeMtx.Lock(msg.ChannelID)
                defer r.channelEdgeMtx.Unlock(msg.ChannelID)

                edge1Timestamp, edge2Timestamp, exists, isZombie, err :=
                        r.cfg.Graph.HasChannelEdge(msg.ChannelID)
                if err != nil && !errors.Is(
                        err, channeldb.ErrGraphNoEdgesFound,
                ) {

                        return errors.Errorf("unable to check for edge "+
                                "existence: %v", err)

                }

                // If the channel is marked as a zombie in our database, and
                // we consider this a stale update, then we should not apply the
                // policy.
                isStaleUpdate := time.Since(msg.LastUpdate) > r.cfg.ChannelPruneExpiry
                if isZombie && isStaleUpdate {
                        return newErrf(ErrIgnored, "ignoring stale update "+
                                "(flags=%v|%v) for zombie chan_id=%v",
                                msg.MessageFlags, msg.ChannelFlags,
                                msg.ChannelID)
                }

                // If the channel doesn't exist in our database, we cannot
                // apply the updated policy.
                if !exists {
                        return newErrf(ErrIgnored, "ignoring update "+
                                "(flags=%v|%v) for unknown chan_id=%v",
                                msg.MessageFlags, msg.ChannelFlags,
                                msg.ChannelID)
                }

                // As edges are directional edge node has a unique policy for
                // the direction of the edge they control. Therefore, we first
                // check if we already have the most up-to-date information for
                // that edge. If this message has a timestamp not strictly
                // newer than what we already know of we can exit early.
                switch {

                // A flag set of 0 indicates this is an announcement for the
                // "first" node in the channel.
                case msg.ChannelFlags&lnwire.ChanUpdateDirection == 0:

                        // Ignore outdated message.
                        if !edge1Timestamp.Before(msg.LastUpdate) {
                                return newErrf(ErrOutdated, "Ignoring "+
                                        "outdated update (flags=%v|%v) for "+
                                        "known chan_id=%v", msg.MessageFlags,
                                        msg.ChannelFlags, msg.ChannelID)
                        }

                // Similarly, a flag set of 1 indicates this is an announcement
                // for the "second" node in the channel.
                case msg.ChannelFlags&lnwire.ChanUpdateDirection == 1:

                        // Ignore outdated message.
                        if !edge2Timestamp.Before(msg.LastUpdate) {
                                return newErrf(ErrOutdated, "Ignoring "+
                                        "outdated update (flags=%v|%v) for "+
                                        "known chan_id=%v", msg.MessageFlags,
                                        msg.ChannelFlags, msg.ChannelID)
                        }
                }

                // Now that we know this isn't a stale update, we'll apply the
                // new edge policy to the proper directional edge within the
                // channel graph.
                if err = r.cfg.Graph.UpdateEdgePolicy(msg, op...); err != nil {
                        err := errors.Errorf("unable to add channel: %v", err)
                        log.Error(err)
                        return err
                }

                log.Tracef("New channel update applied: %v",
                        newLogClosure(func() string { return spew.Sdump(msg) }))
                r.stats.incNumChannelUpdates()

        default:
                return errors.Errorf("wrong routing update message type")
        }

        return nil
}

// fetchFundingTxWrapper is a wrapper around fetchFundingTx, except that it
// will exit if the router has stopped.
func (r *ChannelRouter) fetchFundingTxWrapper(chanID *lnwire.ShortChannelID) (
        *wire.MsgTx, error) {

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

        go func() {
                tx, err := r.fetchFundingTx(chanID)
                if err != nil {
                        errChan <- err
                        return
                }

                txChan <- tx
        }()

        select {
        case tx := <-txChan:
                return tx, nil

        case err := <-errChan:
                return nil, err

        case <-r.quit:
                return nil, ErrRouterShuttingDown
        }
}

// fetchFundingTx returns the funding transaction identified by the passed
// short channel ID.
//
// TODO(roasbeef): replace with call to GetBlockTransaction? (would allow to
// later use getblocktxn)
func (r *ChannelRouter) fetchFundingTx(
        chanID *lnwire.ShortChannelID) (*wire.MsgTx, error) {

        // First fetch the block hash by the block number encoded, then use
        // that hash to fetch the block itself.
        blockNum := int64(chanID.BlockHeight)
        blockHash, err := r.cfg.Chain.GetBlockHash(blockNum)
        if err != nil {
                return nil, err
        }
        fundingBlock, err := r.cfg.Chain.GetBlock(blockHash)
        if err != nil {
                return nil, err
        }

        // As a sanity check, ensure that the advertised transaction index is
        // within the bounds of the total number of transactions within a
        // block.
        numTxns := uint32(len(fundingBlock.Transactions))
        if chanID.TxIndex > numTxns-1 {
                return nil, fmt.Errorf("tx_index=#%v "+
                        "is out of range (max_index=%v), network_chan_id=%v",
                        chanID.TxIndex, numTxns-1, chanID)
        }

        return fundingBlock.Transactions[chanID.TxIndex].Copy(), nil
}

// routingMsg couples a routing related routing topology update to the
// error channel.
type routingMsg struct {
        msg interface{}
        op  []batch.SchedulerOption
        err chan error
}

// RouteRequest contains the parameters for a pathfinding request. It may
// describe a request to make a regular payment or one to a blinded path
// (incdicated by a non-nil BlindedPayment field).
type RouteRequest struct {
        // Source is the node that the path originates from.
        Source route.Vertex

        // Target is the node that the path terminates at. If the route
        // includes a blinded path, target will be the blinded node id of the
        // final hop in the blinded route.
        Target route.Vertex

        // Amount is the Amount in millisatoshis to be delivered to the target
        // node.
        Amount lnwire.MilliSatoshi

        // TimePreference expresses the caller's time preference for
        // pathfinding.
        TimePreference float64

        // Restrictions provides a set of additional Restrictions that the
        // route must adhere to.
        Restrictions *RestrictParams

        // CustomRecords is a set of custom tlv records to include for the
        // final hop.
        CustomRecords record.CustomSet

        // RouteHints contains an additional set of edges to include in our
        // view of the graph. This may either be a set of hints for private
        // channels or a "virtual" hop hint that represents a blinded route.
        RouteHints RouteHints

        // FinalExpiry is the cltv delta for the final hop. If paying to a
        // blinded path, this value is a duplicate of the delta provided
        // in blinded payment.
        FinalExpiry uint16

        // BlindedPayment contains an optional blinded path and parameters
        // used to reach a target node via a blinded path. This field is
        // mutually exclusive with the Target field.
        BlindedPayment *BlindedPayment
}

// RouteHints is an alias type for a set of route hints, with the source node
// as the map's key and the details of the hint(s) in the edge policy.
type RouteHints map[route.Vertex][]AdditionalEdge

// NewRouteRequest produces a new route request for a regular payment or one
// to a blinded route, validating that the target, routeHints and finalExpiry
// parameters are mutually exclusive with the blindedPayment parameter (which
// contains these values for blinded payments).
func NewRouteRequest(source route.Vertex, target *route.Vertex,
        amount lnwire.MilliSatoshi, timePref float64,
        restrictions *RestrictParams, customRecords record.CustomSet,
        routeHints RouteHints, blindedPayment *BlindedPayment,
        finalExpiry uint16) (*RouteRequest, error) {

        var (
                // Assume that we're starting off with a regular payment.
                requestHints  = routeHints
                requestExpiry = finalExpiry
                err           error
        )

        if blindedPayment != nil {
                if err := blindedPayment.Validate(); err != nil {
                        return nil, fmt.Errorf("invalid blinded payment: %w",
                                err)
                }

                introVertex := route.NewVertex(
                        blindedPayment.BlindedPath.IntroductionPoint,
                )
                if source == introVertex {
                        return nil, ErrSelfIntro
                }

                // Check that the values for a clear path have not been set,
                // as this is an ambiguous signal from the caller.
                if routeHints != nil {
                        return nil, ErrHintsAndBlinded
                }

                if finalExpiry != 0 {
                        return nil, ErrExpiryAndBlinded
                }

                // If we have a blinded path with 1 hop, the cltv expiry
                // will not be included in any hop hints (since we're just
                // sending to the introduction node and need no blinded hints).
                // In this case, we include it to make sure that the final
                // cltv delta is accounted for (since it's part of the blinded
                // delta). In the case of a multi-hop route, we set our final
                // cltv to zero, since it's going to be accounted for in the
                // delta for our hints.
                if len(blindedPayment.BlindedPath.BlindedHops) == 1 {
                        requestExpiry = blindedPayment.CltvExpiryDelta
                }

                requestHints, err = blindedPayment.toRouteHints()
                if err != nil {
                        return nil, err
                }
        }

        requestTarget, err := getTargetNode(target, blindedPayment)
        if err != nil {
                return nil, err
        }

        return &RouteRequest{
                Source:         source,
                Target:         requestTarget,
                Amount:         amount,
                TimePreference: timePref,
                Restrictions:   restrictions,
                CustomRecords:  customRecords,
                RouteHints:     requestHints,
                FinalExpiry:    requestExpiry,
                BlindedPayment: blindedPayment,
        }, nil
}

func getTargetNode(target *route.Vertex, blindedPayment *BlindedPayment) (
        route.Vertex, error) {

        var (
                blinded   = blindedPayment != nil
                targetSet = target != nil
        )

        switch {
        case blinded && targetSet:
                return route.Vertex{}, ErrTargetAndBlinded

        case blinded:
                // If we're dealing with an edge-case blinded path that just
                // has an introduction node (first hop expected to be the intro
                // hop), then we return the unblinded introduction node as our
                // target.
                hops := blindedPayment.BlindedPath.BlindedHops
                if len(hops) == 1 {
                        return route.NewVertex(
                                blindedPayment.BlindedPath.IntroductionPoint,
                        ), nil
                }

                return route.NewVertex(hops[len(hops)-1].BlindedNodePub), nil

        case targetSet:
                return *target, nil

        default:
                return route.Vertex{}, ErrNoTarget
        }
}

// blindedPath returns the request's blinded path, which is set if the payment
// is to a blinded route.
func (r *RouteRequest) blindedPath() *sphinx.BlindedPath {
        if r.BlindedPayment == nil {
                return nil
        }

        return r.BlindedPayment.BlindedPath
}

// FindRoute attempts to query the ChannelRouter for the optimum path to a
// particular target destination to which it is able to send `amt` after
// factoring in channel capacities and cumulative fees along the route.
func (r *ChannelRouter) FindRoute(req *RouteRequest) (*route.Route, float64,
        error) {

        log.Debugf("Searching for path to %v, sending %v", req.Target,
                req.Amount)

        // We'll attempt to obtain a set of bandwidth hints that can help us
        // eliminate certain routes early on in the path finding process.
        bandwidthHints, err := newBandwidthManager(
                r.cachedGraph, r.selfNode.PubKeyBytes, r.cfg.GetLink,
        )
        if err != nil {
                return nil, 0, err
        }

        // We'll fetch the current block height, so we can properly calculate
        // the required HTLC time locks within the route.
        _, currentHeight, err := r.cfg.Chain.GetBestBlock()
        if err != nil {
                return nil, 0, err
        }

        // Now that we know the destination is reachable within the graph, we'll
        // execute our path finding algorithm.
        finalHtlcExpiry := currentHeight + int32(req.FinalExpiry)

        // Validate time preference.
        timePref := req.TimePreference
        if timePref < -1 || timePref > 1 {
                return nil, 0, errors.New("time preference out of range")
        }

        path, probability, err := findPath(
                &graphParams{
                        additionalEdges: req.RouteHints,
                        bandwidthHints:  bandwidthHints,
                        graph:           r.cachedGraph,
                },
                req.Restrictions, &r.cfg.PathFindingConfig, req.Source,
                req.Target, req.Amount, req.TimePreference, finalHtlcExpiry,
        )
        if err != nil {
                return nil, 0, err
        }

        // Create the route with absolute time lock values.
        route, err := newRoute(
                req.Source, path, uint32(currentHeight),
                finalHopParams{
                        amt:       req.Amount,
                        totalAmt:  req.Amount,
                        cltvDelta: req.FinalExpiry,
                        records:   req.CustomRecords,
                }, req.blindedPath(),
        )
        if err != nil {
                return nil, 0, err
        }

        go log.Tracef("Obtained path to send %v to %x: %v",
                req.Amount, req.Target, newLogClosure(func() string {
                        return spew.Sdump(route)
                }),
        )

        return route, probability, nil
}

// generateNewSessionKey generates a new ephemeral private key to be used for a
// payment attempt.
func generateNewSessionKey() (*btcec.PrivateKey, error) {
        // Generate a new random session key to ensure that we don't trigger
        // any replay.
        //
        // TODO(roasbeef): add more sources of randomness?
        return btcec.NewPrivateKey()
}

// generateSphinxPacket generates then encodes a sphinx packet which encodes
// the onion route specified by the passed layer 3 route. The blob returned
// from this function can immediately be included within an HTLC add packet to
// be sent to the first hop within the route.
func generateSphinxPacket(rt *route.Route, paymentHash []byte,
        sessionKey *btcec.PrivateKey) ([]byte, *sphinx.Circuit, error) {

        // Now that we know we have an actual route, we'll map the route into a
        // sphinx payment path which includes per-hop payloads for each hop
        // that give each node within the route the necessary information
        // (fees, CLTV value, etc.) to properly forward the payment.
        sphinxPath, err := rt.ToSphinxPath()
        if err != nil {
                return nil, nil, err
        }

        log.Tracef("Constructed per-hop payloads for payment_hash=%x: %v",
                paymentHash[:], newLogClosure(func() string {
                        path := make(
                                []sphinx.OnionHop, sphinxPath.TrueRouteLength(),
                        )
                        for i := range path {
                                hopCopy := sphinxPath[i]
                                path[i] = hopCopy
                        }
                        return spew.Sdump(path)
                }),
        )

        // Next generate the onion routing packet which allows us to perform
        // privacy preserving source routing across the network.
        sphinxPacket, err := sphinx.NewOnionPacket(
                sphinxPath, sessionKey, paymentHash,
                sphinx.DeterministicPacketFiller,
        )
        if err != nil {
                return nil, nil, err
        }

        // Finally, encode Sphinx packet using its wire representation to be
        // included within the HTLC add packet.
        var onionBlob bytes.Buffer
        if err := sphinxPacket.Encode(&onionBlob); err != nil {
                return nil, nil, err
        }

        log.Tracef("Generated sphinx packet: %v",
                newLogClosure(func() string {
                        // We make a copy of the ephemeral key and unset the
                        // internal curve here in order to keep the logs from
                        // getting noisy.
                        key := *sphinxPacket.EphemeralKey
                        packetCopy := *sphinxPacket
                        packetCopy.EphemeralKey = &key
                        return spew.Sdump(packetCopy)
                }),
        )

        return onionBlob.Bytes(), &sphinx.Circuit{
                SessionKey:  sessionKey,
                PaymentPath: sphinxPath.NodeKeys(),
        }, nil
}

// LightningPayment describes a payment to be sent through the network to the
// final destination.
type LightningPayment struct {
        // Target is the node in which the payment should be routed towards.
        Target route.Vertex

        // Amount is the value of the payment to send through the network in
        // milli-satoshis.
        Amount lnwire.MilliSatoshi

        // FeeLimit is the maximum fee in millisatoshis that the payment should
        // accept when sending it through the network. The payment will fail
        // if there isn't a route with lower fees than this limit.
        FeeLimit lnwire.MilliSatoshi

        // CltvLimit is the maximum time lock that is allowed for attempts to
        // complete this payment.
        CltvLimit uint32

        // paymentHash is the r-hash value to use within the HTLC extended to
        // the first hop. This won't be set for AMP payments.
        paymentHash *lntypes.Hash

        // amp is an optional field that is set if and only if this is am AMP
        // payment.
        amp *AMPOptions

        // FinalCLTVDelta is the CTLV expiry delta to use for the _final_ hop
        // in the route. This means that the final hop will have a CLTV delta
        // of at least: currentHeight + FinalCLTVDelta.
        FinalCLTVDelta uint16

        // PayAttemptTimeout is a timeout value that we'll use to determine
        // when we should should abandon the payment attempt after consecutive
        // payment failure. This prevents us from attempting to send a payment
        // indefinitely. A zero value means the payment will never time out.
        //
        // TODO(halseth): make wallclock time to allow resume after startup.
        PayAttemptTimeout time.Duration

        // RouteHints represents the different routing hints that can be used to
        // assist a payment in reaching its destination successfully. These
        // hints will act as intermediate hops along the route.
        //
        // NOTE: This is optional unless required by the payment. When providing
        // multiple routes, ensure the hop hints within each route are chained
        // together and sorted in forward order in order to reach the
        // destination successfully.
        RouteHints [][]zpay32.HopHint

        // OutgoingChannelIDs is the list of channels that are allowed for the
        // first hop. If nil, any channel may be used.
        OutgoingChannelIDs []uint64

        // LastHop is the pubkey of the last node before the final destination
        // is reached. If nil, any node may be used.
        LastHop *route.Vertex

        // DestFeatures specifies the set of features we assume the final node
        // has for pathfinding. Typically, these will be taken directly from an
        // invoice, but they can also be manually supplied or assumed by the
        // sender. If a nil feature vector is provided, the router will try to
        // fall back to the graph in order to load a feature vector for a node
        // in the public graph.
        DestFeatures *lnwire.FeatureVector

        // PaymentAddr is the payment address specified by the receiver. This
        // field should be a random 32-byte nonce presented in the receiver's
        // invoice to prevent probing of the destination.
        PaymentAddr *[32]byte

        // PaymentRequest is an optional payment request that this payment is
        // attempting to complete.
        PaymentRequest []byte

        // DestCustomRecords are TLV records that are to be sent to the final
        // hop in the new onion payload format. If the destination does not
        // understand this new onion payload format, then the payment will
        // fail.
        DestCustomRecords record.CustomSet

        // MaxParts is the maximum number of partial payments that may be used
        // to complete the full amount.
        MaxParts uint32

        // MaxShardAmt is the largest shard that we'll attempt to split using.
        // If this field is set, and we need to split, rather than attempting
        // half of the original payment amount, we'll use this value if half
        // the payment amount is greater than it.
        //
        // NOTE: This field is _optional_.
        MaxShardAmt *lnwire.MilliSatoshi

        // TimePref is the time preference for this payment. Set to -1 to
        // optimize for fees only, to 1 to optimize for reliability only or a
        // value in between for a mix.
        TimePref float64

        // Metadata is additional data that is sent along with the payment to
        // the payee.
        Metadata []byte
}

// AMPOptions houses information that must be known in order to send an AMP
// payment.
type AMPOptions struct {
        SetID     [32]byte
        RootShare [32]byte
}

// SetPaymentHash sets the given hash as the payment's overall hash. This
// should only be used for non-AMP payments.
func (l *LightningPayment) SetPaymentHash(hash lntypes.Hash) error {
        if l.amp != nil {
                return fmt.Errorf("cannot set payment hash for AMP payment")
        }

        l.paymentHash = &hash
        return nil
}

// SetAMP sets the given AMP options for the payment.
func (l *LightningPayment) SetAMP(amp *AMPOptions) error {
        if l.paymentHash != nil {
                return fmt.Errorf("cannot set amp options for payment " +
                        "with payment hash")
        }

        l.amp = amp
        return nil
}

// Identifier returns a 32-byte slice that uniquely identifies this single
// payment. For non-AMP payments this will be the payment hash, for AMP
// payments this will be the used SetID.
func (l *LightningPayment) Identifier() [32]byte {
        if l.amp != nil {
                return l.amp.SetID
        }

        return *l.paymentHash
}

// SendPayment attempts to send a payment as described within the passed
// LightningPayment. This function is blocking and will return either: when the
// payment is successful, or all candidates routes have been attempted and
// resulted in a failed payment. If the payment succeeds, then a non-nil Route
// will be returned which describes the path the successful payment traversed
// within the network to reach the destination. Additionally, the payment
// preimage will also be returned.
func (r *ChannelRouter) SendPayment(payment *LightningPayment) ([32]byte,
        *route.Route, error) {

        paySession, shardTracker, err := r.PreparePayment(payment)
        if err != nil {
                return [32]byte{}, nil, err
        }

        log.Tracef("Dispatching SendPayment for lightning payment: %v",
                spewPayment(payment))

        return r.sendPayment(
                context.Background(), payment.FeeLimit, payment.Identifier(),
                payment.PayAttemptTimeout, paySession, shardTracker,
        )
}

// SendPaymentAsync is the non-blocking version of SendPayment. The payment
// result needs to be retrieved via the control tower.
func (r *ChannelRouter) SendPaymentAsync(ctx context.Context,
        payment *LightningPayment, ps PaymentSession, st shards.ShardTracker) {

        // Since this is the first time this payment is being made, we pass nil
        // for the existing attempt.
        r.wg.Add(1)
        go func() {
                defer r.wg.Done()

                log.Tracef("Dispatching SendPayment for lightning payment: %v",
                        spewPayment(payment))

                _, _, err := r.sendPayment(
                        ctx, payment.FeeLimit, payment.Identifier(),
                        payment.PayAttemptTimeout, ps, st,
                )
                if err != nil {
                        log.Errorf("Payment %x failed: %v",
                                payment.Identifier(), err)
                }
        }()
}

// spewPayment returns a log closures that provides a spewed string
// representation of the passed payment.
func spewPayment(payment *LightningPayment) logClosure {
        return newLogClosure(func() string {
                // Make a copy of the payment with a nilled Curve
                // before spewing.
                var routeHints [][]zpay32.HopHint
                for _, routeHint := range payment.RouteHints {
                        var hopHints []zpay32.HopHint
                        for _, hopHint := range routeHint {
                                h := hopHint.Copy()
                                hopHints = append(hopHints, h)
                        }
                        routeHints = append(routeHints, hopHints)
                }
                p := *payment
                p.RouteHints = routeHints
                return spew.Sdump(p)
        })
}

// PreparePayment creates the payment session and registers the payment with the
// control tower.
func (r *ChannelRouter) PreparePayment(payment *LightningPayment) (
        PaymentSession, shards.ShardTracker, error) {

        // Before starting the HTLC routing attempt, we'll create a fresh
        // payment session which will report our errors back to mission
        // control.
        paySession, err := r.cfg.SessionSource.NewPaymentSession(payment)
        if err != nil {
                return nil, nil, err
        }

        // Record this payment hash with the ControlTower, ensuring it is not
        // already in-flight.
        //
        // TODO(roasbeef): store records as part of creation info?
        info := &channeldb.PaymentCreationInfo{
                PaymentIdentifier: payment.Identifier(),
                Value:             payment.Amount,
                CreationTime:      r.cfg.Clock.Now(),
                PaymentRequest:    payment.PaymentRequest,
        }

        // Create a new ShardTracker that we'll use during the life cycle of
        // this payment.
        var shardTracker shards.ShardTracker
        switch {
        // If this is an AMP payment, we'll use the AMP shard tracker.
        case payment.amp != nil:
                shardTracker = amp.NewShardTracker(
                        payment.amp.RootShare, payment.amp.SetID,
                        *payment.PaymentAddr, payment.Amount,
                )

        // Otherwise we'll use the simple tracker that will map each attempt to
        // the same payment hash.
        default:
                shardTracker = shards.NewSimpleShardTracker(
                        payment.Identifier(), nil,
                )
        }

        err = r.cfg.Control.InitPayment(payment.Identifier(), info)
        if err != nil {
                return nil, nil, err
        }

        return paySession, shardTracker, nil
}

// SendToRoute sends a payment using the provided route and fails the payment
// when an error is returned from the attempt.
func (r *ChannelRouter) SendToRoute(htlcHash lntypes.Hash,
        rt *route.Route) (*channeldb.HTLCAttempt, error) {

        return r.sendToRoute(htlcHash, rt, false)
}

// SendToRouteSkipTempErr sends a payment using the provided route and fails
// the payment ONLY when a terminal error is returned from the attempt.
func (r *ChannelRouter) SendToRouteSkipTempErr(htlcHash lntypes.Hash,
        rt *route.Route) (*channeldb.HTLCAttempt, error) {

        return r.sendToRoute(htlcHash, rt, true)
}

// sendToRoute attempts to send a payment with the given hash through the
// provided route. This function is blocking and will return the attempt
// information as it is stored in the database. For a successful htlc, this
// information will contain the preimage. If an error occurs after the attempt
// was initiated, both return values will be non-nil. If skipTempErr is true,
// the payment won't be failed unless a terminal error has occurred.
func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
        skipTempErr bool) (*channeldb.HTLCAttempt, error) {

        // Calculate amount paid to receiver.
        amt := rt.ReceiverAmt()

        // If this is meant as an MP payment shard, we set the amount for the
        // creating info to the total amount of the payment.
        finalHop := rt.Hops[len(rt.Hops)-1]
        mpp := finalHop.MPP
        if mpp != nil {
                amt = mpp.TotalMsat()
        }

        // For non-MPP, there's no such thing as temp error as there's only one
        // HTLC attempt being made. When this HTLC is failed, the payment is
        // failed hence cannot be retried.
        if skipTempErr && mpp == nil {
                return nil, ErrSkipTempErr
        }

        // For non-AMP payments the overall payment identifier will be the same
        // hash as used for this HTLC.
        paymentIdentifier := htlcHash

        // For AMP-payments, we'll use the setID as the unique ID for the
        // overall payment.
        amp := finalHop.AMP
        if amp != nil {
                paymentIdentifier = amp.SetID()
        }

        // Record this payment hash with the ControlTower, ensuring it is not
        // already in-flight.
        info := &channeldb.PaymentCreationInfo{
                PaymentIdentifier: paymentIdentifier,
                Value:             amt,
                CreationTime:      r.cfg.Clock.Now(),
                PaymentRequest:    nil,
        }

        err := r.cfg.Control.InitPayment(paymentIdentifier, info)
        switch {
        // If this is an MPP attempt and the hash is already registered with
        // the database, we can go on to launch the shard.
        case mpp != nil && errors.Is(err, channeldb.ErrPaymentInFlight):
        case mpp != nil && errors.Is(err, channeldb.ErrPaymentExists):

        // Any other error is not tolerated.
        case err != nil:
                return nil, err
        }

        log.Tracef("Dispatching SendToRoute for HTLC hash %v: %v",
                htlcHash, newLogClosure(func() string {
                        return spew.Sdump(rt)
                }),
        )

        // Since the HTLC hashes and preimages are specified manually over the
        // RPC for SendToRoute requests, we don't have to worry about creating
        // a ShardTracker that can generate hashes for AMP payments. Instead, we
        // create a simple tracker that can just return the hash for the single
        // shard we'll now launch.
        shardTracker := shards.NewSimpleShardTracker(htlcHash, nil)

        // Create a payment lifecycle using the given route with,
        // - zero fee limit as we are not requesting routes.
        // - nil payment session (since we already have a route).
        // - no payment timeout.
        // - no current block height.
        p := newPaymentLifecycle(r, 0, paymentIdentifier, nil, shardTracker, 0)

        // We found a route to try, create a new HTLC attempt to try.
        //
        // NOTE: we use zero `remainingAmt` here to simulate the same effect of
        // setting the lastShard to be false, which is used by previous
        // implementation.
        attempt, err := p.registerAttempt(rt, 0)
        if err != nil {
                return nil, err
        }

        // Once the attempt is created, send it to the htlcswitch. Notice that
        // the `err` returned here has already been processed by
        // `handleSwitchErr`, which means if there's a terminal failure, the
        // payment has been failed.
        result, err := p.sendAttempt(attempt)
        if err != nil {
                return nil, err
        }

        // We now look up the payment to see if it's already failed.
        payment, err := p.router.cfg.Control.FetchPayment(p.identifier)
        if err != nil {
                return result.attempt, err
        }

        // Exit if the above error has caused the payment to be failed, we also
        // return the error from sending attempt to mimic the old behavior of
        // this method.
        _, failedReason := payment.TerminalInfo()
        if failedReason != nil {
                return result.attempt, result.err
        }

        // Since for SendToRoute we won't retry in case the shard fails, we'll
        // mark the payment failed with the control tower immediately if the
        // skipTempErr is false.
        reason := channeldb.FailureReasonError

        // If we failed to send the HTLC, we need to further decide if we want
        // to fail the payment.
        if result.err != nil {
                // If skipTempErr, we'll return the attempt and the temp error.
                if skipTempErr {
                        return result.attempt, result.err
                }

                // Otherwise we need to fail the payment.
                err := r.cfg.Control.FailPayment(paymentIdentifier, reason)
                if err != nil {
                        return nil, err
                }

                return result.attempt, result.err
        }

        // The attempt was successfully sent, wait for the result to be
        // available.
        result, err = p.collectResult(attempt)
        if err != nil {
                return nil, err
        }

        // We got a successful result.
        if result.err == nil {
                return result.attempt, nil
        }

        // An error returned from collecting the result, we'll mark the payment
        // as failed if we don't skip temp error.
        if !skipTempErr {
                err := r.cfg.Control.FailPayment(paymentIdentifier, reason)
                if err != nil {
                        return nil, err
                }
        }

        return result.attempt, result.err
}

// sendPayment attempts to send a payment to the passed payment hash. This
// function is blocking and will return either: when the payment is successful,
// or all candidates routes have been attempted and resulted in a failed
// payment. If the payment succeeds, then a non-nil Route will be returned
// which describes the path the successful payment traversed within the network
// to reach the destination. Additionally, the payment preimage will also be
// returned.
//
// This method relies on the ControlTower's internal payment state machine to
// carry out its execution. After restarts, it is safe, and assumed, that the
// router will call this method for every payment still in-flight according to
// the ControlTower.
func (r *ChannelRouter) sendPayment(ctx context.Context,
        feeLimit lnwire.MilliSatoshi, identifier lntypes.Hash,
        paymentAttemptTimeout time.Duration, paySession PaymentSession,
        shardTracker shards.ShardTracker) ([32]byte, *route.Route, error) {

        // If the user provides a timeout, we will additionally wrap the context
        // in a deadline.
        cancel := func() {}
        if paymentAttemptTimeout > 0 {
                ctx, cancel = context.WithTimeout(ctx, paymentAttemptTimeout)
        }

        // Since resumePayment is a blocking call, we'll cancel this
        // context if the payment completes before the optional
        // deadline.
        defer cancel()

        // We'll also fetch the current block height, so we can properly
        // calculate the required HTLC time locks within the route.
        _, currentHeight, err := r.cfg.Chain.GetBestBlock()
        if err != nil {
                return [32]byte{}, nil, err
        }

        // Now set up a paymentLifecycle struct with these params, such that we
        // can resume the payment from the current state.
        p := newPaymentLifecycle(
                r, feeLimit, identifier, paySession, shardTracker,
                currentHeight,
        )

        return p.resumePayment(ctx)
}

// extractChannelUpdate examines the error and extracts the channel update.
func (r *ChannelRouter) extractChannelUpdate(
        failure lnwire.FailureMessage) *lnwire.ChannelUpdate {

        var update *lnwire.ChannelUpdate
        switch onionErr := failure.(type) {
        case *lnwire.FailExpiryTooSoon:
                update = &onionErr.Update
        case *lnwire.FailAmountBelowMinimum:
                update = &onionErr.Update
        case *lnwire.FailFeeInsufficient:
                update = &onionErr.Update
        case *lnwire.FailIncorrectCltvExpiry:
                update = &onionErr.Update
        case *lnwire.FailChannelDisabled:
                update = &onionErr.Update
        case *lnwire.FailTemporaryChannelFailure:
                update = onionErr.Update
        }

        return update
}

// applyChannelUpdate validates a channel update and if valid, applies it to the
// database. It returns a bool indicating whether the updates were successful.
func (r *ChannelRouter) applyChannelUpdate(msg *lnwire.ChannelUpdate) bool {
        ch, _, _, err := r.GetChannelByID(msg.ShortChannelID)
        if err != nil {
                log.Errorf("Unable to retrieve channel by id: %v", err)
                return false
        }

        var pubKey *btcec.PublicKey

        switch msg.ChannelFlags & lnwire.ChanUpdateDirection {
        case 0:
                pubKey, _ = ch.NodeKey1()

        case 1:
                pubKey, _ = ch.NodeKey2()
        }

        // Exit early if the pubkey cannot be decided.
        if pubKey == nil {
                log.Errorf("Unable to decide pubkey with ChannelFlags=%v",
                        msg.ChannelFlags)
                return false
        }

        err = ValidateChannelUpdateAnn(pubKey, ch.Capacity, msg)
        if err != nil {
                log.Errorf("Unable to validate channel update: %v", err)
                return false
        }

        err = r.UpdateEdge(&models.ChannelEdgePolicy{
                SigBytes:                  msg.Signature.ToSignatureBytes(),
                ChannelID:                 msg.ShortChannelID.ToUint64(),
                LastUpdate:                time.Unix(int64(msg.Timestamp), 0),
                MessageFlags:              msg.MessageFlags,
                ChannelFlags:              msg.ChannelFlags,
                TimeLockDelta:             msg.TimeLockDelta,
                MinHTLC:                   msg.HtlcMinimumMsat,
                MaxHTLC:                   msg.HtlcMaximumMsat,
                FeeBaseMSat:               lnwire.MilliSatoshi(msg.BaseFee),
                FeeProportionalMillionths: lnwire.MilliSatoshi(msg.FeeRate),
                ExtraOpaqueData:           msg.ExtraOpaqueData,
        })
        if err != nil && !IsError(err, ErrIgnored, ErrOutdated) {
                log.Errorf("Unable to apply channel update: %v", err)
                return false
        }

        return true
}

// AddNode is used to add information about a node to the router database. If
// the node with this pubkey is not present in an existing channel, it will
// be ignored.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) AddNode(node *channeldb.LightningNode,
        op ...batch.SchedulerOption) error {

        rMsg := &routingMsg{
                msg: node,
                op:  op,
                err: make(chan error, 1),
        }

        select {
        case r.networkUpdates <- rMsg:
                select {
                case err := <-rMsg.err:
                        return err
                case <-r.quit:
                        return ErrRouterShuttingDown
                }
        case <-r.quit:
                return ErrRouterShuttingDown
        }
}

// AddEdge is used to add edge/channel to the topology of the router, after all
// information about channel will be gathered this edge/channel might be used
// in construction of payment path.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) AddEdge(edge *models.ChannelEdgeInfo,
        op ...batch.SchedulerOption) error {

        rMsg := &routingMsg{
                msg: edge,
                op:  op,
                err: make(chan error, 1),
        }

        select {
        case r.networkUpdates <- rMsg:
                select {
                case err := <-rMsg.err:
                        return err
                case <-r.quit:
                        return ErrRouterShuttingDown
                }
        case <-r.quit:
                return ErrRouterShuttingDown
        }
}

// UpdateEdge is used to update edge information, without this message edge
// considered as not fully constructed.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) UpdateEdge(update *models.ChannelEdgePolicy,
        op ...batch.SchedulerOption) error {

        rMsg := &routingMsg{
                msg: update,
                op:  op,
                err: make(chan error, 1),
        }

        select {
        case r.networkUpdates <- rMsg:
                select {
                case err := <-rMsg.err:
                        return err
                case <-r.quit:
                        return ErrRouterShuttingDown
                }
        case <-r.quit:
                return ErrRouterShuttingDown
        }
}

// CurrentBlockHeight returns the block height from POV of the router subsystem.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) CurrentBlockHeight() (uint32, error) {
        _, height, err := r.cfg.Chain.GetBestBlock()
        return uint32(height), err
}

// SyncedHeight returns the block height to which the router subsystem currently
// is synced to. This can differ from the above chain height if the goroutine
// responsible for processing the blocks isn't yet up to speed.
func (r *ChannelRouter) SyncedHeight() uint32 {
        return atomic.LoadUint32(&r.bestHeight)
}

// GetChannelByID return the channel by the channel id.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) GetChannelByID(chanID lnwire.ShortChannelID) (
        *models.ChannelEdgeInfo,
        *models.ChannelEdgePolicy,
        *models.ChannelEdgePolicy, error) {

        return r.cfg.Graph.FetchChannelEdgesByID(chanID.ToUint64())
}

// FetchLightningNode attempts to look up a target node by its identity public
// key. channeldb.ErrGraphNodeNotFound is returned if the node doesn't exist
// within the graph.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) FetchLightningNode(
        node route.Vertex) (*channeldb.LightningNode, error) {

        return r.cfg.Graph.FetchLightningNode(nil, node)
}

// ForEachNode is used to iterate over every node in router topology.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) ForEachNode(
        cb func(*channeldb.LightningNode) error) error {

        return r.cfg.Graph.ForEachNode(
                func(_ kvdb.RTx, n *channeldb.LightningNode) error {
                        return cb(n)
                })
}

// ForAllOutgoingChannels is used to iterate over all outgoing channels owned by
// the router.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) ForAllOutgoingChannels(cb func(kvdb.RTx,
        *models.ChannelEdgeInfo, *models.ChannelEdgePolicy) error) error {

        return r.cfg.Graph.ForEachNodeChannel(nil, r.selfNode.PubKeyBytes,
                func(tx kvdb.RTx, c *models.ChannelEdgeInfo,
                        e *models.ChannelEdgePolicy,
                        _ *models.ChannelEdgePolicy) error {

                        if e == nil {
                                return fmt.Errorf("channel from self node " +
                                        "has no policy")
                        }

                        return cb(tx, c, e)
                },
        )
}

// AddProof updates the channel edge info with proof which is needed to
// properly announce the edge to the rest of the network.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) AddProof(chanID lnwire.ShortChannelID,
        proof *models.ChannelAuthProof) error {

        info, _, _, err := r.cfg.Graph.FetchChannelEdgesByID(chanID.ToUint64())
        if err != nil {
                return err
        }

        info.AuthProof = proof
        return r.cfg.Graph.UpdateChannelEdge(info)
}

// IsStaleNode returns true if the graph source has a node announcement for the
// target node with a more recent timestamp.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) IsStaleNode(node route.Vertex,
        timestamp time.Time) bool {

        // If our attempt to assert that the node announcement is fresh fails,
        // then we know that this is actually a stale announcement.
        err := r.assertNodeAnnFreshness(node, timestamp)
        if err != nil {
                log.Debugf("Checking stale node %x got %v", node, err)
                return true
        }

        return false
}

// IsPublicNode determines whether the given vertex is seen as a public node in
// the graph from the graph's source node's point of view.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) IsPublicNode(node route.Vertex) (bool, error) {
        return r.cfg.Graph.IsPublicNode(node)
}

// IsKnownEdge returns true if the graph source already knows of the passed
// channel ID either as a live or zombie edge.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) IsKnownEdge(chanID lnwire.ShortChannelID) bool {
        _, _, exists, isZombie, _ := r.cfg.Graph.HasChannelEdge(
                chanID.ToUint64(),
        )
        return exists || isZombie
}

// IsStaleEdgePolicy returns true if the graph source has a channel edge for
// the passed channel ID (and flags) that have a more recent timestamp.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) IsStaleEdgePolicy(chanID lnwire.ShortChannelID,
        timestamp time.Time, flags lnwire.ChanUpdateChanFlags) bool {

        edge1Timestamp, edge2Timestamp, exists, isZombie, err :=
                r.cfg.Graph.HasChannelEdge(chanID.ToUint64())
        if err != nil {
                log.Debugf("Check stale edge policy got error: %v", err)
                return false

        }

        // If we know of the edge as a zombie, then we'll make some additional
        // checks to determine if the new policy is fresh.
        if isZombie {
                // When running with AssumeChannelValid, we also prune channels
                // if both of their edges are disabled. We'll mark the new
                // policy as stale if it remains disabled.
                if r.cfg.AssumeChannelValid {
                        isDisabled := flags&lnwire.ChanUpdateDisabled ==
                                lnwire.ChanUpdateDisabled
                        if isDisabled {
                                return true
                        }
                }

                // Otherwise, we'll fall back to our usual ChannelPruneExpiry.
                return time.Since(timestamp) > r.cfg.ChannelPruneExpiry
        }

        // If we don't know of the edge, then it means it's fresh (thus not
        // stale).
        if !exists {
                return false
        }

        // As edges are directional edge node has a unique policy for the
        // direction of the edge they control. Therefore, we first check if we
        // already have the most up-to-date information for that edge. If so,
        // then we can exit early.
        switch {
        // A flag set of 0 indicates this is an announcement for the "first"
        // node in the channel.
        case flags&lnwire.ChanUpdateDirection == 0:
                return !edge1Timestamp.Before(timestamp)

        // Similarly, a flag set of 1 indicates this is an announcement for the
        // "second" node in the channel.
        case flags&lnwire.ChanUpdateDirection == 1:
                return !edge2Timestamp.Before(timestamp)
        }

        return false
}

// MarkEdgeLive clears an edge from our zombie index, deeming it as live.
//
// NOTE: This method is part of the ChannelGraphSource interface.
func (r *ChannelRouter) MarkEdgeLive(chanID lnwire.ShortChannelID) error {
        return r.cfg.Graph.MarkEdgeLive(chanID.ToUint64())
}

// ErrNoChannel is returned when a route cannot be built because there are no
// channels that satisfy all requirements.
type ErrNoChannel struct {
        position int
        fromNode route.Vertex
}

// Error returns a human-readable string describing the error.
func (e ErrNoChannel) Error() string {
        return fmt.Sprintf("no matching outgoing channel available for "+
                "node %v (%v)", e.position, e.fromNode)
}

// BuildRoute returns a fully specified route based on a list of pubkeys. If
// amount is nil, the minimum routable amount is used. To force a specific
// outgoing channel, use the outgoingChan parameter.
func (r *ChannelRouter) BuildRoute(amt *lnwire.MilliSatoshi,
        hops []route.Vertex, outgoingChan *uint64,
        finalCltvDelta int32, payAddr *[32]byte) (*route.Route, error) {

        log.Tracef("BuildRoute called: hopsCount=%v, amt=%v",
                len(hops), amt)

        var outgoingChans map[uint64]struct{}
        if outgoingChan != nil {
                outgoingChans = map[uint64]struct{}{
                        *outgoingChan: {},
                }
        }

        // If no amount is specified, we need to build a route for the minimum
        // amount that this route can carry.
        useMinAmt := amt == nil

        var runningAmt lnwire.MilliSatoshi
        if useMinAmt {
                // For minimum amount routes, aim to deliver at least 1 msat to
                // the destination. There are nodes in the wild that have a
                // min_htlc channel policy of zero, which could lead to a zero
                // amount payment being made.
                runningAmt = 1
        } else {
                // If an amount is specified, we need to build a route that
                // delivers exactly this amount to the final destination.
                runningAmt = *amt
        }

        // We'll attempt to obtain a set of bandwidth hints that helps us select
        // the best outgoing channel to use in case no outgoing channel is set.
        bandwidthHints, err := newBandwidthManager(
                r.cachedGraph, r.selfNode.PubKeyBytes, r.cfg.GetLink,
        )
        if err != nil {
                return nil, err
        }

        // Fetch the current block height outside the routing transaction, to
        // prevent the rpc call blocking the database.
        _, height, err := r.cfg.Chain.GetBestBlock()
        if err != nil {
                return nil, err
        }

        sourceNode := r.selfNode.PubKeyBytes
        unifiers, senderAmt, err := getRouteUnifiers(
                sourceNode, hops, useMinAmt, runningAmt, outgoingChans,
                r.cachedGraph, bandwidthHints,
        )
        if err != nil {
                return nil, err
        }

        pathEdges, receiverAmt, err := getPathEdges(
                sourceNode, senderAmt, unifiers, bandwidthHints, hops,
        )
        if err != nil {
                return nil, err
        }

        // Build and return the final route.
        return newRoute(
                sourceNode, pathEdges, uint32(height),
                finalHopParams{
                        amt:         receiverAmt,
                        totalAmt:    receiverAmt,
                        cltvDelta:   uint16(finalCltvDelta),
                        records:     nil,
                        paymentAddr: payAddr,
                }, nil,
        )
}

// getRouteUnifiers returns a list of edge unifiers for the given route.
func getRouteUnifiers(source route.Vertex, hops []route.Vertex,
        useMinAmt bool, runningAmt lnwire.MilliSatoshi,
        outgoingChans map[uint64]struct{}, graph routingGraph,
        bandwidthHints *bandwidthManager) ([]*edgeUnifier, lnwire.MilliSatoshi,
        error) {

        // Allocate a list that will contain the edge unifiers for this route.
        unifiers := make([]*edgeUnifier, len(hops))

        // Traverse hops backwards to accumulate fees in the running amounts.
        for i := len(hops) - 1; i >= 0; i-- {
                toNode := hops[i]

                var fromNode route.Vertex
                if i == 0 {
                        fromNode = source
                } else {
                        fromNode = hops[i-1]
                }

                localChan := i == 0

                // Build unified policies for this hop based on the channels
                // known in the graph. Don't use inbound fees.
                //
                // TODO: Add inbound fees support for BuildRoute.
                u := newNodeEdgeUnifier(
                        source, toNode, false, outgoingChans,
                )

                err := u.addGraphPolicies(graph)
                if err != nil {
                        return nil, 0, err
                }

                // Exit if there are no channels.
                edgeUnifier, ok := u.edgeUnifiers[fromNode]
                if !ok {
                        log.Errorf("Cannot find policy for node %v", fromNode)
                        return nil, 0, ErrNoChannel{
                                fromNode: fromNode,
                                position: i,
                        }
                }

                // If using min amt, increase amt if needed.
                if useMinAmt {
                        min := edgeUnifier.minAmt()
                        if min > runningAmt {
                                runningAmt = min
                        }
                }

                // Get an edge for the specific amount that we want to forward.
                edge := edgeUnifier.getEdge(runningAmt, bandwidthHints, 0)
                if edge == nil {
                        log.Errorf("Cannot find policy with amt=%v for node %v",
                                runningAmt, fromNode)

                        return nil, 0, ErrNoChannel{
                                fromNode: fromNode,
                                position: i,
                        }
                }

                // Add fee for this hop.
                if !localChan {
                        runningAmt += edge.policy.ComputeFee(runningAmt)
                }

                log.Tracef("Select channel %v at position %v",
                        edge.policy.ChannelID, i)

                unifiers[i] = edgeUnifier
        }

        return unifiers, runningAmt, nil
}

// getPathEdges returns the edges that make up the path and the total amount,
// including fees, to send the payment.
func getPathEdges(source route.Vertex, receiverAmt lnwire.MilliSatoshi,
        unifiers []*edgeUnifier, bandwidthHints *bandwidthManager,
        hops []route.Vertex) ([]*unifiedEdge,
        lnwire.MilliSatoshi, error) {

        // Now that we arrived at the start of the route and found out the route
        // total amount, we make a forward pass. Because the amount may have
        // been increased in the backward pass, fees need to be recalculated and
        // amount ranges re-checked.
        var pathEdges []*unifiedEdge
        for i, unifier := range unifiers {
                edge := unifier.getEdge(receiverAmt, bandwidthHints, 0)
                if edge == nil {
                        fromNode := source
                        if i > 0 {
                                fromNode = hops[i-1]
                        }

                        return nil, 0, ErrNoChannel{
                                fromNode: fromNode,
                                position: i,
                        }
                }

                if i > 0 {
                        // Decrease the amount to send while going forward.
                        receiverAmt -= edge.policy.ComputeFeeFromIncoming(
                                receiverAmt,
                        )
                }

                pathEdges = append(pathEdges, edge)
        }

        return pathEdges, receiverAmt, nil
}

lightningnetwork / lnd / 9915780197

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