• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

lightningnetwork / lnd / 15016440952

14 May 2025 08:51AM UTC coverage: 69.031% (+0.03%) from 68.997%
15016440952

Pull #9692

github

web-flow
Merge b7e72b2ef into b0cba7dd0
Pull Request #9692: [graph-work-side-branch]: temp side branch for graph work

292 of 349 new or added lines in 32 files covered. (83.67%)

45 existing lines in 13 files now uncovered.

134025 of 194151 relevant lines covered (69.03%)

22047.58 hits per line

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

78.9
/peer/brontide.go
1
package peer
2

3
import (
4
        "bytes"
5
        "container/list"
6
        "context"
7
        "errors"
8
        "fmt"
9
        "math/rand"
10
        "net"
11
        "strings"
12
        "sync"
13
        "sync/atomic"
14
        "time"
15

16
        "github.com/btcsuite/btcd/btcec/v2"
17
        "github.com/btcsuite/btcd/chaincfg/chainhash"
18
        "github.com/btcsuite/btcd/connmgr"
19
        "github.com/btcsuite/btcd/txscript"
20
        "github.com/btcsuite/btcd/wire"
21
        "github.com/btcsuite/btclog/v2"
22
        "github.com/davecgh/go-spew/spew"
23
        "github.com/lightningnetwork/lnd/buffer"
24
        "github.com/lightningnetwork/lnd/chainntnfs"
25
        "github.com/lightningnetwork/lnd/channeldb"
26
        "github.com/lightningnetwork/lnd/channelnotifier"
27
        "github.com/lightningnetwork/lnd/contractcourt"
28
        "github.com/lightningnetwork/lnd/discovery"
29
        "github.com/lightningnetwork/lnd/feature"
30
        "github.com/lightningnetwork/lnd/fn/v2"
31
        "github.com/lightningnetwork/lnd/funding"
32
        graphdb "github.com/lightningnetwork/lnd/graph/db"
33
        "github.com/lightningnetwork/lnd/graph/db/models"
34
        "github.com/lightningnetwork/lnd/htlcswitch"
35
        "github.com/lightningnetwork/lnd/htlcswitch/hodl"
36
        "github.com/lightningnetwork/lnd/htlcswitch/hop"
37
        "github.com/lightningnetwork/lnd/input"
38
        "github.com/lightningnetwork/lnd/invoices"
39
        "github.com/lightningnetwork/lnd/keychain"
40
        "github.com/lightningnetwork/lnd/lnpeer"
41
        "github.com/lightningnetwork/lnd/lntypes"
42
        "github.com/lightningnetwork/lnd/lnutils"
43
        "github.com/lightningnetwork/lnd/lnwallet"
44
        "github.com/lightningnetwork/lnd/lnwallet/chainfee"
45
        "github.com/lightningnetwork/lnd/lnwallet/chancloser"
46
        "github.com/lightningnetwork/lnd/lnwire"
47
        "github.com/lightningnetwork/lnd/msgmux"
48
        "github.com/lightningnetwork/lnd/netann"
49
        "github.com/lightningnetwork/lnd/pool"
50
        "github.com/lightningnetwork/lnd/protofsm"
51
        "github.com/lightningnetwork/lnd/queue"
52
        "github.com/lightningnetwork/lnd/subscribe"
53
        "github.com/lightningnetwork/lnd/ticker"
54
        "github.com/lightningnetwork/lnd/tlv"
55
        "github.com/lightningnetwork/lnd/watchtower/wtclient"
56
)
57

58
const (
59
        // pingInterval is the interval at which ping messages are sent.
60
        pingInterval = 1 * time.Minute
61

62
        // pingTimeout is the amount of time we will wait for a pong response
63
        // before considering the peer to be unresponsive.
64
        //
65
        // This MUST be a smaller value than the pingInterval.
66
        pingTimeout = 30 * time.Second
67

68
        // idleTimeout is the duration of inactivity before we time out a peer.
69
        idleTimeout = 5 * time.Minute
70

71
        // writeMessageTimeout is the timeout used when writing a message to the
72
        // peer.
73
        writeMessageTimeout = 5 * time.Second
74

75
        // readMessageTimeout is the timeout used when reading a message from a
76
        // peer.
77
        readMessageTimeout = 5 * time.Second
78

79
        // handshakeTimeout is the timeout used when waiting for the peer's init
80
        // message.
81
        handshakeTimeout = 15 * time.Second
82

83
        // ErrorBufferSize is the number of historic peer errors that we store.
84
        ErrorBufferSize = 10
85

86
        // pongSizeCeiling is the upper bound on a uniformly distributed random
87
        // variable that we use for requesting pong responses. We don't use the
88
        // MaxPongBytes (upper bound accepted by the protocol) because it is
89
        // needlessly wasteful of precious Tor bandwidth for little to no gain.
90
        pongSizeCeiling = 4096
91

92
        // torTimeoutMultiplier is the scaling factor we use on network timeouts
93
        // for Tor peers.
94
        torTimeoutMultiplier = 3
95

96
        // msgStreamSize is the size of the message streams.
97
        msgStreamSize = 5
98
)
99

100
var (
101
        // ErrChannelNotFound is an error returned when a channel is queried and
102
        // either the Brontide doesn't know of it, or the channel in question
103
        // is pending.
104
        ErrChannelNotFound = fmt.Errorf("channel not found")
105
)
106

107
// outgoingMsg packages an lnwire.Message to be sent out on the wire, along with
108
// a buffered channel which will be sent upon once the write is complete. This
109
// buffered channel acts as a semaphore to be used for synchronization purposes.
110
type outgoingMsg struct {
111
        priority bool
112
        msg      lnwire.Message
113
        errChan  chan error // MUST be buffered.
114
}
115

116
// newChannelMsg packages a channeldb.OpenChannel with a channel that allows
117
// the receiver of the request to report when the channel creation process has
118
// completed.
119
type newChannelMsg struct {
120
        // channel is used when the pending channel becomes active.
121
        channel *lnpeer.NewChannel
122

123
        // channelID is used when there's a new pending channel.
124
        channelID lnwire.ChannelID
125

126
        err chan error
127
}
128

129
type customMsg struct {
130
        peer [33]byte
131
        msg  lnwire.Custom
132
}
133

134
// closeMsg is a wrapper struct around any wire messages that deal with the
135
// cooperative channel closure negotiation process. This struct includes the
136
// raw channel ID targeted along with the original message.
137
type closeMsg struct {
138
        cid lnwire.ChannelID
139
        msg lnwire.Message
140
}
141

142
// PendingUpdate describes the pending state of a closing channel.
143
type PendingUpdate struct {
144
        // Txid is the txid of the closing transaction.
145
        Txid []byte
146

147
        // OutputIndex is the output index of our output in the closing
148
        // transaction.
149
        OutputIndex uint32
150

151
        // FeePerVByte is an optional field, that is set only when the new RBF
152
        // coop close flow is used. This indicates the new closing fee rate on
153
        // the closing transaction.
154
        FeePerVbyte fn.Option[chainfee.SatPerVByte]
155

156
        // IsLocalCloseTx is an optional field that indicates if this update is
157
        // sent for our local close txn, or the close txn of the remote party.
158
        // This is only set if the new RBF coop close flow is used.
159
        IsLocalCloseTx fn.Option[bool]
160
}
161

162
// ChannelCloseUpdate contains the outcome of the close channel operation.
163
type ChannelCloseUpdate struct {
164
        ClosingTxid []byte
165
        Success     bool
166

167
        // LocalCloseOutput is an optional, additional output on the closing
168
        // transaction that the local party should be paid to. This will only be
169
        // populated if the local balance isn't dust.
170
        LocalCloseOutput fn.Option[chancloser.CloseOutput]
171

172
        // RemoteCloseOutput is an optional, additional output on the closing
173
        // transaction that the remote party should be paid to. This will only
174
        // be populated if the remote balance isn't dust.
175
        RemoteCloseOutput fn.Option[chancloser.CloseOutput]
176

177
        // AuxOutputs is an optional set of additional outputs that might be
178
        // included in the closing transaction. These are used for custom
179
        // channel types.
180
        AuxOutputs fn.Option[chancloser.AuxCloseOutputs]
181
}
182

183
// TimestampedError is a timestamped error that is used to store the most recent
184
// errors we have experienced with our peers.
185
type TimestampedError struct {
186
        Error     error
187
        Timestamp time.Time
188
}
189

190
// Config defines configuration fields that are necessary for a peer object
191
// to function.
192
type Config struct {
193
        // Conn is the underlying network connection for this peer.
194
        Conn MessageConn
195

196
        // ConnReq stores information related to the persistent connection request
197
        // for this peer.
198
        ConnReq *connmgr.ConnReq
199

200
        // PubKeyBytes is the serialized, compressed public key of this peer.
201
        PubKeyBytes [33]byte
202

203
        // Addr is the network address of the peer.
204
        Addr *lnwire.NetAddress
205

206
        // Inbound indicates whether or not the peer is an inbound peer.
207
        Inbound bool
208

209
        // Features is the set of features that we advertise to the remote party.
210
        Features *lnwire.FeatureVector
211

212
        // LegacyFeatures is the set of features that we advertise to the remote
213
        // peer for backwards compatibility. Nodes that have not implemented
214
        // flat features will still be able to read our feature bits from the
215
        // legacy global field, but we will also advertise everything in the
216
        // default features field.
217
        LegacyFeatures *lnwire.FeatureVector
218

219
        // OutgoingCltvRejectDelta defines the number of blocks before expiry of
220
        // an htlc where we don't offer it anymore.
221
        OutgoingCltvRejectDelta uint32
222

223
        // ChanActiveTimeout specifies the duration the peer will wait to request
224
        // a channel reenable, beginning from the time the peer was started.
225
        ChanActiveTimeout time.Duration
226

227
        // ErrorBuffer stores a set of errors related to a peer. It contains error
228
        // messages that our peer has recently sent us over the wire and records of
229
        // unknown messages that were sent to us so that we can have a full track
230
        // record of the communication errors we have had with our peer. If we
231
        // choose to disconnect from a peer, it also stores the reason we had for
232
        // disconnecting.
233
        ErrorBuffer *queue.CircularBuffer
234

235
        // WritePool is the task pool that manages reuse of write buffers. Write
236
        // tasks are submitted to the pool in order to conserve the total number of
237
        // write buffers allocated at any one time, and decouple write buffer
238
        // allocation from the peer life cycle.
239
        WritePool *pool.Write
240

241
        // ReadPool is the task pool that manages reuse of read buffers.
242
        ReadPool *pool.Read
243

244
        // Switch is a pointer to the htlcswitch. It is used to setup, get, and
245
        // tear-down ChannelLinks.
246
        Switch messageSwitch
247

248
        // InterceptSwitch is a pointer to the InterceptableSwitch, a wrapper around
249
        // the regular Switch. We only export it here to pass ForwardPackets to the
250
        // ChannelLinkConfig.
251
        InterceptSwitch *htlcswitch.InterceptableSwitch
252

253
        // ChannelDB is used to fetch opened channels, and closed channels.
254
        ChannelDB *channeldb.ChannelStateDB
255

256
        // ChannelGraph is a pointer to the channel graph which is used to
257
        // query information about the set of known active channels.
258
        ChannelGraph *graphdb.ChannelGraph
259

260
        // ChainArb is used to subscribe to channel events, update contract signals,
261
        // and force close channels.
262
        ChainArb *contractcourt.ChainArbitrator
263

264
        // AuthGossiper is needed so that the Brontide impl can register with the
265
        // gossiper and process remote channel announcements.
266
        AuthGossiper *discovery.AuthenticatedGossiper
267

268
        // ChanStatusMgr is used to set or un-set the disabled bit in channel
269
        // updates.
270
        ChanStatusMgr *netann.ChanStatusManager
271

272
        // ChainIO is used to retrieve the best block.
273
        ChainIO lnwallet.BlockChainIO
274

275
        // FeeEstimator is used to compute our target ideal fee-per-kw when
276
        // initializing the coop close process.
277
        FeeEstimator chainfee.Estimator
278

279
        // Signer is used when creating *lnwallet.LightningChannel instances.
280
        Signer input.Signer
281

282
        // SigPool is used when creating *lnwallet.LightningChannel instances.
283
        SigPool *lnwallet.SigPool
284

285
        // Wallet is used to publish transactions and generates delivery
286
        // scripts during the coop close process.
287
        Wallet *lnwallet.LightningWallet
288

289
        // ChainNotifier is used to receive confirmations of a coop close
290
        // transaction.
291
        ChainNotifier chainntnfs.ChainNotifier
292

293
        // BestBlockView is used to efficiently query for up-to-date
294
        // blockchain state information
295
        BestBlockView chainntnfs.BestBlockView
296

297
        // RoutingPolicy is used to set the forwarding policy for links created by
298
        // the Brontide.
299
        RoutingPolicy models.ForwardingPolicy
300

301
        // Sphinx is used when setting up ChannelLinks so they can decode sphinx
302
        // onion blobs.
303
        Sphinx *hop.OnionProcessor
304

305
        // WitnessBeacon is used when setting up ChannelLinks so they can add any
306
        // preimages that they learn.
307
        WitnessBeacon contractcourt.WitnessBeacon
308

309
        // Invoices is passed to the ChannelLink on creation and handles all
310
        // invoice-related logic.
311
        Invoices *invoices.InvoiceRegistry
312

313
        // ChannelNotifier is used by the link to notify other sub-systems about
314
        // channel-related events and by the Brontide to subscribe to
315
        // ActiveLinkEvents.
316
        ChannelNotifier *channelnotifier.ChannelNotifier
317

318
        // HtlcNotifier is used when creating a ChannelLink.
319
        HtlcNotifier *htlcswitch.HtlcNotifier
320

321
        // TowerClient is used to backup revoked states.
322
        TowerClient wtclient.ClientManager
323

324
        // DisconnectPeer is used to disconnect this peer if the cooperative close
325
        // process fails.
326
        DisconnectPeer func(*btcec.PublicKey) error
327

328
        // GenNodeAnnouncement is used to send our node announcement to the remote
329
        // on startup.
330
        GenNodeAnnouncement func(...netann.NodeAnnModifier) (
331
                lnwire.NodeAnnouncement, error)
332

333
        // PrunePersistentPeerConnection is used to remove all internal state
334
        // related to this peer in the server.
335
        PrunePersistentPeerConnection func([33]byte)
336

337
        // FetchLastChanUpdate fetches our latest channel update for a target
338
        // channel.
339
        FetchLastChanUpdate func(lnwire.ShortChannelID) (*lnwire.ChannelUpdate1,
340
                error)
341

342
        // FundingManager is an implementation of the funding.Controller interface.
343
        FundingManager funding.Controller
344

345
        // Hodl is used when creating ChannelLinks to specify HodlFlags as
346
        // breakpoints in dev builds.
347
        Hodl *hodl.Config
348

349
        // UnsafeReplay is used when creating ChannelLinks to specify whether or
350
        // not to replay adds on its commitment tx.
351
        UnsafeReplay bool
352

353
        // MaxOutgoingCltvExpiry is used when creating ChannelLinks and is the max
354
        // number of blocks that funds could be locked up for when forwarding
355
        // payments.
356
        MaxOutgoingCltvExpiry uint32
357

358
        // MaxChannelFeeAllocation is used when creating ChannelLinks and is the
359
        // maximum percentage of total funds that can be allocated to a channel's
360
        // commitment fee. This only applies for the initiator of the channel.
361
        MaxChannelFeeAllocation float64
362

363
        // MaxAnchorsCommitFeeRate is the maximum fee rate we'll use as an
364
        // initiator for anchor channel commitments.
365
        MaxAnchorsCommitFeeRate chainfee.SatPerKWeight
366

367
        // CoopCloseTargetConfs is the confirmation target that will be used
368
        // to estimate the fee rate to use during a cooperative channel
369
        // closure initiated by the remote peer.
370
        CoopCloseTargetConfs uint32
371

372
        // ServerPubKey is the serialized, compressed public key of our lnd node.
373
        // It is used to determine which policy (channel edge) to pass to the
374
        // ChannelLink.
375
        ServerPubKey [33]byte
376

377
        // ChannelCommitInterval is the maximum time that is allowed to pass between
378
        // receiving a channel state update and signing the next commitment.
379
        // Setting this to a longer duration allows for more efficient channel
380
        // operations at the cost of latency.
381
        ChannelCommitInterval time.Duration
382

383
        // PendingCommitInterval is the maximum time that is allowed to pass
384
        // while waiting for the remote party to revoke a locally initiated
385
        // commitment state. Setting this to a longer duration if a slow
386
        // response is expected from the remote party or large number of
387
        // payments are attempted at the same time.
388
        PendingCommitInterval time.Duration
389

390
        // ChannelCommitBatchSize is the maximum number of channel state updates
391
        // that is accumulated before signing a new commitment.
392
        ChannelCommitBatchSize uint32
393

394
        // HandleCustomMessage is called whenever a custom message is received
395
        // from the peer.
396
        HandleCustomMessage func(peer [33]byte, msg *lnwire.Custom) error
397

398
        // GetAliases is passed to created links so the Switch and link can be
399
        // aware of the channel's aliases.
400
        GetAliases func(base lnwire.ShortChannelID) []lnwire.ShortChannelID
401

402
        // RequestAlias allows the Brontide struct to request an alias to send
403
        // to the peer.
404
        RequestAlias func() (lnwire.ShortChannelID, error)
405

406
        // AddLocalAlias persists an alias to an underlying alias store.
407
        AddLocalAlias func(alias, base lnwire.ShortChannelID,
408
                gossip, liveUpdate bool) error
409

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

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

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

422
        // AuxTrafficShaper is an optional auxiliary traffic shaper that can be
423
        // used to manage the bandwidth of peer links.
424
        AuxTrafficShaper fn.Option[htlcswitch.AuxTrafficShaper]
425

426
        // PongBuf is a slice we'll reuse instead of allocating memory on the
427
        // heap. Since only reads will occur and no writes, there is no need
428
        // for any synchronization primitives. As a result, it's safe to share
429
        // this across multiple Peer struct instances.
430
        PongBuf []byte
431

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

437
        // DisallowQuiescence is a flag that indicates whether the Brontide
438
        // should have the quiescence feature disabled.
439
        DisallowQuiescence bool
440

441
        // MaxFeeExposure limits the number of outstanding fees in a channel.
442
        // This value will be passed to created links.
443
        MaxFeeExposure lnwire.MilliSatoshi
444

445
        // MsgRouter is an optional instance of the main message router that
446
        // the peer will use. If None, then a new default version will be used
447
        // in place.
448
        MsgRouter fn.Option[msgmux.Router]
449

450
        // AuxChanCloser is an optional instance of an abstraction that can be
451
        // used to modify the way the co-op close transaction is constructed.
452
        AuxChanCloser fn.Option[chancloser.AuxChanCloser]
453

454
        // ShouldFwdExpEndorsement is a closure that indicates whether
455
        // experimental endorsement signals should be set.
456
        ShouldFwdExpEndorsement func() bool
457

458
        // Quit is the server's quit channel. If this is closed, we halt operation.
459
        Quit chan struct{}
460
}
461

462
// chanCloserFsm is a union-like type that can hold the two versions of co-op
463
// close we support: negotiation, and RBF based.
464
//
465
// TODO(roasbeef): rename to chancloser.Negotiator and chancloser.RBF?
466
type chanCloserFsm = fn.Either[*chancloser.ChanCloser, *chancloser.RbfChanCloser] //nolint:ll
467

468
// makeNegotiateCloser creates a new negotiate closer from a
469
// chancloser.ChanCloser.
470
func makeNegotiateCloser(chanCloser *chancloser.ChanCloser) chanCloserFsm {
12✔
471
        return fn.NewLeft[*chancloser.ChanCloser, *chancloser.RbfChanCloser](
12✔
472
                chanCloser,
12✔
473
        )
12✔
474
}
12✔
475

476
// makeRbfCloser creates a new RBF closer from a chancloser.RbfChanCloser.
477
func makeRbfCloser(rbfCloser *chancloser.RbfChanCloser) chanCloserFsm {
3✔
478
        return fn.NewRight[*chancloser.ChanCloser](
3✔
479
                rbfCloser,
3✔
480
        )
3✔
481
}
3✔
482

483
// Brontide is an active peer on the Lightning Network. This struct is responsible
484
// for managing any channel state related to this peer. To do so, it has
485
// several helper goroutines to handle events such as HTLC timeouts, new
486
// funding workflow, and detecting an uncooperative closure of any active
487
// channels.
488
type Brontide struct {
489
        // MUST be used atomically.
490
        started    int32
491
        disconnect int32
492

493
        // MUST be used atomically.
494
        bytesReceived uint64
495
        bytesSent     uint64
496

497
        // isTorConnection is a flag that indicates whether or not we believe
498
        // the remote peer is a tor connection. It is not always possible to
499
        // know this with certainty but we have heuristics we use that should
500
        // catch most cases.
501
        //
502
        // NOTE: We judge the tor-ness of a connection by if the remote peer has
503
        // ".onion" in the address OR if it's connected over localhost.
504
        // This will miss cases where our peer is connected to our clearnet
505
        // address over the tor network (via exit nodes). It will also misjudge
506
        // actual localhost connections as tor. We need to include this because
507
        // inbound connections to our tor address will appear to come from the
508
        // local socks5 proxy. This heuristic is only used to expand the timeout
509
        // window for peers so it is OK to misjudge this. If you use this field
510
        // for any other purpose you should seriously consider whether or not
511
        // this heuristic is good enough for your use case.
512
        isTorConnection bool
513

514
        pingManager *PingManager
515

516
        // lastPingPayload stores an unsafe pointer wrapped as an atomic
517
        // variable which points to the last payload the remote party sent us
518
        // as their ping.
519
        //
520
        // MUST be used atomically.
521
        lastPingPayload atomic.Value
522

523
        cfg Config
524

525
        // activeSignal when closed signals that the peer is now active and
526
        // ready to process messages.
527
        activeSignal chan struct{}
528

529
        // startTime is the time this peer connection was successfully established.
530
        // It will be zero for peers that did not successfully call Start().
531
        startTime time.Time
532

533
        // sendQueue is the channel which is used to queue outgoing messages to be
534
        // written onto the wire. Note that this channel is unbuffered.
535
        sendQueue chan outgoingMsg
536

537
        // outgoingQueue is a buffered channel which allows second/third party
538
        // objects to queue messages to be sent out on the wire.
539
        outgoingQueue chan outgoingMsg
540

541
        // activeChannels is a map which stores the state machines of all
542
        // active channels. Channels are indexed into the map by the txid of
543
        // the funding transaction which opened the channel.
544
        //
545
        // NOTE: On startup, pending channels are stored as nil in this map.
546
        // Confirmed channels have channel data populated in the map. This means
547
        // that accesses to this map should nil-check the LightningChannel to
548
        // see if this is a pending channel or not. The tradeoff here is either
549
        // having two maps everywhere (one for pending, one for confirmed chans)
550
        // or having an extra nil-check per access.
551
        activeChannels *lnutils.SyncMap[
552
                lnwire.ChannelID, *lnwallet.LightningChannel]
553

554
        // addedChannels tracks any new channels opened during this peer's
555
        // lifecycle. We use this to filter out these new channels when the time
556
        // comes to request a reenable for active channels, since they will have
557
        // waited a shorter duration.
558
        addedChannels *lnutils.SyncMap[lnwire.ChannelID, struct{}]
559

560
        // newActiveChannel is used by the fundingManager to send fully opened
561
        // channels to the source peer which handled the funding workflow.
562
        newActiveChannel chan *newChannelMsg
563

564
        // newPendingChannel is used by the fundingManager to send pending open
565
        // channels to the source peer which handled the funding workflow.
566
        newPendingChannel chan *newChannelMsg
567

568
        // removePendingChannel is used by the fundingManager to cancel pending
569
        // open channels to the source peer when the funding flow is failed.
570
        removePendingChannel chan *newChannelMsg
571

572
        // activeMsgStreams is a map from channel id to the channel streams that
573
        // proxy messages to individual, active links.
574
        activeMsgStreams map[lnwire.ChannelID]*msgStream
575

576
        // activeChanCloses is a map that keeps track of all the active
577
        // cooperative channel closures. Any channel closing messages are directed
578
        // to one of these active state machines. Once the channel has been closed,
579
        // the state machine will be deleted from the map.
580
        activeChanCloses *lnutils.SyncMap[lnwire.ChannelID, chanCloserFsm]
581

582
        // localCloseChanReqs is a channel in which any local requests to close
583
        // a particular channel are sent over.
584
        localCloseChanReqs chan *htlcswitch.ChanClose
585

586
        // linkFailures receives all reported channel failures from the switch,
587
        // and instructs the channelManager to clean remaining channel state.
588
        linkFailures chan linkFailureReport
589

590
        // chanCloseMsgs is a channel that any message related to channel
591
        // closures are sent over. This includes lnwire.Shutdown message as
592
        // well as lnwire.ClosingSigned messages.
593
        chanCloseMsgs chan *closeMsg
594

595
        // remoteFeatures is the feature vector received from the peer during
596
        // the connection handshake.
597
        remoteFeatures *lnwire.FeatureVector
598

599
        // resentChanSyncMsg is a set that keeps track of which channels we
600
        // have re-sent channel reestablishment messages for. This is done to
601
        // avoid getting into loop where both peers will respond to the other
602
        // peer's chansync message with its own over and over again.
603
        resentChanSyncMsg map[lnwire.ChannelID]struct{}
604

605
        // channelEventClient is the channel event subscription client that's
606
        // used to assist retry enabling the channels. This client is only
607
        // created when the reenableTimeout is no greater than 1 minute. Once
608
        // created, it is canceled once the reenabling has been finished.
609
        //
610
        // NOTE: we choose to create the client conditionally to avoid
611
        // potentially holding lots of un-consumed events.
612
        channelEventClient *subscribe.Client
613

614
        // msgRouter is an instance of the msgmux.Router which is used to send
615
        // off new wire messages for handing.
616
        msgRouter fn.Option[msgmux.Router]
617

618
        // globalMsgRouter is a flag that indicates whether we have a global
619
        // msg router. If so, then we don't worry about stopping the msg router
620
        // when a peer disconnects.
621
        globalMsgRouter bool
622

623
        startReady chan struct{}
624

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

630
        // log is a peer-specific logging instance.
631
        log btclog.Logger
632
}
633

634
// A compile-time check to ensure that Brontide satisfies the lnpeer.Peer
635
// interface.
636
var _ lnpeer.Peer = (*Brontide)(nil)
637

638
// NewBrontide creates a new Brontide from a peer.Config struct.
639
func NewBrontide(cfg Config) *Brontide {
28✔
640
        logPrefix := fmt.Sprintf("Peer(%x):", cfg.PubKeyBytes)
28✔
641

28✔
642
        // We have a global message router if one was passed in via the config.
28✔
643
        // In this case, we don't need to attempt to tear it down when the peer
28✔
644
        // is stopped.
28✔
645
        globalMsgRouter := cfg.MsgRouter.IsSome()
28✔
646

28✔
647
        // We'll either use the msg router instance passed in, or create a new
28✔
648
        // blank instance.
28✔
649
        msgRouter := cfg.MsgRouter.Alt(fn.Some[msgmux.Router](
28✔
650
                msgmux.NewMultiMsgRouter(),
28✔
651
        ))
28✔
652

28✔
653
        p := &Brontide{
28✔
654
                cfg:           cfg,
28✔
655
                activeSignal:  make(chan struct{}),
28✔
656
                sendQueue:     make(chan outgoingMsg),
28✔
657
                outgoingQueue: make(chan outgoingMsg),
28✔
658
                addedChannels: &lnutils.SyncMap[lnwire.ChannelID, struct{}]{},
28✔
659
                activeChannels: &lnutils.SyncMap[
28✔
660
                        lnwire.ChannelID, *lnwallet.LightningChannel,
28✔
661
                ]{},
28✔
662
                newActiveChannel:     make(chan *newChannelMsg, 1),
28✔
663
                newPendingChannel:    make(chan *newChannelMsg, 1),
28✔
664
                removePendingChannel: make(chan *newChannelMsg),
28✔
665

28✔
666
                activeMsgStreams: make(map[lnwire.ChannelID]*msgStream),
28✔
667
                activeChanCloses: &lnutils.SyncMap[
28✔
668
                        lnwire.ChannelID, chanCloserFsm,
28✔
669
                ]{},
28✔
670
                localCloseChanReqs: make(chan *htlcswitch.ChanClose),
28✔
671
                linkFailures:       make(chan linkFailureReport),
28✔
672
                chanCloseMsgs:      make(chan *closeMsg),
28✔
673
                resentChanSyncMsg:  make(map[lnwire.ChannelID]struct{}),
28✔
674
                startReady:         make(chan struct{}),
28✔
675
                log:                peerLog.WithPrefix(logPrefix),
28✔
676
                msgRouter:          msgRouter,
28✔
677
                globalMsgRouter:    globalMsgRouter,
28✔
678
                cg:                 fn.NewContextGuard(),
28✔
679
        }
28✔
680

28✔
681
        if cfg.Conn != nil && cfg.Conn.RemoteAddr() != nil {
31✔
682
                remoteAddr := cfg.Conn.RemoteAddr().String()
3✔
683
                p.isTorConnection = strings.Contains(remoteAddr, ".onion") ||
3✔
684
                        strings.Contains(remoteAddr, "127.0.0.1")
3✔
685
        }
3✔
686

687
        var (
28✔
688
                lastBlockHeader           *wire.BlockHeader
28✔
689
                lastSerializedBlockHeader [wire.MaxBlockHeaderPayload]byte
28✔
690
        )
28✔
691
        newPingPayload := func() []byte {
28✔
692
                // We query the BestBlockHeader from our BestBlockView each time
×
693
                // this is called, and update our serialized block header if
×
694
                // they differ.  Over time, we'll use this to disseminate the
×
695
                // latest block header between all our peers, which can later be
×
696
                // used to cross-check our own view of the network to mitigate
×
697
                // various types of eclipse attacks.
×
698
                header, err := p.cfg.BestBlockView.BestBlockHeader()
×
699
                if err != nil && header == lastBlockHeader {
×
700
                        return lastSerializedBlockHeader[:]
×
701
                }
×
702

703
                buf := bytes.NewBuffer(lastSerializedBlockHeader[0:0])
×
704
                err = header.Serialize(buf)
×
705
                if err == nil {
×
706
                        lastBlockHeader = header
×
707
                } else {
×
708
                        p.log.Warn("unable to serialize current block" +
×
709
                                "header for ping payload generation." +
×
710
                                "This should be impossible and means" +
×
711
                                "there is an implementation bug.")
×
712
                }
×
713

714
                return lastSerializedBlockHeader[:]
×
715
        }
716

717
        // TODO(roasbeef): make dynamic in order to create fake cover traffic.
718
        //
719
        // NOTE(proofofkeags): this was changed to be dynamic to allow better
720
        // pong identification, however, more thought is needed to make this
721
        // actually usable as a traffic decoy.
722
        randPongSize := func() uint16 {
28✔
723
                return uint16(
×
724
                        // We don't need cryptographic randomness here.
×
725
                        /* #nosec */
×
726
                        rand.Intn(pongSizeCeiling) + 1,
×
727
                )
×
728
        }
×
729

730
        p.pingManager = NewPingManager(&PingManagerConfig{
28✔
731
                NewPingPayload:   newPingPayload,
28✔
732
                NewPongSize:      randPongSize,
28✔
733
                IntervalDuration: p.scaleTimeout(pingInterval),
28✔
734
                TimeoutDuration:  p.scaleTimeout(pingTimeout),
28✔
735
                SendPing: func(ping *lnwire.Ping) {
28✔
736
                        p.queueMsg(ping, nil)
×
737
                },
×
738
                OnPongFailure: func(err error) {
×
739
                        eStr := "pong response failure for %s: %v " +
×
740
                                "-- disconnecting"
×
741
                        p.log.Warnf(eStr, p, err)
×
742
                        go p.Disconnect(fmt.Errorf(eStr, p, err))
×
743
                },
×
744
        })
745

746
        return p
28✔
747
}
748

749
// Start starts all helper goroutines the peer needs for normal operations.  In
750
// the case this peer has already been started, then this function is a noop.
751
func (p *Brontide) Start() error {
6✔
752
        if atomic.AddInt32(&p.started, 1) != 1 {
6✔
753
                return nil
×
754
        }
×
755

756
        // Once we've finished starting up the peer, we'll signal to other
757
        // goroutines that the they can move forward to tear down the peer, or
758
        // carry out other relevant changes.
759
        defer close(p.startReady)
6✔
760

6✔
761
        p.log.Tracef("starting with conn[%v->%v]",
6✔
762
                p.cfg.Conn.LocalAddr(), p.cfg.Conn.RemoteAddr())
6✔
763

6✔
764
        // Fetch and then load all the active channels we have with this remote
6✔
765
        // peer from the database.
6✔
766
        activeChans, err := p.cfg.ChannelDB.FetchOpenChannels(
6✔
767
                p.cfg.Addr.IdentityKey,
6✔
768
        )
6✔
769
        if err != nil {
6✔
770
                p.log.Errorf("Unable to fetch active chans "+
×
771
                        "for peer: %v", err)
×
772
                return err
×
773
        }
×
774

775
        if len(activeChans) == 0 {
10✔
776
                go p.cfg.PrunePersistentPeerConnection(p.cfg.PubKeyBytes)
4✔
777
        }
4✔
778

779
        // Quickly check if we have any existing legacy channels with this
780
        // peer.
781
        haveLegacyChan := false
6✔
782
        for _, c := range activeChans {
11✔
783
                if c.ChanType.IsTweakless() {
10✔
784
                        continue
5✔
785
                }
786

787
                haveLegacyChan = true
3✔
788
                break
3✔
789
        }
790

791
        // Exchange local and global features, the init message should be very
792
        // first between two nodes.
793
        if err := p.sendInitMsg(haveLegacyChan); err != nil {
9✔
794
                return fmt.Errorf("unable to send init msg: %w", err)
3✔
795
        }
3✔
796

797
        // Before we launch any of the helper goroutines off the peer struct,
798
        // we'll first ensure proper adherence to the p2p protocol. The init
799
        // message MUST be sent before any other message.
800
        readErr := make(chan error, 1)
6✔
801
        msgChan := make(chan lnwire.Message, 1)
6✔
802
        p.cg.WgAdd(1)
6✔
803
        go func() {
12✔
804
                defer p.cg.WgDone()
6✔
805

6✔
806
                msg, err := p.readNextMessage()
6✔
807
                if err != nil {
6✔
UNCOV
808
                        readErr <- err
×
UNCOV
809
                        msgChan <- nil
×
UNCOV
810
                        return
×
UNCOV
811
                }
×
812
                readErr <- nil
6✔
813
                msgChan <- msg
6✔
814
        }()
815

816
        select {
6✔
817
        // In order to avoid blocking indefinitely, we'll give the other peer
818
        // an upper timeout to respond before we bail out early.
819
        case <-time.After(handshakeTimeout):
×
820
                return fmt.Errorf("peer did not complete handshake within %v",
×
821
                        handshakeTimeout)
×
822
        case err := <-readErr:
6✔
823
                if err != nil {
6✔
UNCOV
824
                        return fmt.Errorf("unable to read init msg: %w", err)
×
UNCOV
825
                }
×
826
        }
827

828
        // Once the init message arrives, we can parse it so we can figure out
829
        // the negotiation of features for this session.
830
        msg := <-msgChan
6✔
831
        if msg, ok := msg.(*lnwire.Init); ok {
12✔
832
                if err := p.handleInitMsg(msg); err != nil {
6✔
833
                        p.storeError(err)
×
834
                        return err
×
835
                }
×
836
        } else {
×
837
                return errors.New("very first message between nodes " +
×
838
                        "must be init message")
×
839
        }
×
840

841
        // Next, load all the active channels we have with this peer,
842
        // registering them with the switch and launching the necessary
843
        // goroutines required to operate them.
844
        p.log.Debugf("Loaded %v active channels from database",
6✔
845
                len(activeChans))
6✔
846

6✔
847
        // Conditionally subscribe to channel events before loading channels so
6✔
848
        // we won't miss events. This subscription is used to listen to active
6✔
849
        // channel event when reenabling channels. Once the reenabling process
6✔
850
        // is finished, this subscription will be canceled.
6✔
851
        //
6✔
852
        // NOTE: ChannelNotifier must be started before subscribing events
6✔
853
        // otherwise we'd panic here.
6✔
854
        if err := p.attachChannelEventSubscription(); err != nil {
6✔
855
                return err
×
856
        }
×
857

858
        // Register the message router now as we may need to register some
859
        // endpoints while loading the channels below.
860
        p.msgRouter.WhenSome(func(router msgmux.Router) {
12✔
861
                router.Start(context.Background())
6✔
862
        })
6✔
863

864
        msgs, err := p.loadActiveChannels(activeChans)
6✔
865
        if err != nil {
6✔
866
                return fmt.Errorf("unable to load channels: %w", err)
×
867
        }
×
868

869
        p.startTime = time.Now()
6✔
870

6✔
871
        // Before launching the writeHandler goroutine, we send any channel
6✔
872
        // sync messages that must be resent for borked channels. We do this to
6✔
873
        // avoid data races with WriteMessage & Flush calls.
6✔
874
        if len(msgs) > 0 {
11✔
875
                p.log.Infof("Sending %d channel sync messages to peer after "+
5✔
876
                        "loading active channels", len(msgs))
5✔
877

5✔
878
                // Send the messages directly via writeMessage and bypass the
5✔
879
                // writeHandler goroutine.
5✔
880
                for _, msg := range msgs {
10✔
881
                        if err := p.writeMessage(msg); err != nil {
5✔
882
                                return fmt.Errorf("unable to send "+
×
883
                                        "reestablish msg: %v", err)
×
884
                        }
×
885
                }
886
        }
887

888
        err = p.pingManager.Start()
6✔
889
        if err != nil {
6✔
890
                return fmt.Errorf("could not start ping manager %w", err)
×
891
        }
×
892

893
        p.cg.WgAdd(4)
6✔
894
        go p.queueHandler()
6✔
895
        go p.writeHandler()
6✔
896
        go p.channelManager()
6✔
897
        go p.readHandler()
6✔
898

6✔
899
        // Signal to any external processes that the peer is now active.
6✔
900
        close(p.activeSignal)
6✔
901

6✔
902
        // Node announcements don't propagate very well throughout the network
6✔
903
        // as there isn't a way to efficiently query for them through their
6✔
904
        // timestamp, mostly affecting nodes that were offline during the time
6✔
905
        // of broadcast. We'll resend our node announcement to the remote peer
6✔
906
        // as a best-effort delivery such that it can also propagate to their
6✔
907
        // peers. To ensure they can successfully process it in most cases,
6✔
908
        // we'll only resend it as long as we have at least one confirmed
6✔
909
        // advertised channel with the remote peer.
6✔
910
        //
6✔
911
        // TODO(wilmer): Remove this once we're able to query for node
6✔
912
        // announcements through their timestamps.
6✔
913
        p.cg.WgAdd(2)
6✔
914
        go p.maybeSendNodeAnn(activeChans)
6✔
915
        go p.maybeSendChannelUpdates()
6✔
916

6✔
917
        return nil
6✔
918
}
919

920
// initGossipSync initializes either a gossip syncer or an initial routing
921
// dump, depending on the negotiated synchronization method.
922
func (p *Brontide) initGossipSync() {
6✔
923
        // If the remote peer knows of the new gossip queries feature, then
6✔
924
        // we'll create a new gossipSyncer in the AuthenticatedGossiper for it.
6✔
925
        if p.remoteFeatures.HasFeature(lnwire.GossipQueriesOptional) {
12✔
926
                p.log.Info("Negotiated chan series queries")
6✔
927

6✔
928
                if p.cfg.AuthGossiper == nil {
9✔
929
                        // This should only ever be hit in the unit tests.
3✔
930
                        p.log.Warn("No AuthGossiper configured. Abandoning " +
3✔
931
                                "gossip sync.")
3✔
932
                        return
3✔
933
                }
3✔
934

935
                // Register the peer's gossip syncer with the gossiper.
936
                // This blocks synchronously to ensure the gossip syncer is
937
                // registered with the gossiper before attempting to read
938
                // messages from the remote peer.
939
                //
940
                // TODO(wilmer): Only sync updates from non-channel peers. This
941
                // requires an improved version of the current network
942
                // bootstrapper to ensure we can find and connect to non-channel
943
                // peers.
944
                p.cfg.AuthGossiper.InitSyncState(p)
3✔
945
        }
946
}
947

948
// taprootShutdownAllowed returns true if both parties have negotiated the
949
// shutdown-any-segwit feature.
950
func (p *Brontide) taprootShutdownAllowed() bool {
9✔
951
        return p.RemoteFeatures().HasFeature(lnwire.ShutdownAnySegwitOptional) &&
9✔
952
                p.LocalFeatures().HasFeature(lnwire.ShutdownAnySegwitOptional)
9✔
953
}
9✔
954

955
// rbfCoopCloseAllowed returns true if both parties have negotiated the new RBF
956
// coop close feature.
957
func (p *Brontide) rbfCoopCloseAllowed() bool {
10✔
958
        return p.RemoteFeatures().HasFeature(
10✔
959
                lnwire.RbfCoopCloseOptionalStaging,
10✔
960
        ) && p.LocalFeatures().HasFeature(
10✔
961
                lnwire.RbfCoopCloseOptionalStaging,
10✔
962
        )
10✔
963
}
10✔
964

965
// QuitSignal is a method that should return a channel which will be sent upon
966
// or closed once the backing peer exits. This allows callers using the
967
// interface to cancel any processing in the event the backing implementation
968
// exits.
969
//
970
// NOTE: Part of the lnpeer.Peer interface.
971
func (p *Brontide) QuitSignal() <-chan struct{} {
3✔
972
        return p.cg.Done()
3✔
973
}
3✔
974

975
// addrWithInternalKey takes a delivery script, then attempts to supplement it
976
// with information related to the internal key for the addr, but only if it's
977
// a taproot addr.
978
func (p *Brontide) addrWithInternalKey(
979
        deliveryScript []byte) (*chancloser.DeliveryAddrWithKey, error) {
12✔
980

12✔
981
        // Currently, custom channels cannot be created with external upfront
12✔
982
        // shutdown addresses, so this shouldn't be an issue. We only require
12✔
983
        // the internal key for taproot addresses to be able to provide a non
12✔
984
        // inclusion proof of any scripts.
12✔
985
        internalKeyDesc, err := lnwallet.InternalKeyForAddr(
12✔
986
                p.cfg.Wallet, &p.cfg.Wallet.Cfg.NetParams, deliveryScript,
12✔
987
        )
12✔
988
        if err != nil {
12✔
989
                return nil, fmt.Errorf("unable to fetch internal key: %w", err)
×
990
        }
×
991

992
        return &chancloser.DeliveryAddrWithKey{
12✔
993
                DeliveryAddress: deliveryScript,
12✔
994
                InternalKey: fn.MapOption(
12✔
995
                        func(desc keychain.KeyDescriptor) btcec.PublicKey {
15✔
996
                                return *desc.PubKey
3✔
997
                        },
3✔
998
                )(internalKeyDesc),
999
        }, nil
1000
}
1001

1002
// loadActiveChannels creates indexes within the peer for tracking all active
1003
// channels returned by the database. It returns a slice of channel reestablish
1004
// messages that should be sent to the peer immediately, in case we have borked
1005
// channels that haven't been closed yet.
1006
func (p *Brontide) loadActiveChannels(chans []*channeldb.OpenChannel) (
1007
        []lnwire.Message, error) {
6✔
1008

6✔
1009
        // Return a slice of messages to send to the peers in case the channel
6✔
1010
        // cannot be loaded normally.
6✔
1011
        var msgs []lnwire.Message
6✔
1012

6✔
1013
        scidAliasNegotiated := p.hasNegotiatedScidAlias()
6✔
1014

6✔
1015
        for _, dbChan := range chans {
11✔
1016
                hasScidFeature := dbChan.ChanType.HasScidAliasFeature()
5✔
1017
                if scidAliasNegotiated && !hasScidFeature {
8✔
1018
                        // We'll request and store an alias, making sure that a
3✔
1019
                        // gossiper mapping is not created for the alias to the
3✔
1020
                        // real SCID. This is done because the peer and funding
3✔
1021
                        // manager are not aware of each other's states and if
3✔
1022
                        // we did not do this, we would accept alias channel
3✔
1023
                        // updates after 6 confirmations, which would be buggy.
3✔
1024
                        // We'll queue a channel_ready message with the new
3✔
1025
                        // alias. This should technically be done *after* the
3✔
1026
                        // reestablish, but this behavior is pre-existing since
3✔
1027
                        // the funding manager may already queue a
3✔
1028
                        // channel_ready before the channel_reestablish.
3✔
1029
                        if !dbChan.IsPending {
6✔
1030
                                aliasScid, err := p.cfg.RequestAlias()
3✔
1031
                                if err != nil {
3✔
1032
                                        return nil, err
×
1033
                                }
×
1034

1035
                                err = p.cfg.AddLocalAlias(
3✔
1036
                                        aliasScid, dbChan.ShortChanID(), false,
3✔
1037
                                        false,
3✔
1038
                                )
3✔
1039
                                if err != nil {
3✔
1040
                                        return nil, err
×
1041
                                }
×
1042

1043
                                chanID := lnwire.NewChanIDFromOutPoint(
3✔
1044
                                        dbChan.FundingOutpoint,
3✔
1045
                                )
3✔
1046

3✔
1047
                                // Fetch the second commitment point to send in
3✔
1048
                                // the channel_ready message.
3✔
1049
                                second, err := dbChan.SecondCommitmentPoint()
3✔
1050
                                if err != nil {
3✔
1051
                                        return nil, err
×
1052
                                }
×
1053

1054
                                channelReadyMsg := lnwire.NewChannelReady(
3✔
1055
                                        chanID, second,
3✔
1056
                                )
3✔
1057
                                channelReadyMsg.AliasScid = &aliasScid
3✔
1058

3✔
1059
                                msgs = append(msgs, channelReadyMsg)
3✔
1060
                        }
1061

1062
                        // If we've negotiated the option-scid-alias feature
1063
                        // and this channel does not have ScidAliasFeature set
1064
                        // to true due to an upgrade where the feature bit was
1065
                        // turned on, we'll update the channel's database
1066
                        // state.
1067
                        err := dbChan.MarkScidAliasNegotiated()
3✔
1068
                        if err != nil {
3✔
1069
                                return nil, err
×
1070
                        }
×
1071
                }
1072

1073
                var chanOpts []lnwallet.ChannelOpt
5✔
1074
                p.cfg.AuxLeafStore.WhenSome(func(s lnwallet.AuxLeafStore) {
5✔
1075
                        chanOpts = append(chanOpts, lnwallet.WithLeafStore(s))
×
1076
                })
×
1077
                p.cfg.AuxSigner.WhenSome(func(s lnwallet.AuxSigner) {
5✔
1078
                        chanOpts = append(chanOpts, lnwallet.WithAuxSigner(s))
×
1079
                })
×
1080
                p.cfg.AuxResolver.WhenSome(
5✔
1081
                        func(s lnwallet.AuxContractResolver) {
5✔
1082
                                chanOpts = append(
×
1083
                                        chanOpts, lnwallet.WithAuxResolver(s),
×
1084
                                )
×
1085
                        },
×
1086
                )
1087

1088
                lnChan, err := lnwallet.NewLightningChannel(
5✔
1089
                        p.cfg.Signer, dbChan, p.cfg.SigPool, chanOpts...,
5✔
1090
                )
5✔
1091
                if err != nil {
5✔
1092
                        return nil, fmt.Errorf("unable to create channel "+
×
1093
                                "state machine: %w", err)
×
1094
                }
×
1095

1096
                chanPoint := dbChan.FundingOutpoint
5✔
1097

5✔
1098
                chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
5✔
1099

5✔
1100
                p.log.Infof("Loading ChannelPoint(%v), isPending=%v",
5✔
1101
                        chanPoint, lnChan.IsPending())
5✔
1102

5✔
1103
                // Skip adding any permanently irreconcilable channels to the
5✔
1104
                // htlcswitch.
5✔
1105
                if !dbChan.HasChanStatus(channeldb.ChanStatusDefault) &&
5✔
1106
                        !dbChan.HasChanStatus(channeldb.ChanStatusRestored) {
10✔
1107

5✔
1108
                        p.log.Warnf("ChannelPoint(%v) has status %v, won't "+
5✔
1109
                                "start.", chanPoint, dbChan.ChanStatus())
5✔
1110

5✔
1111
                        // To help our peer recover from a potential data loss,
5✔
1112
                        // we resend our channel reestablish message if the
5✔
1113
                        // channel is in a borked state. We won't process any
5✔
1114
                        // channel reestablish message sent from the peer, but
5✔
1115
                        // that's okay since the assumption is that we did when
5✔
1116
                        // marking the channel borked.
5✔
1117
                        chanSync, err := dbChan.ChanSyncMsg()
5✔
1118
                        if err != nil {
5✔
1119
                                p.log.Errorf("Unable to create channel "+
×
1120
                                        "reestablish message for channel %v: "+
×
1121
                                        "%v", chanPoint, err)
×
1122
                                continue
×
1123
                        }
1124

1125
                        msgs = append(msgs, chanSync)
5✔
1126

5✔
1127
                        // Check if this channel needs to have the cooperative
5✔
1128
                        // close process restarted. If so, we'll need to send
5✔
1129
                        // the Shutdown message that is returned.
5✔
1130
                        if dbChan.HasChanStatus(
5✔
1131
                                channeldb.ChanStatusCoopBroadcasted,
5✔
1132
                        ) {
8✔
1133

3✔
1134
                                shutdownMsg, err := p.restartCoopClose(lnChan)
3✔
1135
                                if err != nil {
3✔
1136
                                        p.log.Errorf("Unable to restart "+
×
1137
                                                "coop close for channel: %v",
×
1138
                                                err)
×
1139
                                        continue
×
1140
                                }
1141

1142
                                if shutdownMsg == nil {
6✔
1143
                                        continue
3✔
1144
                                }
1145

1146
                                // Append the message to the set of messages to
1147
                                // send.
1148
                                msgs = append(msgs, shutdownMsg)
×
1149
                        }
1150

1151
                        continue
5✔
1152
                }
1153

1154
                // Before we register this new link with the HTLC Switch, we'll
1155
                // need to fetch its current link-layer forwarding policy from
1156
                // the database.
1157
                graph := p.cfg.ChannelGraph
3✔
1158
                info, p1, p2, err := graph.FetchChannelEdgesByOutpoint(
3✔
1159
                        &chanPoint,
3✔
1160
                )
3✔
1161
                if err != nil && !errors.Is(err, graphdb.ErrEdgeNotFound) {
3✔
1162
                        return nil, err
×
1163
                }
×
1164

1165
                // We'll filter out our policy from the directional channel
1166
                // edges based whom the edge connects to. If it doesn't connect
1167
                // to us, then we know that we were the one that advertised the
1168
                // policy.
1169
                //
1170
                // TODO(roasbeef): can add helper method to get policy for
1171
                // particular channel.
1172
                var selfPolicy *models.ChannelEdgePolicy
3✔
1173
                if info != nil && bytes.Equal(info.NodeKey1Bytes[:],
3✔
1174
                        p.cfg.ServerPubKey[:]) {
6✔
1175

3✔
1176
                        selfPolicy = p1
3✔
1177
                } else {
6✔
1178
                        selfPolicy = p2
3✔
1179
                }
3✔
1180

1181
                // If we don't yet have an advertised routing policy, then
1182
                // we'll use the current default, otherwise we'll translate the
1183
                // routing policy into a forwarding policy.
1184
                var forwardingPolicy *models.ForwardingPolicy
3✔
1185
                if selfPolicy != nil {
6✔
1186
                        var inboundWireFee lnwire.Fee
3✔
1187
                        _, err := selfPolicy.ExtraOpaqueData.ExtractRecords(
3✔
1188
                                &inboundWireFee,
3✔
1189
                        )
3✔
1190
                        if err != nil {
3✔
1191
                                return nil, err
×
1192
                        }
×
1193

1194
                        inboundFee := models.NewInboundFeeFromWire(
3✔
1195
                                inboundWireFee,
3✔
1196
                        )
3✔
1197

3✔
1198
                        forwardingPolicy = &models.ForwardingPolicy{
3✔
1199
                                MinHTLCOut:    selfPolicy.MinHTLC,
3✔
1200
                                MaxHTLC:       selfPolicy.MaxHTLC,
3✔
1201
                                BaseFee:       selfPolicy.FeeBaseMSat,
3✔
1202
                                FeeRate:       selfPolicy.FeeProportionalMillionths,
3✔
1203
                                TimeLockDelta: uint32(selfPolicy.TimeLockDelta),
3✔
1204

3✔
1205
                                InboundFee: inboundFee,
3✔
1206
                        }
3✔
1207
                } else {
3✔
1208
                        p.log.Warnf("Unable to find our forwarding policy "+
3✔
1209
                                "for channel %v, using default values",
3✔
1210
                                chanPoint)
3✔
1211
                        forwardingPolicy = &p.cfg.RoutingPolicy
3✔
1212
                }
3✔
1213

1214
                p.log.Tracef("Using link policy of: %v",
3✔
1215
                        spew.Sdump(forwardingPolicy))
3✔
1216

3✔
1217
                // If the channel is pending, set the value to nil in the
3✔
1218
                // activeChannels map. This is done to signify that the channel
3✔
1219
                // is pending. We don't add the link to the switch here - it's
3✔
1220
                // the funding manager's responsibility to spin up pending
3✔
1221
                // channels. Adding them here would just be extra work as we'll
3✔
1222
                // tear them down when creating + adding the final link.
3✔
1223
                if lnChan.IsPending() {
6✔
1224
                        p.activeChannels.Store(chanID, nil)
3✔
1225

3✔
1226
                        continue
3✔
1227
                }
1228

1229
                shutdownInfo, err := lnChan.State().ShutdownInfo()
3✔
1230
                if err != nil && !errors.Is(err, channeldb.ErrNoShutdownInfo) {
3✔
1231
                        return nil, err
×
1232
                }
×
1233

1234
                isTaprootChan := lnChan.ChanType().IsTaproot()
3✔
1235

3✔
1236
                var (
3✔
1237
                        shutdownMsg     fn.Option[lnwire.Shutdown]
3✔
1238
                        shutdownInfoErr error
3✔
1239
                )
3✔
1240
                shutdownInfo.WhenSome(func(info channeldb.ShutdownInfo) {
6✔
1241
                        // If we can use the new RBF close feature, we don't
3✔
1242
                        // need to create the legacy closer. However for taproot
3✔
1243
                        // channels, we'll continue to use the legacy closer.
3✔
1244
                        if p.rbfCoopCloseAllowed() && !isTaprootChan {
6✔
1245
                                return
3✔
1246
                        }
3✔
1247

1248
                        // Compute an ideal fee.
1249
                        feePerKw, err := p.cfg.FeeEstimator.EstimateFeePerKW(
3✔
1250
                                p.cfg.CoopCloseTargetConfs,
3✔
1251
                        )
3✔
1252
                        if err != nil {
3✔
1253
                                shutdownInfoErr = fmt.Errorf("unable to "+
×
1254
                                        "estimate fee: %w", err)
×
1255

×
1256
                                return
×
1257
                        }
×
1258

1259
                        addr, err := p.addrWithInternalKey(
3✔
1260
                                info.DeliveryScript.Val,
3✔
1261
                        )
3✔
1262
                        if err != nil {
3✔
1263
                                shutdownInfoErr = fmt.Errorf("unable to make "+
×
1264
                                        "delivery addr: %w", err)
×
1265
                                return
×
1266
                        }
×
1267
                        negotiateChanCloser, err := p.createChanCloser(
3✔
1268
                                lnChan, addr, feePerKw, nil,
3✔
1269
                                info.Closer(),
3✔
1270
                        )
3✔
1271
                        if err != nil {
3✔
1272
                                shutdownInfoErr = fmt.Errorf("unable to "+
×
1273
                                        "create chan closer: %w", err)
×
1274

×
1275
                                return
×
1276
                        }
×
1277

1278
                        chanID := lnwire.NewChanIDFromOutPoint(
3✔
1279
                                lnChan.State().FundingOutpoint,
3✔
1280
                        )
3✔
1281

3✔
1282
                        p.activeChanCloses.Store(chanID, makeNegotiateCloser(
3✔
1283
                                negotiateChanCloser,
3✔
1284
                        ))
3✔
1285

3✔
1286
                        // Create the Shutdown message.
3✔
1287
                        shutdown, err := negotiateChanCloser.ShutdownChan()
3✔
1288
                        if err != nil {
3✔
1289
                                p.activeChanCloses.Delete(chanID)
×
1290
                                shutdownInfoErr = err
×
1291

×
1292
                                return
×
1293
                        }
×
1294

1295
                        shutdownMsg = fn.Some(*shutdown)
3✔
1296
                })
1297
                if shutdownInfoErr != nil {
3✔
1298
                        return nil, shutdownInfoErr
×
1299
                }
×
1300

1301
                // Subscribe to the set of on-chain events for this channel.
1302
                chainEvents, err := p.cfg.ChainArb.SubscribeChannelEvents(
3✔
1303
                        chanPoint,
3✔
1304
                )
3✔
1305
                if err != nil {
3✔
1306
                        return nil, err
×
1307
                }
×
1308

1309
                err = p.addLink(
3✔
1310
                        &chanPoint, lnChan, forwardingPolicy, chainEvents,
3✔
1311
                        true, shutdownMsg,
3✔
1312
                )
3✔
1313
                if err != nil {
3✔
1314
                        return nil, fmt.Errorf("unable to add link %v to "+
×
1315
                                "switch: %v", chanPoint, err)
×
1316
                }
×
1317

1318
                p.activeChannels.Store(chanID, lnChan)
3✔
1319

3✔
1320
                // We're using the old co-op close, so we don't need to init
3✔
1321
                // the new RBF chan closer. If we have a taproot chan, then
3✔
1322
                // we'll also use the legacy type, so we don't need to make the
3✔
1323
                // new closer.
3✔
1324
                if !p.rbfCoopCloseAllowed() || isTaprootChan {
6✔
1325
                        continue
3✔
1326
                }
1327

1328
                // Now that the link has been added above, we'll also init an
1329
                // RBF chan closer for this channel, but only if the new close
1330
                // feature is negotiated.
1331
                //
1332
                // Creating this here ensures that any shutdown messages sent
1333
                // will be automatically routed by the msg router.
1334
                if _, err := p.initRbfChanCloser(lnChan); err != nil {
3✔
1335
                        p.activeChanCloses.Delete(chanID)
×
1336

×
1337
                        return nil, fmt.Errorf("unable to init RBF chan "+
×
1338
                                "closer during peer connect: %w", err)
×
1339
                }
×
1340

1341
                // If the shutdown info isn't blank, then we should kick things
1342
                // off by sending a shutdown message to the remote party to
1343
                // continue the old shutdown flow.
1344
                restartShutdown := func(s channeldb.ShutdownInfo) error {
6✔
1345
                        return p.startRbfChanCloser(
3✔
1346
                                newRestartShutdownInit(s),
3✔
1347
                                lnChan.ChannelPoint(),
3✔
1348
                        )
3✔
1349
                }
3✔
1350
                err = fn.MapOptionZ(shutdownInfo, restartShutdown)
3✔
1351
                if err != nil {
3✔
1352
                        return nil, fmt.Errorf("unable to start RBF "+
×
1353
                                "chan closer: %w", err)
×
1354
                }
×
1355
        }
1356

1357
        return msgs, nil
6✔
1358
}
1359

1360
// addLink creates and adds a new ChannelLink from the specified channel.
1361
func (p *Brontide) addLink(chanPoint *wire.OutPoint,
1362
        lnChan *lnwallet.LightningChannel,
1363
        forwardingPolicy *models.ForwardingPolicy,
1364
        chainEvents *contractcourt.ChainEventSubscription,
1365
        syncStates bool, shutdownMsg fn.Option[lnwire.Shutdown]) error {
3✔
1366

3✔
1367
        // onChannelFailure will be called by the link in case the channel
3✔
1368
        // fails for some reason.
3✔
1369
        onChannelFailure := func(chanID lnwire.ChannelID,
3✔
1370
                shortChanID lnwire.ShortChannelID,
3✔
1371
                linkErr htlcswitch.LinkFailureError) {
6✔
1372

3✔
1373
                failure := linkFailureReport{
3✔
1374
                        chanPoint:   *chanPoint,
3✔
1375
                        chanID:      chanID,
3✔
1376
                        shortChanID: shortChanID,
3✔
1377
                        linkErr:     linkErr,
3✔
1378
                }
3✔
1379

3✔
1380
                select {
3✔
1381
                case p.linkFailures <- failure:
3✔
1382
                case <-p.cg.Done():
×
1383
                case <-p.cfg.Quit:
×
1384
                }
1385
        }
1386

1387
        updateContractSignals := func(signals *contractcourt.ContractSignals) error {
6✔
1388
                return p.cfg.ChainArb.UpdateContractSignals(*chanPoint, signals)
3✔
1389
        }
3✔
1390

1391
        notifyContractUpdate := func(update *contractcourt.ContractUpdate) error {
6✔
1392
                return p.cfg.ChainArb.NotifyContractUpdate(*chanPoint, update)
3✔
1393
        }
3✔
1394

1395
        //nolint:ll
1396
        linkCfg := htlcswitch.ChannelLinkConfig{
3✔
1397
                Peer:                   p,
3✔
1398
                DecodeHopIterators:     p.cfg.Sphinx.DecodeHopIterators,
3✔
1399
                ExtractErrorEncrypter:  p.cfg.Sphinx.ExtractErrorEncrypter,
3✔
1400
                FetchLastChannelUpdate: p.cfg.FetchLastChanUpdate,
3✔
1401
                HodlMask:               p.cfg.Hodl.Mask(),
3✔
1402
                Registry:               p.cfg.Invoices,
3✔
1403
                BestHeight:             p.cfg.Switch.BestHeight,
3✔
1404
                Circuits:               p.cfg.Switch.CircuitModifier(),
3✔
1405
                ForwardPackets:         p.cfg.InterceptSwitch.ForwardPackets,
3✔
1406
                FwrdingPolicy:          *forwardingPolicy,
3✔
1407
                FeeEstimator:           p.cfg.FeeEstimator,
3✔
1408
                PreimageCache:          p.cfg.WitnessBeacon,
3✔
1409
                ChainEvents:            chainEvents,
3✔
1410
                UpdateContractSignals:  updateContractSignals,
3✔
1411
                NotifyContractUpdate:   notifyContractUpdate,
3✔
1412
                OnChannelFailure:       onChannelFailure,
3✔
1413
                SyncStates:             syncStates,
3✔
1414
                BatchTicker:            ticker.New(p.cfg.ChannelCommitInterval),
3✔
1415
                FwdPkgGCTicker:         ticker.New(time.Hour),
3✔
1416
                PendingCommitTicker: ticker.New(
3✔
1417
                        p.cfg.PendingCommitInterval,
3✔
1418
                ),
3✔
1419
                BatchSize:               p.cfg.ChannelCommitBatchSize,
3✔
1420
                UnsafeReplay:            p.cfg.UnsafeReplay,
3✔
1421
                MinUpdateTimeout:        htlcswitch.DefaultMinLinkFeeUpdateTimeout,
3✔
1422
                MaxUpdateTimeout:        htlcswitch.DefaultMaxLinkFeeUpdateTimeout,
3✔
1423
                OutgoingCltvRejectDelta: p.cfg.OutgoingCltvRejectDelta,
3✔
1424
                TowerClient:             p.cfg.TowerClient,
3✔
1425
                MaxOutgoingCltvExpiry:   p.cfg.MaxOutgoingCltvExpiry,
3✔
1426
                MaxFeeAllocation:        p.cfg.MaxChannelFeeAllocation,
3✔
1427
                MaxAnchorsCommitFeeRate: p.cfg.MaxAnchorsCommitFeeRate,
3✔
1428
                NotifyActiveLink:        p.cfg.ChannelNotifier.NotifyActiveLinkEvent,
3✔
1429
                NotifyActiveChannel:     p.cfg.ChannelNotifier.NotifyActiveChannelEvent,
3✔
1430
                NotifyInactiveChannel:   p.cfg.ChannelNotifier.NotifyInactiveChannelEvent,
3✔
1431
                NotifyInactiveLinkEvent: p.cfg.ChannelNotifier.NotifyInactiveLinkEvent,
3✔
1432
                HtlcNotifier:            p.cfg.HtlcNotifier,
3✔
1433
                GetAliases:              p.cfg.GetAliases,
3✔
1434
                PreviouslySentShutdown:  shutdownMsg,
3✔
1435
                DisallowRouteBlinding:   p.cfg.DisallowRouteBlinding,
3✔
1436
                MaxFeeExposure:          p.cfg.MaxFeeExposure,
3✔
1437
                ShouldFwdExpEndorsement: p.cfg.ShouldFwdExpEndorsement,
3✔
1438
                DisallowQuiescence: p.cfg.DisallowQuiescence ||
3✔
1439
                        !p.remoteFeatures.HasFeature(lnwire.QuiescenceOptional),
3✔
1440
                AuxTrafficShaper: p.cfg.AuxTrafficShaper,
3✔
1441
        }
3✔
1442

3✔
1443
        // Before adding our new link, purge the switch of any pending or live
3✔
1444
        // links going by the same channel id. If one is found, we'll shut it
3✔
1445
        // down to ensure that the mailboxes are only ever under the control of
3✔
1446
        // one link.
3✔
1447
        chanID := lnwire.NewChanIDFromOutPoint(*chanPoint)
3✔
1448
        p.cfg.Switch.RemoveLink(chanID)
3✔
1449

3✔
1450
        // With the channel link created, we'll now notify the htlc switch so
3✔
1451
        // this channel can be used to dispatch local payments and also
3✔
1452
        // passively forward payments.
3✔
1453
        return p.cfg.Switch.CreateAndAddLink(linkCfg, lnChan)
3✔
1454
}
1455

1456
// maybeSendNodeAnn sends our node announcement to the remote peer if at least
1457
// one confirmed public channel exists with them.
1458
func (p *Brontide) maybeSendNodeAnn(channels []*channeldb.OpenChannel) {
6✔
1459
        defer p.cg.WgDone()
6✔
1460

6✔
1461
        hasConfirmedPublicChan := false
6✔
1462
        for _, channel := range channels {
11✔
1463
                if channel.IsPending {
8✔
1464
                        continue
3✔
1465
                }
1466
                if channel.ChannelFlags&lnwire.FFAnnounceChannel == 0 {
10✔
1467
                        continue
5✔
1468
                }
1469

1470
                hasConfirmedPublicChan = true
3✔
1471
                break
3✔
1472
        }
1473
        if !hasConfirmedPublicChan {
12✔
1474
                return
6✔
1475
        }
6✔
1476

1477
        ourNodeAnn, err := p.cfg.GenNodeAnnouncement()
3✔
1478
        if err != nil {
3✔
1479
                p.log.Debugf("Unable to retrieve node announcement: %v", err)
×
1480
                return
×
1481
        }
×
1482

1483
        if err := p.SendMessageLazy(false, &ourNodeAnn); err != nil {
3✔
1484
                p.log.Debugf("Unable to resend node announcement: %v", err)
×
1485
        }
×
1486
}
1487

1488
// maybeSendChannelUpdates sends our channel updates to the remote peer if we
1489
// have any active channels with them.
1490
func (p *Brontide) maybeSendChannelUpdates() {
6✔
1491
        defer p.cg.WgDone()
6✔
1492

6✔
1493
        // If we don't have any active channels, then we can exit early.
6✔
1494
        if p.activeChannels.Len() == 0 {
10✔
1495
                return
4✔
1496
        }
4✔
1497

1498
        maybeSendUpd := func(cid lnwire.ChannelID,
5✔
1499
                lnChan *lnwallet.LightningChannel) error {
10✔
1500

5✔
1501
                // Nil channels are pending, so we'll skip them.
5✔
1502
                if lnChan == nil {
8✔
1503
                        return nil
3✔
1504
                }
3✔
1505

1506
                dbChan := lnChan.State()
5✔
1507
                scid := func() lnwire.ShortChannelID {
10✔
1508
                        switch {
5✔
1509
                        // Otherwise if it's a zero conf channel and confirmed,
1510
                        // then we need to use the "real" scid.
1511
                        case dbChan.IsZeroConf() && dbChan.ZeroConfConfirmed():
3✔
1512
                                return dbChan.ZeroConfRealScid()
3✔
1513

1514
                        // Otherwise, we can use the normal scid.
1515
                        default:
5✔
1516
                                return dbChan.ShortChanID()
5✔
1517
                        }
1518
                }()
1519

1520
                // Now that we know the channel is in a good state, we'll try
1521
                // to fetch the update to send to the remote peer. If the
1522
                // channel is pending, and not a zero conf channel, we'll get
1523
                // an error here which we'll ignore.
1524
                chanUpd, err := p.cfg.FetchLastChanUpdate(scid)
5✔
1525
                if err != nil {
8✔
1526
                        p.log.Debugf("Unable to fetch channel update for "+
3✔
1527
                                "ChannelPoint(%v), scid=%v: %v",
3✔
1528
                                dbChan.FundingOutpoint, dbChan.ShortChanID, err)
3✔
1529

3✔
1530
                        return nil
3✔
1531
                }
3✔
1532

1533
                p.log.Debugf("Sending channel update for ChannelPoint(%v), "+
5✔
1534
                        "scid=%v", dbChan.FundingOutpoint, dbChan.ShortChanID)
5✔
1535

5✔
1536
                // We'll send it as a normal message instead of using the lazy
5✔
1537
                // queue to prioritize transmission of the fresh update.
5✔
1538
                if err := p.SendMessage(false, chanUpd); err != nil {
5✔
1539
                        err := fmt.Errorf("unable to send channel update for "+
×
1540
                                "ChannelPoint(%v), scid=%v: %w",
×
1541
                                dbChan.FundingOutpoint, dbChan.ShortChanID(),
×
1542
                                err)
×
1543
                        p.log.Errorf(err.Error())
×
1544

×
1545
                        return err
×
1546
                }
×
1547

1548
                return nil
5✔
1549
        }
1550

1551
        p.activeChannels.ForEach(maybeSendUpd)
5✔
1552
}
1553

1554
// WaitForDisconnect waits until the peer has disconnected. A peer may be
1555
// disconnected if the local or remote side terminates the connection, or an
1556
// irrecoverable protocol error has been encountered. This method will only
1557
// begin watching the peer's waitgroup after the ready channel or the peer's
1558
// quit channel are signaled. The ready channel should only be signaled if a
1559
// call to Start returns no error. Otherwise, if the peer fails to start,
1560
// calling Disconnect will signal the quit channel and the method will not
1561
// block, since no goroutines were spawned.
1562
func (p *Brontide) WaitForDisconnect(ready chan struct{}) {
3✔
1563
        // Before we try to call the `Wait` goroutine, we'll make sure the main
3✔
1564
        // set of goroutines are already active.
3✔
1565
        select {
3✔
1566
        case <-p.startReady:
3✔
1567
        case <-p.cg.Done():
×
1568
                return
×
1569
        }
1570

1571
        select {
3✔
1572
        case <-ready:
3✔
1573
        case <-p.cg.Done():
3✔
1574
        }
1575

1576
        p.cg.WgWait()
3✔
1577
}
1578

1579
// Disconnect terminates the connection with the remote peer. Additionally, a
1580
// signal is sent to the server and htlcSwitch indicating the resources
1581
// allocated to the peer can now be cleaned up.
1582
func (p *Brontide) Disconnect(reason error) {
3✔
1583
        if !atomic.CompareAndSwapInt32(&p.disconnect, 0, 1) {
6✔
1584
                return
3✔
1585
        }
3✔
1586

1587
        // Make sure initialization has completed before we try to tear things
1588
        // down.
1589
        //
1590
        // NOTE: We only read the `startReady` chan if the peer has been
1591
        // started, otherwise we will skip reading it as this chan won't be
1592
        // closed, hence blocks forever.
1593
        if atomic.LoadInt32(&p.started) == 1 {
6✔
1594
                p.log.Debugf("Started, waiting on startReady signal")
3✔
1595

3✔
1596
                select {
3✔
1597
                case <-p.startReady:
3✔
1598
                case <-p.cg.Done():
×
1599
                        return
×
1600
                }
1601
        }
1602

1603
        err := fmt.Errorf("disconnecting %s, reason: %v", p, reason)
3✔
1604
        p.storeError(err)
3✔
1605

3✔
1606
        p.log.Infof(err.Error())
3✔
1607

3✔
1608
        // Stop PingManager before closing TCP connection.
3✔
1609
        p.pingManager.Stop()
3✔
1610

3✔
1611
        // Ensure that the TCP connection is properly closed before continuing.
3✔
1612
        p.cfg.Conn.Close()
3✔
1613

3✔
1614
        p.cg.Quit()
3✔
1615

3✔
1616
        // If our msg router isn't global (local to this instance), then we'll
3✔
1617
        // stop it. Otherwise, we'll leave it running.
3✔
1618
        if !p.globalMsgRouter {
6✔
1619
                p.msgRouter.WhenSome(func(router msgmux.Router) {
6✔
1620
                        router.Stop()
3✔
1621
                })
3✔
1622
        }
1623
}
1624

1625
// String returns the string representation of this peer.
1626
func (p *Brontide) String() string {
3✔
1627
        return fmt.Sprintf("%x@%s", p.cfg.PubKeyBytes, p.cfg.Conn.RemoteAddr())
3✔
1628
}
3✔
1629

1630
// readNextMessage reads, and returns the next message on the wire along with
1631
// any additional raw payload.
1632
func (p *Brontide) readNextMessage() (lnwire.Message, error) {
10✔
1633
        noiseConn := p.cfg.Conn
10✔
1634
        err := noiseConn.SetReadDeadline(time.Time{})
10✔
1635
        if err != nil {
10✔
1636
                return nil, err
×
1637
        }
×
1638

1639
        pktLen, err := noiseConn.ReadNextHeader()
10✔
1640
        if err != nil {
13✔
1641
                return nil, fmt.Errorf("read next header: %w", err)
3✔
1642
        }
3✔
1643

1644
        // First we'll read the next _full_ message. We do this rather than
1645
        // reading incrementally from the stream as the Lightning wire protocol
1646
        // is message oriented and allows nodes to pad on additional data to
1647
        // the message stream.
1648
        var (
7✔
1649
                nextMsg lnwire.Message
7✔
1650
                msgLen  uint64
7✔
1651
        )
7✔
1652
        err = p.cfg.ReadPool.Submit(func(buf *buffer.Read) error {
14✔
1653
                // Before reading the body of the message, set the read timeout
7✔
1654
                // accordingly to ensure we don't block other readers using the
7✔
1655
                // pool. We do so only after the task has been scheduled to
7✔
1656
                // ensure the deadline doesn't expire while the message is in
7✔
1657
                // the process of being scheduled.
7✔
1658
                readDeadline := time.Now().Add(
7✔
1659
                        p.scaleTimeout(readMessageTimeout),
7✔
1660
                )
7✔
1661
                readErr := noiseConn.SetReadDeadline(readDeadline)
7✔
1662
                if readErr != nil {
7✔
1663
                        return readErr
×
1664
                }
×
1665

1666
                // The ReadNextBody method will actually end up re-using the
1667
                // buffer, so within this closure, we can continue to use
1668
                // rawMsg as it's just a slice into the buf from the buffer
1669
                // pool.
1670
                rawMsg, readErr := noiseConn.ReadNextBody(buf[:pktLen])
7✔
1671
                if readErr != nil {
7✔
1672
                        return fmt.Errorf("read next body: %w", readErr)
×
1673
                }
×
1674
                msgLen = uint64(len(rawMsg))
7✔
1675

7✔
1676
                // Next, create a new io.Reader implementation from the raw
7✔
1677
                // message, and use this to decode the message directly from.
7✔
1678
                msgReader := bytes.NewReader(rawMsg)
7✔
1679
                nextMsg, err = lnwire.ReadMessage(msgReader, 0)
7✔
1680
                if err != nil {
10✔
1681
                        return err
3✔
1682
                }
3✔
1683

1684
                // At this point, rawMsg and buf will be returned back to the
1685
                // buffer pool for re-use.
1686
                return nil
7✔
1687
        })
1688
        atomic.AddUint64(&p.bytesReceived, msgLen)
7✔
1689
        if err != nil {
10✔
1690
                return nil, err
3✔
1691
        }
3✔
1692

1693
        p.logWireMessage(nextMsg, true)
7✔
1694

7✔
1695
        return nextMsg, nil
7✔
1696
}
1697

1698
// msgStream implements a goroutine-safe, in-order stream of messages to be
1699
// delivered via closure to a receiver. These messages MUST be in order due to
1700
// the nature of the lightning channel commitment and gossiper state machines.
1701
// TODO(conner): use stream handler interface to abstract out stream
1702
// state/logging.
1703
type msgStream struct {
1704
        streamShutdown int32 // To be used atomically.
1705

1706
        peer *Brontide
1707

1708
        apply func(lnwire.Message)
1709

1710
        startMsg string
1711
        stopMsg  string
1712

1713
        msgCond *sync.Cond
1714
        msgs    []lnwire.Message
1715

1716
        mtx sync.Mutex
1717

1718
        producerSema chan struct{}
1719

1720
        wg   sync.WaitGroup
1721
        quit chan struct{}
1722
}
1723

1724
// newMsgStream creates a new instance of a chanMsgStream for a particular
1725
// channel identified by its channel ID. bufSize is the max number of messages
1726
// that should be buffered in the internal queue. Callers should set this to a
1727
// sane value that avoids blocking unnecessarily, but doesn't allow an
1728
// unbounded amount of memory to be allocated to buffer incoming messages.
1729
func newMsgStream(p *Brontide, startMsg, stopMsg string, bufSize uint32,
1730
        apply func(lnwire.Message)) *msgStream {
6✔
1731

6✔
1732
        stream := &msgStream{
6✔
1733
                peer:         p,
6✔
1734
                apply:        apply,
6✔
1735
                startMsg:     startMsg,
6✔
1736
                stopMsg:      stopMsg,
6✔
1737
                producerSema: make(chan struct{}, bufSize),
6✔
1738
                quit:         make(chan struct{}),
6✔
1739
        }
6✔
1740
        stream.msgCond = sync.NewCond(&stream.mtx)
6✔
1741

6✔
1742
        // Before we return the active stream, we'll populate the producer's
6✔
1743
        // semaphore channel. We'll use this to ensure that the producer won't
6✔
1744
        // attempt to allocate memory in the queue for an item until it has
6✔
1745
        // sufficient extra space.
6✔
1746
        for i := uint32(0); i < bufSize; i++ {
24✔
1747
                stream.producerSema <- struct{}{}
18✔
1748
        }
18✔
1749

1750
        return stream
6✔
1751
}
1752

1753
// Start starts the chanMsgStream.
1754
func (ms *msgStream) Start() {
6✔
1755
        ms.wg.Add(1)
6✔
1756
        go ms.msgConsumer()
6✔
1757
}
6✔
1758

1759
// Stop stops the chanMsgStream.
1760
func (ms *msgStream) Stop() {
3✔
1761
        // TODO(roasbeef): signal too?
3✔
1762

3✔
1763
        close(ms.quit)
3✔
1764

3✔
1765
        // Now that we've closed the channel, we'll repeatedly signal the msg
3✔
1766
        // consumer until we've detected that it has exited.
3✔
1767
        for atomic.LoadInt32(&ms.streamShutdown) == 0 {
6✔
1768
                ms.msgCond.Signal()
3✔
1769
                time.Sleep(time.Millisecond * 100)
3✔
1770
        }
3✔
1771

1772
        ms.wg.Wait()
3✔
1773
}
1774

1775
// msgConsumer is the main goroutine that streams messages from the peer's
1776
// readHandler directly to the target channel.
1777
func (ms *msgStream) msgConsumer() {
6✔
1778
        defer ms.wg.Done()
6✔
1779
        defer peerLog.Tracef(ms.stopMsg)
6✔
1780
        defer atomic.StoreInt32(&ms.streamShutdown, 1)
6✔
1781

6✔
1782
        peerLog.Tracef(ms.startMsg)
6✔
1783

6✔
1784
        for {
12✔
1785
                // First, we'll check our condition. If the queue of messages
6✔
1786
                // is empty, then we'll wait until a new item is added.
6✔
1787
                ms.msgCond.L.Lock()
6✔
1788
                for len(ms.msgs) == 0 {
12✔
1789
                        ms.msgCond.Wait()
6✔
1790

6✔
1791
                        // If we woke up in order to exit, then we'll do so.
6✔
1792
                        // Otherwise, we'll check the message queue for any new
6✔
1793
                        // items.
6✔
1794
                        select {
6✔
1795
                        case <-ms.peer.cg.Done():
3✔
1796
                                ms.msgCond.L.Unlock()
3✔
1797
                                return
3✔
1798
                        case <-ms.quit:
3✔
1799
                                ms.msgCond.L.Unlock()
3✔
1800
                                return
3✔
1801
                        default:
3✔
1802
                        }
1803
                }
1804

1805
                // Grab the message off the front of the queue, shifting the
1806
                // slice's reference down one in order to remove the message
1807
                // from the queue.
1808
                msg := ms.msgs[0]
3✔
1809
                ms.msgs[0] = nil // Set to nil to prevent GC leak.
3✔
1810
                ms.msgs = ms.msgs[1:]
3✔
1811

3✔
1812
                ms.msgCond.L.Unlock()
3✔
1813

3✔
1814
                ms.apply(msg)
3✔
1815

3✔
1816
                // We've just successfully processed an item, so we'll signal
3✔
1817
                // to the producer that a new slot in the buffer. We'll use
3✔
1818
                // this to bound the size of the buffer to avoid allowing it to
3✔
1819
                // grow indefinitely.
3✔
1820
                select {
3✔
1821
                case ms.producerSema <- struct{}{}:
3✔
1822
                case <-ms.peer.cg.Done():
3✔
1823
                        return
3✔
1824
                case <-ms.quit:
3✔
1825
                        return
3✔
1826
                }
1827
        }
1828
}
1829

1830
// AddMsg adds a new message to the msgStream. This function is safe for
1831
// concurrent access.
1832
func (ms *msgStream) AddMsg(msg lnwire.Message) {
3✔
1833
        // First, we'll attempt to receive from the producerSema struct. This
3✔
1834
        // acts as a semaphore to prevent us from indefinitely buffering
3✔
1835
        // incoming items from the wire. Either the msg queue isn't full, and
3✔
1836
        // we'll not block, or the queue is full, and we'll block until either
3✔
1837
        // we're signalled to quit, or a slot is freed up.
3✔
1838
        select {
3✔
1839
        case <-ms.producerSema:
3✔
1840
        case <-ms.peer.cg.Done():
×
1841
                return
×
1842
        case <-ms.quit:
×
1843
                return
×
1844
        }
1845

1846
        // Next, we'll lock the condition, and add the message to the end of
1847
        // the message queue.
1848
        ms.msgCond.L.Lock()
3✔
1849
        ms.msgs = append(ms.msgs, msg)
3✔
1850
        ms.msgCond.L.Unlock()
3✔
1851

3✔
1852
        // With the message added, we signal to the msgConsumer that there are
3✔
1853
        // additional messages to consume.
3✔
1854
        ms.msgCond.Signal()
3✔
1855
}
1856

1857
// waitUntilLinkActive waits until the target link is active and returns a
1858
// ChannelLink to pass messages to. It accomplishes this by subscribing to
1859
// an ActiveLinkEvent which is emitted by the link when it first starts up.
1860
func waitUntilLinkActive(p *Brontide,
1861
        cid lnwire.ChannelID) htlcswitch.ChannelUpdateHandler {
3✔
1862

3✔
1863
        p.log.Tracef("Waiting for link=%v to be active", cid)
3✔
1864

3✔
1865
        // Subscribe to receive channel events.
3✔
1866
        //
3✔
1867
        // NOTE: If the link is already active by SubscribeChannelEvents, then
3✔
1868
        // GetLink will retrieve the link and we can send messages. If the link
3✔
1869
        // becomes active between SubscribeChannelEvents and GetLink, then GetLink
3✔
1870
        // will retrieve the link. If the link becomes active after GetLink, then
3✔
1871
        // we will get an ActiveLinkEvent notification and retrieve the link. If
3✔
1872
        // the call to GetLink is before SubscribeChannelEvents, however, there
3✔
1873
        // will be a race condition.
3✔
1874
        sub, err := p.cfg.ChannelNotifier.SubscribeChannelEvents()
3✔
1875
        if err != nil {
6✔
1876
                // If we have a non-nil error, then the server is shutting down and we
3✔
1877
                // can exit here and return nil. This means no message will be delivered
3✔
1878
                // to the link.
3✔
1879
                return nil
3✔
1880
        }
3✔
1881
        defer sub.Cancel()
3✔
1882

3✔
1883
        // The link may already be active by this point, and we may have missed the
3✔
1884
        // ActiveLinkEvent. Check if the link exists.
3✔
1885
        link := p.fetchLinkFromKeyAndCid(cid)
3✔
1886
        if link != nil {
6✔
1887
                return link
3✔
1888
        }
3✔
1889

1890
        // If the link is nil, we must wait for it to be active.
1891
        for {
6✔
1892
                select {
3✔
1893
                // A new event has been sent by the ChannelNotifier. We first check
1894
                // whether the event is an ActiveLinkEvent. If it is, we'll check
1895
                // that the event is for this channel. Otherwise, we discard the
1896
                // message.
1897
                case e := <-sub.Updates():
3✔
1898
                        event, ok := e.(channelnotifier.ActiveLinkEvent)
3✔
1899
                        if !ok {
6✔
1900
                                // Ignore this notification.
3✔
1901
                                continue
3✔
1902
                        }
1903

1904
                        chanPoint := event.ChannelPoint
3✔
1905

3✔
1906
                        // Check whether the retrieved chanPoint matches the target
3✔
1907
                        // channel id.
3✔
1908
                        if !cid.IsChanPoint(chanPoint) {
3✔
1909
                                continue
×
1910
                        }
1911

1912
                        // The link shouldn't be nil as we received an
1913
                        // ActiveLinkEvent. If it is nil, we return nil and the
1914
                        // calling function should catch it.
1915
                        return p.fetchLinkFromKeyAndCid(cid)
3✔
1916

1917
                case <-p.cg.Done():
3✔
1918
                        return nil
3✔
1919
                }
1920
        }
1921
}
1922

1923
// newChanMsgStream is used to create a msgStream between the peer and
1924
// particular channel link in the htlcswitch. We utilize additional
1925
// synchronization with the fundingManager to ensure we don't attempt to
1926
// dispatch a message to a channel before it is fully active. A reference to the
1927
// channel this stream forwards to is held in scope to prevent unnecessary
1928
// lookups.
1929
func newChanMsgStream(p *Brontide, cid lnwire.ChannelID) *msgStream {
3✔
1930
        var chanLink htlcswitch.ChannelUpdateHandler
3✔
1931

3✔
1932
        apply := func(msg lnwire.Message) {
6✔
1933
                // This check is fine because if the link no longer exists, it will
3✔
1934
                // be removed from the activeChannels map and subsequent messages
3✔
1935
                // shouldn't reach the chan msg stream.
3✔
1936
                if chanLink == nil {
6✔
1937
                        chanLink = waitUntilLinkActive(p, cid)
3✔
1938

3✔
1939
                        // If the link is still not active and the calling function
3✔
1940
                        // errored out, just return.
3✔
1941
                        if chanLink == nil {
6✔
1942
                                p.log.Warnf("Link=%v is not active", cid)
3✔
1943
                                return
3✔
1944
                        }
3✔
1945
                }
1946

1947
                // In order to avoid unnecessarily delivering message
1948
                // as the peer is exiting, we'll check quickly to see
1949
                // if we need to exit.
1950
                select {
3✔
1951
                case <-p.cg.Done():
×
1952
                        return
×
1953
                default:
3✔
1954
                }
1955

1956
                chanLink.HandleChannelUpdate(msg)
3✔
1957
        }
1958

1959
        return newMsgStream(p,
3✔
1960
                fmt.Sprintf("Update stream for ChannelID(%x) created", cid[:]),
3✔
1961
                fmt.Sprintf("Update stream for ChannelID(%x) exiting", cid[:]),
3✔
1962
                msgStreamSize,
3✔
1963
                apply,
3✔
1964
        )
3✔
1965
}
1966

1967
// newDiscMsgStream is used to setup a msgStream between the peer and the
1968
// authenticated gossiper. This stream should be used to forward all remote
1969
// channel announcements.
1970
func newDiscMsgStream(p *Brontide) *msgStream {
6✔
1971
        apply := func(msg lnwire.Message) {
9✔
1972
                // TODO(elle): thread contexts through the peer system properly
3✔
1973
                // so that a parent context can be passed in here.
3✔
1974
                ctx := context.TODO()
3✔
1975

3✔
1976
                // TODO(yy): `ProcessRemoteAnnouncement` returns an error chan
3✔
1977
                // and we need to process it.
3✔
1978
                p.cfg.AuthGossiper.ProcessRemoteAnnouncement(ctx, msg, p)
3✔
1979
        }
3✔
1980

1981
        return newMsgStream(
6✔
1982
                p,
6✔
1983
                "Update stream for gossiper created",
6✔
1984
                "Update stream for gossiper exited",
6✔
1985
                msgStreamSize,
6✔
1986
                apply,
6✔
1987
        )
6✔
1988
}
1989

1990
// readHandler is responsible for reading messages off the wire in series, then
1991
// properly dispatching the handling of the message to the proper subsystem.
1992
//
1993
// NOTE: This method MUST be run as a goroutine.
1994
func (p *Brontide) readHandler() {
6✔
1995
        defer p.cg.WgDone()
6✔
1996

6✔
1997
        // We'll stop the timer after a new messages is received, and also
6✔
1998
        // reset it after we process the next message.
6✔
1999
        idleTimer := time.AfterFunc(idleTimeout, func() {
6✔
2000
                err := fmt.Errorf("peer %s no answer for %s -- disconnecting",
×
2001
                        p, idleTimeout)
×
2002
                p.Disconnect(err)
×
2003
        })
×
2004

2005
        // Initialize our negotiated gossip sync method before reading messages
2006
        // off the wire. When using gossip queries, this ensures a gossip
2007
        // syncer is active by the time query messages arrive.
2008
        //
2009
        // TODO(conner): have peer store gossip syncer directly and bypass
2010
        // gossiper?
2011
        p.initGossipSync()
6✔
2012

6✔
2013
        discStream := newDiscMsgStream(p)
6✔
2014
        discStream.Start()
6✔
2015
        defer discStream.Stop()
6✔
2016
out:
6✔
2017
        for atomic.LoadInt32(&p.disconnect) == 0 {
13✔
2018
                nextMsg, err := p.readNextMessage()
7✔
2019
                if !idleTimer.Stop() {
10✔
2020
                        select {
3✔
2021
                        case <-idleTimer.C:
×
2022
                        default:
3✔
2023
                        }
2024
                }
2025
                if err != nil {
7✔
2026
                        p.log.Infof("unable to read message from peer: %v", err)
3✔
2027

3✔
2028
                        // If we could not read our peer's message due to an
3✔
2029
                        // unknown type or invalid alias, we continue processing
3✔
2030
                        // as normal. We store unknown message and address
3✔
2031
                        // types, as they may provide debugging insight.
3✔
2032
                        switch e := err.(type) {
3✔
2033
                        // If this is just a message we don't yet recognize,
2034
                        // we'll continue processing as normal as this allows
2035
                        // us to introduce new messages in a forwards
2036
                        // compatible manner.
2037
                        case *lnwire.UnknownMessage:
3✔
2038
                                p.storeError(e)
3✔
2039
                                idleTimer.Reset(idleTimeout)
3✔
2040
                                continue
3✔
2041

2042
                        // If they sent us an address type that we don't yet
2043
                        // know of, then this isn't a wire error, so we'll
2044
                        // simply continue parsing the remainder of their
2045
                        // messages.
2046
                        case *lnwire.ErrUnknownAddrType:
×
2047
                                p.storeError(e)
×
2048
                                idleTimer.Reset(idleTimeout)
×
2049
                                continue
×
2050

2051
                        // If the NodeAnnouncement has an invalid alias, then
2052
                        // we'll log that error above and continue so we can
2053
                        // continue to read messages from the peer. We do not
2054
                        // store this error because it is of little debugging
2055
                        // value.
2056
                        case *lnwire.ErrInvalidNodeAlias:
×
2057
                                idleTimer.Reset(idleTimeout)
×
2058
                                continue
×
2059

2060
                        // If the error we encountered wasn't just a message we
2061
                        // didn't recognize, then we'll stop all processing as
2062
                        // this is a fatal error.
2063
                        default:
3✔
2064
                                break out
3✔
2065
                        }
2066
                }
2067

2068
                // If a message router is active, then we'll try to have it
2069
                // handle this message. If it can, then we're able to skip the
2070
                // rest of the message handling logic.
2071
                err = fn.MapOptionZ(p.msgRouter, func(r msgmux.Router) error {
8✔
2072
                        return r.RouteMsg(msgmux.PeerMsg{
4✔
2073
                                PeerPub: *p.IdentityKey(),
4✔
2074
                                Message: nextMsg,
4✔
2075
                        })
4✔
2076
                })
4✔
2077

2078
                // No error occurred, and the message was handled by the
2079
                // router.
2080
                if err == nil {
7✔
2081
                        continue
3✔
2082
                }
2083

2084
                var (
4✔
2085
                        targetChan   lnwire.ChannelID
4✔
2086
                        isLinkUpdate bool
4✔
2087
                )
4✔
2088

4✔
2089
                switch msg := nextMsg.(type) {
4✔
2090
                case *lnwire.Pong:
×
2091
                        // When we receive a Pong message in response to our
×
2092
                        // last ping message, we send it to the pingManager
×
2093
                        p.pingManager.ReceivedPong(msg)
×
2094

2095
                case *lnwire.Ping:
×
2096
                        // First, we'll store their latest ping payload within
×
2097
                        // the relevant atomic variable.
×
2098
                        p.lastPingPayload.Store(msg.PaddingBytes[:])
×
2099

×
2100
                        // Next, we'll send over the amount of specified pong
×
2101
                        // bytes.
×
2102
                        pong := lnwire.NewPong(p.cfg.PongBuf[0:msg.NumPongBytes])
×
2103
                        p.queueMsg(pong, nil)
×
2104

2105
                case *lnwire.OpenChannel,
2106
                        *lnwire.AcceptChannel,
2107
                        *lnwire.FundingCreated,
2108
                        *lnwire.FundingSigned,
2109
                        *lnwire.ChannelReady:
3✔
2110

3✔
2111
                        p.cfg.FundingManager.ProcessFundingMsg(msg, p)
3✔
2112

2113
                case *lnwire.Shutdown:
3✔
2114
                        select {
3✔
2115
                        case p.chanCloseMsgs <- &closeMsg{msg.ChannelID, msg}:
3✔
2116
                        case <-p.cg.Done():
×
2117
                                break out
×
2118
                        }
2119
                case *lnwire.ClosingSigned:
3✔
2120
                        select {
3✔
2121
                        case p.chanCloseMsgs <- &closeMsg{msg.ChannelID, msg}:
3✔
2122
                        case <-p.cg.Done():
×
2123
                                break out
×
2124
                        }
2125

2126
                case *lnwire.Warning:
×
2127
                        targetChan = msg.ChanID
×
2128
                        isLinkUpdate = p.handleWarningOrError(targetChan, msg)
×
2129

2130
                case *lnwire.Error:
3✔
2131
                        targetChan = msg.ChanID
3✔
2132
                        isLinkUpdate = p.handleWarningOrError(targetChan, msg)
3✔
2133

2134
                case *lnwire.ChannelReestablish:
3✔
2135
                        targetChan = msg.ChanID
3✔
2136
                        isLinkUpdate = p.hasChannel(targetChan)
3✔
2137

3✔
2138
                        // If we failed to find the link in question, and the
3✔
2139
                        // message received was a channel sync message, then
3✔
2140
                        // this might be a peer trying to resync closed channel.
3✔
2141
                        // In this case we'll try to resend our last channel
3✔
2142
                        // sync message, such that the peer can recover funds
3✔
2143
                        // from the closed channel.
3✔
2144
                        if !isLinkUpdate {
6✔
2145
                                err := p.resendChanSyncMsg(targetChan)
3✔
2146
                                if err != nil {
6✔
2147
                                        // TODO(halseth): send error to peer?
3✔
2148
                                        p.log.Errorf("resend failed: %v",
3✔
2149
                                                err)
3✔
2150
                                }
3✔
2151
                        }
2152

2153
                // For messages that implement the LinkUpdater interface, we
2154
                // will consider them as link updates and send them to
2155
                // chanStream. These messages will be queued inside chanStream
2156
                // if the channel is not active yet.
2157
                case lnwire.LinkUpdater:
3✔
2158
                        targetChan = msg.TargetChanID()
3✔
2159
                        isLinkUpdate = p.hasChannel(targetChan)
3✔
2160

3✔
2161
                        // Log an error if we don't have this channel. This
3✔
2162
                        // means the peer has sent us a message with unknown
3✔
2163
                        // channel ID.
3✔
2164
                        if !isLinkUpdate {
6✔
2165
                                p.log.Errorf("Unknown channel ID: %v found "+
3✔
2166
                                        "in received msg=%s", targetChan,
3✔
2167
                                        nextMsg.MsgType())
3✔
2168
                        }
3✔
2169

2170
                case *lnwire.ChannelUpdate1,
2171
                        *lnwire.ChannelAnnouncement1,
2172
                        *lnwire.NodeAnnouncement,
2173
                        *lnwire.AnnounceSignatures1,
2174
                        *lnwire.GossipTimestampRange,
2175
                        *lnwire.QueryShortChanIDs,
2176
                        *lnwire.QueryChannelRange,
2177
                        *lnwire.ReplyChannelRange,
2178
                        *lnwire.ReplyShortChanIDsEnd:
3✔
2179

3✔
2180
                        discStream.AddMsg(msg)
3✔
2181

2182
                case *lnwire.Custom:
4✔
2183
                        err := p.handleCustomMessage(msg)
4✔
2184
                        if err != nil {
4✔
2185
                                p.storeError(err)
×
2186
                                p.log.Errorf("%v", err)
×
2187
                        }
×
2188

2189
                default:
×
2190
                        // If the message we received is unknown to us, store
×
2191
                        // the type to track the failure.
×
2192
                        err := fmt.Errorf("unknown message type %v received",
×
2193
                                uint16(msg.MsgType()))
×
2194
                        p.storeError(err)
×
2195

×
2196
                        p.log.Errorf("%v", err)
×
2197
                }
2198

2199
                if isLinkUpdate {
7✔
2200
                        // If this is a channel update, then we need to feed it
3✔
2201
                        // into the channel's in-order message stream.
3✔
2202
                        p.sendLinkUpdateMsg(targetChan, nextMsg)
3✔
2203
                }
3✔
2204

2205
                idleTimer.Reset(idleTimeout)
4✔
2206
        }
2207

2208
        p.Disconnect(errors.New("read handler closed"))
3✔
2209

3✔
2210
        p.log.Trace("readHandler for peer done")
3✔
2211
}
2212

2213
// handleCustomMessage handles the given custom message if a handler is
2214
// registered.
2215
func (p *Brontide) handleCustomMessage(msg *lnwire.Custom) error {
4✔
2216
        if p.cfg.HandleCustomMessage == nil {
4✔
2217
                return fmt.Errorf("no custom message handler for "+
×
2218
                        "message type %v", uint16(msg.MsgType()))
×
2219
        }
×
2220

2221
        return p.cfg.HandleCustomMessage(p.PubKey(), msg)
4✔
2222
}
2223

2224
// isLoadedFromDisk returns true if the provided channel ID is loaded from
2225
// disk.
2226
//
2227
// NOTE: only returns true for pending channels.
2228
func (p *Brontide) isLoadedFromDisk(chanID lnwire.ChannelID) bool {
3✔
2229
        // If this is a newly added channel, no need to reestablish.
3✔
2230
        _, added := p.addedChannels.Load(chanID)
3✔
2231
        if added {
6✔
2232
                return false
3✔
2233
        }
3✔
2234

2235
        // Return false if the channel is unknown.
2236
        channel, ok := p.activeChannels.Load(chanID)
3✔
2237
        if !ok {
3✔
2238
                return false
×
2239
        }
×
2240

2241
        // During startup, we will use a nil value to mark a pending channel
2242
        // that's loaded from disk.
2243
        return channel == nil
3✔
2244
}
2245

2246
// isActiveChannel returns true if the provided channel id is active, otherwise
2247
// returns false.
2248
func (p *Brontide) isActiveChannel(chanID lnwire.ChannelID) bool {
11✔
2249
        // The channel would be nil if,
11✔
2250
        // - the channel doesn't exist, or,
11✔
2251
        // - the channel exists, but is pending. In this case, we don't
11✔
2252
        //   consider this channel active.
11✔
2253
        channel, _ := p.activeChannels.Load(chanID)
11✔
2254

11✔
2255
        return channel != nil
11✔
2256
}
11✔
2257

2258
// isPendingChannel returns true if the provided channel ID is pending, and
2259
// returns false if the channel is active or unknown.
2260
func (p *Brontide) isPendingChannel(chanID lnwire.ChannelID) bool {
9✔
2261
        // Return false if the channel is unknown.
9✔
2262
        channel, ok := p.activeChannels.Load(chanID)
9✔
2263
        if !ok {
15✔
2264
                return false
6✔
2265
        }
6✔
2266

2267
        return channel == nil
6✔
2268
}
2269

2270
// hasChannel returns true if the peer has a pending/active channel specified
2271
// by the channel ID.
2272
func (p *Brontide) hasChannel(chanID lnwire.ChannelID) bool {
3✔
2273
        _, ok := p.activeChannels.Load(chanID)
3✔
2274
        return ok
3✔
2275
}
3✔
2276

2277
// storeError stores an error in our peer's buffer of recent errors with the
2278
// current timestamp. Errors are only stored if we have at least one active
2279
// channel with the peer to mitigate a dos vector where a peer costlessly
2280
// connects to us and spams us with errors.
2281
func (p *Brontide) storeError(err error) {
3✔
2282
        var haveChannels bool
3✔
2283

3✔
2284
        p.activeChannels.Range(func(_ lnwire.ChannelID,
3✔
2285
                channel *lnwallet.LightningChannel) bool {
6✔
2286

3✔
2287
                // Pending channels will be nil in the activeChannels map.
3✔
2288
                if channel == nil {
6✔
2289
                        // Return true to continue the iteration.
3✔
2290
                        return true
3✔
2291
                }
3✔
2292

2293
                haveChannels = true
3✔
2294

3✔
2295
                // Return false to break the iteration.
3✔
2296
                return false
3✔
2297
        })
2298

2299
        // If we do not have any active channels with the peer, we do not store
2300
        // errors as a dos mitigation.
2301
        if !haveChannels {
6✔
2302
                p.log.Trace("no channels with peer, not storing err")
3✔
2303
                return
3✔
2304
        }
3✔
2305

2306
        p.cfg.ErrorBuffer.Add(
3✔
2307
                &TimestampedError{Timestamp: time.Now(), Error: err},
3✔
2308
        )
3✔
2309
}
2310

2311
// handleWarningOrError processes a warning or error msg and returns true if
2312
// msg should be forwarded to the associated channel link. False is returned if
2313
// any necessary forwarding of msg was already handled by this method. If msg is
2314
// an error from a peer with an active channel, we'll store it in memory.
2315
//
2316
// NOTE: This method should only be called from within the readHandler.
2317
func (p *Brontide) handleWarningOrError(chanID lnwire.ChannelID,
2318
        msg lnwire.Message) bool {
3✔
2319

3✔
2320
        if errMsg, ok := msg.(*lnwire.Error); ok {
6✔
2321
                p.storeError(errMsg)
3✔
2322
        }
3✔
2323

2324
        switch {
3✔
2325
        // Connection wide messages should be forwarded to all channel links
2326
        // with this peer.
2327
        case chanID == lnwire.ConnectionWideID:
×
2328
                for _, chanStream := range p.activeMsgStreams {
×
2329
                        chanStream.AddMsg(msg)
×
2330
                }
×
2331

2332
                return false
×
2333

2334
        // If the channel ID for the message corresponds to a pending channel,
2335
        // then the funding manager will handle it.
2336
        case p.cfg.FundingManager.IsPendingChannel(chanID, p):
3✔
2337
                p.cfg.FundingManager.ProcessFundingMsg(msg, p)
3✔
2338
                return false
3✔
2339

2340
        // If not we hand the message to the channel link for this channel.
2341
        case p.isActiveChannel(chanID):
3✔
2342
                return true
3✔
2343

2344
        default:
3✔
2345
                return false
3✔
2346
        }
2347
}
2348

2349
// messageSummary returns a human-readable string that summarizes a
2350
// incoming/outgoing message. Not all messages will have a summary, only those
2351
// which have additional data that can be informative at a glance.
2352
func messageSummary(msg lnwire.Message) string {
3✔
2353
        switch msg := msg.(type) {
3✔
2354
        case *lnwire.Init:
3✔
2355
                // No summary.
3✔
2356
                return ""
3✔
2357

2358
        case *lnwire.OpenChannel:
3✔
2359
                return fmt.Sprintf("temp_chan_id=%x, chain=%v, csv=%v, amt=%v, "+
3✔
2360
                        "push_amt=%v, reserve=%v, flags=%v",
3✔
2361
                        msg.PendingChannelID[:], msg.ChainHash,
3✔
2362
                        msg.CsvDelay, msg.FundingAmount, msg.PushAmount,
3✔
2363
                        msg.ChannelReserve, msg.ChannelFlags)
3✔
2364

2365
        case *lnwire.AcceptChannel:
3✔
2366
                return fmt.Sprintf("temp_chan_id=%x, reserve=%v, csv=%v, num_confs=%v",
3✔
2367
                        msg.PendingChannelID[:], msg.ChannelReserve, msg.CsvDelay,
3✔
2368
                        msg.MinAcceptDepth)
3✔
2369

2370
        case *lnwire.FundingCreated:
3✔
2371
                return fmt.Sprintf("temp_chan_id=%x, chan_point=%v",
3✔
2372
                        msg.PendingChannelID[:], msg.FundingPoint)
3✔
2373

2374
        case *lnwire.FundingSigned:
3✔
2375
                return fmt.Sprintf("chan_id=%v", msg.ChanID)
3✔
2376

2377
        case *lnwire.ChannelReady:
3✔
2378
                return fmt.Sprintf("chan_id=%v, next_point=%x",
3✔
2379
                        msg.ChanID, msg.NextPerCommitmentPoint.SerializeCompressed())
3✔
2380

2381
        case *lnwire.Shutdown:
3✔
2382
                return fmt.Sprintf("chan_id=%v, script=%x", msg.ChannelID,
3✔
2383
                        msg.Address[:])
3✔
2384

2385
        case *lnwire.ClosingComplete:
3✔
2386
                return fmt.Sprintf("chan_id=%v, fee_sat=%v, locktime=%v",
3✔
2387
                        msg.ChannelID, msg.FeeSatoshis, msg.LockTime)
3✔
2388

2389
        case *lnwire.ClosingSig:
3✔
2390
                return fmt.Sprintf("chan_id=%v", msg.ChannelID)
3✔
2391

2392
        case *lnwire.ClosingSigned:
3✔
2393
                return fmt.Sprintf("chan_id=%v, fee_sat=%v", msg.ChannelID,
3✔
2394
                        msg.FeeSatoshis)
3✔
2395

2396
        case *lnwire.UpdateAddHTLC:
3✔
2397
                var blindingPoint []byte
3✔
2398
                msg.BlindingPoint.WhenSome(
3✔
2399
                        func(b tlv.RecordT[lnwire.BlindingPointTlvType,
3✔
2400
                                *btcec.PublicKey]) {
6✔
2401

3✔
2402
                                blindingPoint = b.Val.SerializeCompressed()
3✔
2403
                        },
3✔
2404
                )
2405

2406
                return fmt.Sprintf("chan_id=%v, id=%v, amt=%v, expiry=%v, "+
3✔
2407
                        "hash=%x, blinding_point=%x, custom_records=%v",
3✔
2408
                        msg.ChanID, msg.ID, msg.Amount, msg.Expiry,
3✔
2409
                        msg.PaymentHash[:], blindingPoint, msg.CustomRecords)
3✔
2410

2411
        case *lnwire.UpdateFailHTLC:
3✔
2412
                return fmt.Sprintf("chan_id=%v, id=%v, reason=%x", msg.ChanID,
3✔
2413
                        msg.ID, msg.Reason)
3✔
2414

2415
        case *lnwire.UpdateFulfillHTLC:
3✔
2416
                return fmt.Sprintf("chan_id=%v, id=%v, preimage=%x, "+
3✔
2417
                        "custom_records=%v", msg.ChanID, msg.ID,
3✔
2418
                        msg.PaymentPreimage[:], msg.CustomRecords)
3✔
2419

2420
        case *lnwire.CommitSig:
3✔
2421
                return fmt.Sprintf("chan_id=%v, num_htlcs=%v", msg.ChanID,
3✔
2422
                        len(msg.HtlcSigs))
3✔
2423

2424
        case *lnwire.RevokeAndAck:
3✔
2425
                return fmt.Sprintf("chan_id=%v, rev=%x, next_point=%x",
3✔
2426
                        msg.ChanID, msg.Revocation[:],
3✔
2427
                        msg.NextRevocationKey.SerializeCompressed())
3✔
2428

2429
        case *lnwire.UpdateFailMalformedHTLC:
3✔
2430
                return fmt.Sprintf("chan_id=%v, id=%v, fail_code=%v",
3✔
2431
                        msg.ChanID, msg.ID, msg.FailureCode)
3✔
2432

2433
        case *lnwire.Warning:
×
2434
                return fmt.Sprintf("%v", msg.Warning())
×
2435

2436
        case *lnwire.Error:
3✔
2437
                return fmt.Sprintf("%v", msg.Error())
3✔
2438

2439
        case *lnwire.AnnounceSignatures1:
3✔
2440
                return fmt.Sprintf("chan_id=%v, short_chan_id=%v", msg.ChannelID,
3✔
2441
                        msg.ShortChannelID.ToUint64())
3✔
2442

2443
        case *lnwire.ChannelAnnouncement1:
3✔
2444
                return fmt.Sprintf("chain_hash=%v, short_chan_id=%v",
3✔
2445
                        msg.ChainHash, msg.ShortChannelID.ToUint64())
3✔
2446

2447
        case *lnwire.ChannelUpdate1:
3✔
2448
                return fmt.Sprintf("chain_hash=%v, short_chan_id=%v, "+
3✔
2449
                        "mflags=%v, cflags=%v, update_time=%v", msg.ChainHash,
3✔
2450
                        msg.ShortChannelID.ToUint64(), msg.MessageFlags,
3✔
2451
                        msg.ChannelFlags, time.Unix(int64(msg.Timestamp), 0))
3✔
2452

2453
        case *lnwire.NodeAnnouncement:
3✔
2454
                return fmt.Sprintf("node=%x, update_time=%v",
3✔
2455
                        msg.NodeID, time.Unix(int64(msg.Timestamp), 0))
3✔
2456

2457
        case *lnwire.Ping:
×
2458
                return fmt.Sprintf("ping_bytes=%x", msg.PaddingBytes[:])
×
2459

2460
        case *lnwire.Pong:
×
2461
                return fmt.Sprintf("len(pong_bytes)=%d", len(msg.PongBytes[:]))
×
2462

2463
        case *lnwire.UpdateFee:
×
2464
                return fmt.Sprintf("chan_id=%v, fee_update_sat=%v",
×
2465
                        msg.ChanID, int64(msg.FeePerKw))
×
2466

2467
        case *lnwire.ChannelReestablish:
3✔
2468
                return fmt.Sprintf("chan_id=%v, next_local_height=%v, "+
3✔
2469
                        "remote_tail_height=%v", msg.ChanID,
3✔
2470
                        msg.NextLocalCommitHeight, msg.RemoteCommitTailHeight)
3✔
2471

2472
        case *lnwire.ReplyShortChanIDsEnd:
3✔
2473
                return fmt.Sprintf("chain_hash=%v, complete=%v", msg.ChainHash,
3✔
2474
                        msg.Complete)
3✔
2475

2476
        case *lnwire.ReplyChannelRange:
3✔
2477
                return fmt.Sprintf("start_height=%v, end_height=%v, "+
3✔
2478
                        "num_chans=%v, encoding=%v", msg.FirstBlockHeight,
3✔
2479
                        msg.LastBlockHeight(), len(msg.ShortChanIDs),
3✔
2480
                        msg.EncodingType)
3✔
2481

2482
        case *lnwire.QueryShortChanIDs:
3✔
2483
                return fmt.Sprintf("chain_hash=%v, encoding=%v, num_chans=%v",
3✔
2484
                        msg.ChainHash, msg.EncodingType, len(msg.ShortChanIDs))
3✔
2485

2486
        case *lnwire.QueryChannelRange:
3✔
2487
                return fmt.Sprintf("chain_hash=%v, start_height=%v, "+
3✔
2488
                        "end_height=%v", msg.ChainHash, msg.FirstBlockHeight,
3✔
2489
                        msg.LastBlockHeight())
3✔
2490

2491
        case *lnwire.GossipTimestampRange:
3✔
2492
                return fmt.Sprintf("chain_hash=%v, first_stamp=%v, "+
3✔
2493
                        "stamp_range=%v", msg.ChainHash,
3✔
2494
                        time.Unix(int64(msg.FirstTimestamp), 0),
3✔
2495
                        msg.TimestampRange)
3✔
2496

2497
        case *lnwire.Stfu:
3✔
2498
                return fmt.Sprintf("chan_id=%v, initiator=%v", msg.ChanID,
3✔
2499
                        msg.Initiator)
3✔
2500

2501
        case *lnwire.Custom:
3✔
2502
                return fmt.Sprintf("type=%d", msg.Type)
3✔
2503
        }
2504

2505
        return fmt.Sprintf("unknown msg type=%T", msg)
×
2506
}
2507

2508
// logWireMessage logs the receipt or sending of particular wire message. This
2509
// function is used rather than just logging the message in order to produce
2510
// less spammy log messages in trace mode by setting the 'Curve" parameter to
2511
// nil. Doing this avoids printing out each of the field elements in the curve
2512
// parameters for secp256k1.
2513
func (p *Brontide) logWireMessage(msg lnwire.Message, read bool) {
20✔
2514
        summaryPrefix := "Received"
20✔
2515
        if !read {
36✔
2516
                summaryPrefix = "Sending"
16✔
2517
        }
16✔
2518

2519
        p.log.Debugf("%v", lnutils.NewLogClosure(func() string {
23✔
2520
                // Debug summary of message.
3✔
2521
                summary := messageSummary(msg)
3✔
2522
                if len(summary) > 0 {
6✔
2523
                        summary = "(" + summary + ")"
3✔
2524
                }
3✔
2525

2526
                preposition := "to"
3✔
2527
                if read {
6✔
2528
                        preposition = "from"
3✔
2529
                }
3✔
2530

2531
                var msgType string
3✔
2532
                if msg.MsgType() < lnwire.CustomTypeStart {
6✔
2533
                        msgType = msg.MsgType().String()
3✔
2534
                } else {
6✔
2535
                        msgType = "custom"
3✔
2536
                }
3✔
2537

2538
                return fmt.Sprintf("%v %v%s %v %s", summaryPrefix,
3✔
2539
                        msgType, summary, preposition, p)
3✔
2540
        }))
2541

2542
        prefix := "readMessage from peer"
20✔
2543
        if !read {
36✔
2544
                prefix = "writeMessage to peer"
16✔
2545
        }
16✔
2546

2547
        p.log.Tracef(prefix+": %v", lnutils.SpewLogClosure(msg))
20✔
2548
}
2549

2550
// writeMessage writes and flushes the target lnwire.Message to the remote peer.
2551
// If the passed message is nil, this method will only try to flush an existing
2552
// message buffered on the connection. It is safe to call this method again
2553
// with a nil message iff a timeout error is returned. This will continue to
2554
// flush the pending message to the wire.
2555
//
2556
// NOTE:
2557
// Besides its usage in Start, this function should not be used elsewhere
2558
// except in writeHandler. If multiple goroutines call writeMessage at the same
2559
// time, panics can occur because WriteMessage and Flush don't use any locking
2560
// internally.
2561
func (p *Brontide) writeMessage(msg lnwire.Message) error {
16✔
2562
        // Only log the message on the first attempt.
16✔
2563
        if msg != nil {
32✔
2564
                p.logWireMessage(msg, false)
16✔
2565
        }
16✔
2566

2567
        noiseConn := p.cfg.Conn
16✔
2568

16✔
2569
        flushMsg := func() error {
32✔
2570
                // Ensure the write deadline is set before we attempt to send
16✔
2571
                // the message.
16✔
2572
                writeDeadline := time.Now().Add(
16✔
2573
                        p.scaleTimeout(writeMessageTimeout),
16✔
2574
                )
16✔
2575
                err := noiseConn.SetWriteDeadline(writeDeadline)
16✔
2576
                if err != nil {
16✔
2577
                        return err
×
2578
                }
×
2579

2580
                // Flush the pending message to the wire. If an error is
2581
                // encountered, e.g. write timeout, the number of bytes written
2582
                // so far will be returned.
2583
                n, err := noiseConn.Flush()
16✔
2584

16✔
2585
                // Record the number of bytes written on the wire, if any.
16✔
2586
                if n > 0 {
19✔
2587
                        atomic.AddUint64(&p.bytesSent, uint64(n))
3✔
2588
                }
3✔
2589

2590
                return err
16✔
2591
        }
2592

2593
        // If the current message has already been serialized, encrypted, and
2594
        // buffered on the underlying connection we will skip straight to
2595
        // flushing it to the wire.
2596
        if msg == nil {
16✔
2597
                return flushMsg()
×
2598
        }
×
2599

2600
        // Otherwise, this is a new message. We'll acquire a write buffer to
2601
        // serialize the message and buffer the ciphertext on the connection.
2602
        err := p.cfg.WritePool.Submit(func(buf *bytes.Buffer) error {
32✔
2603
                // Using a buffer allocated by the write pool, encode the
16✔
2604
                // message directly into the buffer.
16✔
2605
                _, writeErr := lnwire.WriteMessage(buf, msg, 0)
16✔
2606
                if writeErr != nil {
16✔
2607
                        return writeErr
×
2608
                }
×
2609

2610
                // Finally, write the message itself in a single swoop. This
2611
                // will buffer the ciphertext on the underlying connection. We
2612
                // will defer flushing the message until the write pool has been
2613
                // released.
2614
                return noiseConn.WriteMessage(buf.Bytes())
16✔
2615
        })
2616
        if err != nil {
16✔
2617
                return err
×
2618
        }
×
2619

2620
        return flushMsg()
16✔
2621
}
2622

2623
// writeHandler is a goroutine dedicated to reading messages off of an incoming
2624
// queue, and writing them out to the wire. This goroutine coordinates with the
2625
// queueHandler in order to ensure the incoming message queue is quickly
2626
// drained.
2627
//
2628
// NOTE: This method MUST be run as a goroutine.
2629
func (p *Brontide) writeHandler() {
6✔
2630
        // We'll stop the timer after a new messages is sent, and also reset it
6✔
2631
        // after we process the next message.
6✔
2632
        idleTimer := time.AfterFunc(idleTimeout, func() {
6✔
2633
                err := fmt.Errorf("peer %s no write for %s -- disconnecting",
×
2634
                        p, idleTimeout)
×
2635
                p.Disconnect(err)
×
2636
        })
×
2637

2638
        var exitErr error
6✔
2639

6✔
2640
out:
6✔
2641
        for {
16✔
2642
                select {
10✔
2643
                case outMsg := <-p.sendQueue:
7✔
2644
                        // Record the time at which we first attempt to send the
7✔
2645
                        // message.
7✔
2646
                        startTime := time.Now()
7✔
2647

7✔
2648
                retry:
7✔
2649
                        // Write out the message to the socket. If a timeout
2650
                        // error is encountered, we will catch this and retry
2651
                        // after backing off in case the remote peer is just
2652
                        // slow to process messages from the wire.
2653
                        err := p.writeMessage(outMsg.msg)
7✔
2654
                        if nerr, ok := err.(net.Error); ok && nerr.Timeout() {
7✔
2655
                                p.log.Debugf("Write timeout detected for "+
×
2656
                                        "peer, first write for message "+
×
2657
                                        "attempted %v ago",
×
2658
                                        time.Since(startTime))
×
2659

×
2660
                                // If we received a timeout error, this implies
×
2661
                                // that the message was buffered on the
×
2662
                                // connection successfully and that a flush was
×
2663
                                // attempted. We'll set the message to nil so
×
2664
                                // that on a subsequent pass we only try to
×
2665
                                // flush the buffered message, and forgo
×
2666
                                // reserializing or reencrypting it.
×
2667
                                outMsg.msg = nil
×
2668

×
2669
                                goto retry
×
2670
                        }
2671

2672
                        // The write succeeded, reset the idle timer to prevent
2673
                        // us from disconnecting the peer.
2674
                        if !idleTimer.Stop() {
7✔
2675
                                select {
×
2676
                                case <-idleTimer.C:
×
2677
                                default:
×
2678
                                }
2679
                        }
2680
                        idleTimer.Reset(idleTimeout)
7✔
2681

7✔
2682
                        // If the peer requested a synchronous write, respond
7✔
2683
                        // with the error.
7✔
2684
                        if outMsg.errChan != nil {
11✔
2685
                                outMsg.errChan <- err
4✔
2686
                        }
4✔
2687

2688
                        if err != nil {
7✔
2689
                                exitErr = fmt.Errorf("unable to write "+
×
2690
                                        "message: %v", err)
×
2691
                                break out
×
2692
                        }
2693

2694
                case <-p.cg.Done():
3✔
2695
                        exitErr = lnpeer.ErrPeerExiting
3✔
2696
                        break out
3✔
2697
                }
2698
        }
2699

2700
        // Avoid an exit deadlock by ensuring WaitGroups are decremented before
2701
        // disconnect.
2702
        p.cg.WgDone()
3✔
2703

3✔
2704
        p.Disconnect(exitErr)
3✔
2705

3✔
2706
        p.log.Trace("writeHandler for peer done")
3✔
2707
}
2708

2709
// queueHandler is responsible for accepting messages from outside subsystems
2710
// to be eventually sent out on the wire by the writeHandler.
2711
//
2712
// NOTE: This method MUST be run as a goroutine.
2713
func (p *Brontide) queueHandler() {
6✔
2714
        defer p.cg.WgDone()
6✔
2715

6✔
2716
        // priorityMsgs holds an in order list of messages deemed high-priority
6✔
2717
        // to be added to the sendQueue. This predominately includes messages
6✔
2718
        // from the funding manager and htlcswitch.
6✔
2719
        priorityMsgs := list.New()
6✔
2720

6✔
2721
        // lazyMsgs holds an in order list of messages deemed low-priority to be
6✔
2722
        // added to the sendQueue only after all high-priority messages have
6✔
2723
        // been queued. This predominately includes messages from the gossiper.
6✔
2724
        lazyMsgs := list.New()
6✔
2725

6✔
2726
        for {
20✔
2727
                // Examine the front of the priority queue, if it is empty check
14✔
2728
                // the low priority queue.
14✔
2729
                elem := priorityMsgs.Front()
14✔
2730
                if elem == nil {
25✔
2731
                        elem = lazyMsgs.Front()
11✔
2732
                }
11✔
2733

2734
                if elem != nil {
21✔
2735
                        front := elem.Value.(outgoingMsg)
7✔
2736

7✔
2737
                        // There's an element on the queue, try adding
7✔
2738
                        // it to the sendQueue. We also watch for
7✔
2739
                        // messages on the outgoingQueue, in case the
7✔
2740
                        // writeHandler cannot accept messages on the
7✔
2741
                        // sendQueue.
7✔
2742
                        select {
7✔
2743
                        case p.sendQueue <- front:
7✔
2744
                                if front.priority {
13✔
2745
                                        priorityMsgs.Remove(elem)
6✔
2746
                                } else {
10✔
2747
                                        lazyMsgs.Remove(elem)
4✔
2748
                                }
4✔
2749
                        case msg := <-p.outgoingQueue:
3✔
2750
                                if msg.priority {
6✔
2751
                                        priorityMsgs.PushBack(msg)
3✔
2752
                                } else {
6✔
2753
                                        lazyMsgs.PushBack(msg)
3✔
2754
                                }
3✔
2755
                        case <-p.cg.Done():
×
2756
                                return
×
2757
                        }
2758
                } else {
10✔
2759
                        // If there weren't any messages to send to the
10✔
2760
                        // writeHandler, then we'll accept a new message
10✔
2761
                        // into the queue from outside sub-systems.
10✔
2762
                        select {
10✔
2763
                        case msg := <-p.outgoingQueue:
7✔
2764
                                if msg.priority {
13✔
2765
                                        priorityMsgs.PushBack(msg)
6✔
2766
                                } else {
10✔
2767
                                        lazyMsgs.PushBack(msg)
4✔
2768
                                }
4✔
2769
                        case <-p.cg.Done():
3✔
2770
                                return
3✔
2771
                        }
2772
                }
2773
        }
2774
}
2775

2776
// PingTime returns the estimated ping time to the peer in microseconds.
2777
func (p *Brontide) PingTime() int64 {
3✔
2778
        return p.pingManager.GetPingTimeMicroSeconds()
3✔
2779
}
3✔
2780

2781
// queueMsg adds the lnwire.Message to the back of the high priority send queue.
2782
// If the errChan is non-nil, an error is sent back if the msg failed to queue
2783
// or failed to write, and nil otherwise.
2784
func (p *Brontide) queueMsg(msg lnwire.Message, errChan chan error) {
28✔
2785
        p.queue(true, msg, errChan)
28✔
2786
}
28✔
2787

2788
// queueMsgLazy adds the lnwire.Message to the back of the low priority send
2789
// queue. If the errChan is non-nil, an error is sent back if the msg failed to
2790
// queue or failed to write, and nil otherwise.
2791
func (p *Brontide) queueMsgLazy(msg lnwire.Message, errChan chan error) {
4✔
2792
        p.queue(false, msg, errChan)
4✔
2793
}
4✔
2794

2795
// queue sends a given message to the queueHandler using the passed priority. If
2796
// the errChan is non-nil, an error is sent back if the msg failed to queue or
2797
// failed to write, and nil otherwise.
2798
func (p *Brontide) queue(priority bool, msg lnwire.Message,
2799
        errChan chan error) {
29✔
2800

29✔
2801
        select {
29✔
2802
        case p.outgoingQueue <- outgoingMsg{priority, msg, errChan}:
28✔
2803
        case <-p.cg.Done():
×
2804
                p.log.Tracef("Peer shutting down, could not enqueue msg: %v.",
×
2805
                        spew.Sdump(msg))
×
2806
                if errChan != nil {
×
2807
                        errChan <- lnpeer.ErrPeerExiting
×
2808
                }
×
2809
        }
2810
}
2811

2812
// ChannelSnapshots returns a slice of channel snapshots detailing all
2813
// currently active channels maintained with the remote peer.
2814
func (p *Brontide) ChannelSnapshots() []*channeldb.ChannelSnapshot {
3✔
2815
        snapshots := make(
3✔
2816
                []*channeldb.ChannelSnapshot, 0, p.activeChannels.Len(),
3✔
2817
        )
3✔
2818

3✔
2819
        p.activeChannels.ForEach(func(_ lnwire.ChannelID,
3✔
2820
                activeChan *lnwallet.LightningChannel) error {
6✔
2821

3✔
2822
                // If the activeChan is nil, then we skip it as the channel is
3✔
2823
                // pending.
3✔
2824
                if activeChan == nil {
6✔
2825
                        return nil
3✔
2826
                }
3✔
2827

2828
                // We'll only return a snapshot for channels that are
2829
                // *immediately* available for routing payments over.
2830
                if activeChan.RemoteNextRevocation() == nil {
6✔
2831
                        return nil
3✔
2832
                }
3✔
2833

2834
                snapshot := activeChan.StateSnapshot()
3✔
2835
                snapshots = append(snapshots, snapshot)
3✔
2836

3✔
2837
                return nil
3✔
2838
        })
2839

2840
        return snapshots
3✔
2841
}
2842

2843
// genDeliveryScript returns a new script to be used to send our funds to in
2844
// the case of a cooperative channel close negotiation.
2845
func (p *Brontide) genDeliveryScript() ([]byte, error) {
9✔
2846
        // We'll send a normal p2wkh address unless we've negotiated the
9✔
2847
        // shutdown-any-segwit feature.
9✔
2848
        addrType := lnwallet.WitnessPubKey
9✔
2849
        if p.taprootShutdownAllowed() {
12✔
2850
                addrType = lnwallet.TaprootPubkey
3✔
2851
        }
3✔
2852

2853
        deliveryAddr, err := p.cfg.Wallet.NewAddress(
9✔
2854
                addrType, false, lnwallet.DefaultAccountName,
9✔
2855
        )
9✔
2856
        if err != nil {
9✔
2857
                return nil, err
×
2858
        }
×
2859
        p.log.Infof("Delivery addr for channel close: %v",
9✔
2860
                deliveryAddr)
9✔
2861

9✔
2862
        return txscript.PayToAddrScript(deliveryAddr)
9✔
2863
}
2864

2865
// channelManager is goroutine dedicated to handling all requests/signals
2866
// pertaining to the opening, cooperative closing, and force closing of all
2867
// channels maintained with the remote peer.
2868
//
2869
// NOTE: This method MUST be run as a goroutine.
2870
func (p *Brontide) channelManager() {
20✔
2871
        defer p.cg.WgDone()
20✔
2872

20✔
2873
        // reenableTimeout will fire once after the configured channel status
20✔
2874
        // interval has elapsed. This will trigger us to sign new channel
20✔
2875
        // updates and broadcast them with the "disabled" flag unset.
20✔
2876
        reenableTimeout := time.After(p.cfg.ChanActiveTimeout)
20✔
2877

20✔
2878
out:
20✔
2879
        for {
61✔
2880
                select {
41✔
2881
                // A new pending channel has arrived which means we are about
2882
                // to complete a funding workflow and is waiting for the final
2883
                // `ChannelReady` messages to be exchanged. We will add this
2884
                // channel to the `activeChannels` with a nil value to indicate
2885
                // this is a pending channel.
2886
                case req := <-p.newPendingChannel:
4✔
2887
                        p.handleNewPendingChannel(req)
4✔
2888

2889
                // A new channel has arrived which means we've just completed a
2890
                // funding workflow. We'll initialize the necessary local
2891
                // state, and notify the htlc switch of a new link.
2892
                case req := <-p.newActiveChannel:
3✔
2893
                        p.handleNewActiveChannel(req)
3✔
2894

2895
                // The funding flow for a pending channel is failed, we will
2896
                // remove it from Brontide.
2897
                case req := <-p.removePendingChannel:
4✔
2898
                        p.handleRemovePendingChannel(req)
4✔
2899

2900
                // We've just received a local request to close an active
2901
                // channel. It will either kick of a cooperative channel
2902
                // closure negotiation, or be a notification of a breached
2903
                // contract that should be abandoned.
2904
                case req := <-p.localCloseChanReqs:
10✔
2905
                        p.handleLocalCloseReq(req)
10✔
2906

2907
                // We've received a link failure from a link that was added to
2908
                // the switch. This will initiate the teardown of the link, and
2909
                // initiate any on-chain closures if necessary.
2910
                case failure := <-p.linkFailures:
3✔
2911
                        p.handleLinkFailure(failure)
3✔
2912

2913
                // We've received a new cooperative channel closure related
2914
                // message from the remote peer, we'll use this message to
2915
                // advance the chan closer state machine.
2916
                case closeMsg := <-p.chanCloseMsgs:
16✔
2917
                        p.handleCloseMsg(closeMsg)
16✔
2918

2919
                // The channel reannounce delay has elapsed, broadcast the
2920
                // reenabled channel updates to the network. This should only
2921
                // fire once, so we set the reenableTimeout channel to nil to
2922
                // mark it for garbage collection. If the peer is torn down
2923
                // before firing, reenabling will not be attempted.
2924
                // TODO(conner): consolidate reenables timers inside chan status
2925
                // manager
2926
                case <-reenableTimeout:
3✔
2927
                        p.reenableActiveChannels()
3✔
2928

3✔
2929
                        // Since this channel will never fire again during the
3✔
2930
                        // lifecycle of the peer, we nil the channel to mark it
3✔
2931
                        // eligible for garbage collection, and make this
3✔
2932
                        // explicitly ineligible to receive in future calls to
3✔
2933
                        // select. This also shaves a few CPU cycles since the
3✔
2934
                        // select will ignore this case entirely.
3✔
2935
                        reenableTimeout = nil
3✔
2936

3✔
2937
                        // Once the reenabling is attempted, we also cancel the
3✔
2938
                        // channel event subscription to free up the overflow
3✔
2939
                        // queue used in channel notifier.
3✔
2940
                        //
3✔
2941
                        // NOTE: channelEventClient will be nil if the
3✔
2942
                        // reenableTimeout is greater than 1 minute.
3✔
2943
                        if p.channelEventClient != nil {
6✔
2944
                                p.channelEventClient.Cancel()
3✔
2945
                        }
3✔
2946

2947
                case <-p.cg.Done():
3✔
2948
                        // As, we've been signalled to exit, we'll reset all
3✔
2949
                        // our active channel back to their default state.
3✔
2950
                        p.activeChannels.ForEach(func(_ lnwire.ChannelID,
3✔
2951
                                lc *lnwallet.LightningChannel) error {
6✔
2952

3✔
2953
                                // Exit if the channel is nil as it's a pending
3✔
2954
                                // channel.
3✔
2955
                                if lc == nil {
6✔
2956
                                        return nil
3✔
2957
                                }
3✔
2958

2959
                                lc.ResetState()
3✔
2960

3✔
2961
                                return nil
3✔
2962
                        })
2963

2964
                        break out
3✔
2965
                }
2966
        }
2967
}
2968

2969
// reenableActiveChannels searches the index of channels maintained with this
2970
// peer, and reenables each public, non-pending channel. This is done at the
2971
// gossip level by broadcasting a new ChannelUpdate with the disabled bit unset.
2972
// No message will be sent if the channel is already enabled.
2973
func (p *Brontide) reenableActiveChannels() {
3✔
2974
        // First, filter all known channels with this peer for ones that are
3✔
2975
        // both public and not pending.
3✔
2976
        activePublicChans := p.filterChannelsToEnable()
3✔
2977

3✔
2978
        // Create a map to hold channels that needs to be retried.
3✔
2979
        retryChans := make(map[wire.OutPoint]struct{}, len(activePublicChans))
3✔
2980

3✔
2981
        // For each of the public, non-pending channels, set the channel
3✔
2982
        // disabled bit to false and send out a new ChannelUpdate. If this
3✔
2983
        // channel is already active, the update won't be sent.
3✔
2984
        for _, chanPoint := range activePublicChans {
6✔
2985
                err := p.cfg.ChanStatusMgr.RequestEnable(chanPoint, false)
3✔
2986

3✔
2987
                switch {
3✔
2988
                // No error occurred, continue to request the next channel.
2989
                case err == nil:
3✔
2990
                        continue
3✔
2991

2992
                // Cannot auto enable a manually disabled channel so we do
2993
                // nothing but proceed to the next channel.
2994
                case errors.Is(err, netann.ErrEnableManuallyDisabledChan):
3✔
2995
                        p.log.Debugf("Channel(%v) was manually disabled, "+
3✔
2996
                                "ignoring automatic enable request", chanPoint)
3✔
2997

3✔
2998
                        continue
3✔
2999

3000
                // If the channel is reported as inactive, we will give it
3001
                // another chance. When handling the request, ChanStatusManager
3002
                // will check whether the link is active or not. One of the
3003
                // conditions is whether the link has been marked as
3004
                // reestablished, which happens inside a goroutine(htlcManager)
3005
                // after the link is started. And we may get a false negative
3006
                // saying the link is not active because that goroutine hasn't
3007
                // reached the line to mark the reestablishment. Thus we give
3008
                // it a second chance to send the request.
3009
                case errors.Is(err, netann.ErrEnableInactiveChan):
×
3010
                        // If we don't have a client created, it means we
×
3011
                        // shouldn't retry enabling the channel.
×
3012
                        if p.channelEventClient == nil {
×
3013
                                p.log.Errorf("Channel(%v) request enabling "+
×
3014
                                        "failed due to inactive link",
×
3015
                                        chanPoint)
×
3016

×
3017
                                continue
×
3018
                        }
3019

3020
                        p.log.Warnf("Channel(%v) cannot be enabled as " +
×
3021
                                "ChanStatusManager reported inactive, retrying")
×
3022

×
3023
                        // Add the channel to the retry map.
×
3024
                        retryChans[chanPoint] = struct{}{}
×
3025
                }
3026
        }
3027

3028
        // Retry the channels if we have any.
3029
        if len(retryChans) != 0 {
3✔
3030
                p.retryRequestEnable(retryChans)
×
3031
        }
×
3032
}
3033

3034
// fetchActiveChanCloser attempts to fetch the active chan closer state machine
3035
// for the target channel ID. If the channel isn't active an error is returned.
3036
// Otherwise, either an existing state machine will be returned, or a new one
3037
// will be created.
3038
func (p *Brontide) fetchActiveChanCloser(chanID lnwire.ChannelID) (
3039
        *chanCloserFsm, error) {
16✔
3040

16✔
3041
        chanCloser, found := p.activeChanCloses.Load(chanID)
16✔
3042
        if found {
29✔
3043
                // An entry will only be found if the closer has already been
13✔
3044
                // created for a non-pending channel or for a channel that had
13✔
3045
                // previously started the shutdown process but the connection
13✔
3046
                // was restarted.
13✔
3047
                return &chanCloser, nil
13✔
3048
        }
13✔
3049

3050
        // First, we'll ensure that we actually know of the target channel. If
3051
        // not, we'll ignore this message.
3052
        channel, ok := p.activeChannels.Load(chanID)
6✔
3053

6✔
3054
        // If the channel isn't in the map or the channel is nil, return
6✔
3055
        // ErrChannelNotFound as the channel is pending.
6✔
3056
        if !ok || channel == nil {
9✔
3057
                return nil, ErrChannelNotFound
3✔
3058
        }
3✔
3059

3060
        // We'll create a valid closing state machine in order to respond to
3061
        // the initiated cooperative channel closure. First, we set the
3062
        // delivery script that our funds will be paid out to. If an upfront
3063
        // shutdown script was set, we will use it. Otherwise, we get a fresh
3064
        // delivery script.
3065
        //
3066
        // TODO: Expose option to allow upfront shutdown script from watch-only
3067
        // accounts.
3068
        deliveryScript := channel.LocalUpfrontShutdownScript()
6✔
3069
        if len(deliveryScript) == 0 {
12✔
3070
                var err error
6✔
3071
                deliveryScript, err = p.genDeliveryScript()
6✔
3072
                if err != nil {
6✔
3073
                        p.log.Errorf("unable to gen delivery script: %v",
×
3074
                                err)
×
3075
                        return nil, fmt.Errorf("close addr unavailable")
×
3076
                }
×
3077
        }
3078

3079
        // In order to begin fee negotiations, we'll first compute our target
3080
        // ideal fee-per-kw.
3081
        feePerKw, err := p.cfg.FeeEstimator.EstimateFeePerKW(
6✔
3082
                p.cfg.CoopCloseTargetConfs,
6✔
3083
        )
6✔
3084
        if err != nil {
6✔
3085
                p.log.Errorf("unable to query fee estimator: %v", err)
×
3086
                return nil, fmt.Errorf("unable to estimate fee")
×
3087
        }
×
3088

3089
        addr, err := p.addrWithInternalKey(deliveryScript)
6✔
3090
        if err != nil {
6✔
3091
                return nil, fmt.Errorf("unable to parse addr: %w", err)
×
3092
        }
×
3093
        negotiateChanCloser, err := p.createChanCloser(
6✔
3094
                channel, addr, feePerKw, nil, lntypes.Remote,
6✔
3095
        )
6✔
3096
        if err != nil {
6✔
3097
                p.log.Errorf("unable to create chan closer: %v", err)
×
3098
                return nil, fmt.Errorf("unable to create chan closer")
×
3099
        }
×
3100

3101
        chanCloser = makeNegotiateCloser(negotiateChanCloser)
6✔
3102

6✔
3103
        p.activeChanCloses.Store(chanID, chanCloser)
6✔
3104

6✔
3105
        return &chanCloser, nil
6✔
3106
}
3107

3108
// filterChannelsToEnable filters a list of channels to be enabled upon start.
3109
// The filtered channels are active channels that's neither private nor
3110
// pending.
3111
func (p *Brontide) filterChannelsToEnable() []wire.OutPoint {
3✔
3112
        var activePublicChans []wire.OutPoint
3✔
3113

3✔
3114
        p.activeChannels.Range(func(chanID lnwire.ChannelID,
3✔
3115
                lnChan *lnwallet.LightningChannel) bool {
6✔
3116

3✔
3117
                // If the lnChan is nil, continue as this is a pending channel.
3✔
3118
                if lnChan == nil {
4✔
3119
                        return true
1✔
3120
                }
1✔
3121

3122
                dbChan := lnChan.State()
3✔
3123
                isPublic := dbChan.ChannelFlags&lnwire.FFAnnounceChannel != 0
3✔
3124
                if !isPublic || dbChan.IsPending {
3✔
3125
                        return true
×
3126
                }
×
3127

3128
                // We'll also skip any channels added during this peer's
3129
                // lifecycle since they haven't waited out the timeout. Their
3130
                // first announcement will be enabled, and the chan status
3131
                // manager will begin monitoring them passively since they exist
3132
                // in the database.
3133
                if _, ok := p.addedChannels.Load(chanID); ok {
5✔
3134
                        return true
2✔
3135
                }
2✔
3136

3137
                activePublicChans = append(
3✔
3138
                        activePublicChans, dbChan.FundingOutpoint,
3✔
3139
                )
3✔
3140

3✔
3141
                return true
3✔
3142
        })
3143

3144
        return activePublicChans
3✔
3145
}
3146

3147
// retryRequestEnable takes a map of channel outpoints and a channel event
3148
// client. It listens to the channel events and removes a channel from the map
3149
// if it's matched to the event. Upon receiving an active channel event, it
3150
// will send the enabling request again.
3151
func (p *Brontide) retryRequestEnable(activeChans map[wire.OutPoint]struct{}) {
×
3152
        p.log.Debugf("Retry enabling %v channels", len(activeChans))
×
3153

×
3154
        // retryEnable is a helper closure that sends an enable request and
×
3155
        // removes the channel from the map if it's matched.
×
3156
        retryEnable := func(chanPoint wire.OutPoint) error {
×
3157
                // If this is an active channel event, check whether it's in
×
3158
                // our targeted channels map.
×
3159
                _, found := activeChans[chanPoint]
×
3160

×
3161
                // If this channel is irrelevant, return nil so the loop can
×
3162
                // jump to next iteration.
×
3163
                if !found {
×
3164
                        return nil
×
3165
                }
×
3166

3167
                // Otherwise we've just received an active signal for a channel
3168
                // that's previously failed to be enabled, we send the request
3169
                // again.
3170
                //
3171
                // We only give the channel one more shot, so we delete it from
3172
                // our map first to keep it from being attempted again.
3173
                delete(activeChans, chanPoint)
×
3174

×
3175
                // Send the request.
×
3176
                err := p.cfg.ChanStatusMgr.RequestEnable(chanPoint, false)
×
3177
                if err != nil {
×
3178
                        return fmt.Errorf("request enabling channel %v "+
×
3179
                                "failed: %w", chanPoint, err)
×
3180
                }
×
3181

3182
                return nil
×
3183
        }
3184

3185
        for {
×
3186
                // If activeChans is empty, we've done processing all the
×
3187
                // channels.
×
3188
                if len(activeChans) == 0 {
×
3189
                        p.log.Debug("Finished retry enabling channels")
×
3190
                        return
×
3191
                }
×
3192

3193
                select {
×
3194
                // A new event has been sent by the ChannelNotifier. We now
3195
                // check whether it's an active or inactive channel event.
3196
                case e := <-p.channelEventClient.Updates():
×
3197
                        // If this is an active channel event, try enable the
×
3198
                        // channel then jump to the next iteration.
×
3199
                        active, ok := e.(channelnotifier.ActiveChannelEvent)
×
3200
                        if ok {
×
3201
                                chanPoint := *active.ChannelPoint
×
3202

×
3203
                                // If we received an error for this particular
×
3204
                                // channel, we log an error and won't quit as
×
3205
                                // we still want to retry other channels.
×
3206
                                if err := retryEnable(chanPoint); err != nil {
×
3207
                                        p.log.Errorf("Retry failed: %v", err)
×
3208
                                }
×
3209

3210
                                continue
×
3211
                        }
3212

3213
                        // Otherwise check for inactive link event, and jump to
3214
                        // next iteration if it's not.
3215
                        inactive, ok := e.(channelnotifier.InactiveLinkEvent)
×
3216
                        if !ok {
×
3217
                                continue
×
3218
                        }
3219

3220
                        // Found an inactive link event, if this is our
3221
                        // targeted channel, remove it from our map.
3222
                        chanPoint := *inactive.ChannelPoint
×
3223
                        _, found := activeChans[chanPoint]
×
3224
                        if !found {
×
3225
                                continue
×
3226
                        }
3227

3228
                        delete(activeChans, chanPoint)
×
3229
                        p.log.Warnf("Re-enable channel %v failed, received "+
×
3230
                                "inactive link event", chanPoint)
×
3231

3232
                case <-p.cg.Done():
×
3233
                        p.log.Debugf("Peer shutdown during retry enabling")
×
3234
                        return
×
3235
                }
3236
        }
3237
}
3238

3239
// chooseDeliveryScript takes two optionally set shutdown scripts and returns
3240
// a suitable script to close out to. This may be nil if neither script is
3241
// set. If both scripts are set, this function will error if they do not match.
3242
func chooseDeliveryScript(upfront, requested lnwire.DeliveryAddress,
3243
        genDeliveryScript func() ([]byte, error),
3244
) (lnwire.DeliveryAddress, error) {
15✔
3245

15✔
3246
        switch {
15✔
3247
        // If no script was provided, then we'll generate a new delivery script.
3248
        case len(upfront) == 0 && len(requested) == 0:
7✔
3249
                return genDeliveryScript()
7✔
3250

3251
        // If no upfront shutdown script was provided, return the user
3252
        // requested address (which may be nil).
3253
        case len(upfront) == 0:
5✔
3254
                return requested, nil
5✔
3255

3256
        // If an upfront shutdown script was provided, and the user did not
3257
        // request a custom shutdown script, return the upfront address.
3258
        case len(requested) == 0:
5✔
3259
                return upfront, nil
5✔
3260

3261
        // If both an upfront shutdown script and a custom close script were
3262
        // provided, error if the user provided shutdown script does not match
3263
        // the upfront shutdown script (because closing out to a different
3264
        // script would violate upfront shutdown).
3265
        case !bytes.Equal(upfront, requested):
2✔
3266
                return nil, chancloser.ErrUpfrontShutdownScriptMismatch
2✔
3267

3268
        // The user requested script matches the upfront shutdown script, so we
3269
        // can return it without error.
3270
        default:
2✔
3271
                return upfront, nil
2✔
3272
        }
3273
}
3274

3275
// restartCoopClose checks whether we need to restart the cooperative close
3276
// process for a given channel.
3277
func (p *Brontide) restartCoopClose(lnChan *lnwallet.LightningChannel) (
3278
        *lnwire.Shutdown, error) {
3✔
3279

3✔
3280
        isTaprootChan := lnChan.ChanType().IsTaproot()
3✔
3281

3✔
3282
        // If this channel has status ChanStatusCoopBroadcasted and does not
3✔
3283
        // have a closing transaction, then the cooperative close process was
3✔
3284
        // started but never finished. We'll re-create the chanCloser state
3✔
3285
        // machine and resend Shutdown. BOLT#2 requires that we retransmit
3✔
3286
        // Shutdown exactly, but doing so would mean persisting the RPC
3✔
3287
        // provided close script. Instead use the LocalUpfrontShutdownScript
3✔
3288
        // or generate a script.
3✔
3289
        c := lnChan.State()
3✔
3290
        _, err := c.BroadcastedCooperative()
3✔
3291
        if err != nil && err != channeldb.ErrNoCloseTx {
3✔
3292
                // An error other than ErrNoCloseTx was encountered.
×
3293
                return nil, err
×
3294
        } else if err == nil && !p.rbfCoopCloseAllowed() {
3✔
3295
                // This is a channel that doesn't support RBF coop close, and it
×
3296
                // already had a coop close txn broadcast. As a result, we can
×
3297
                // just exit here as all we can do is wait for it to confirm.
×
3298
                return nil, nil
×
3299
        }
×
3300

3301
        chanID := lnwire.NewChanIDFromOutPoint(c.FundingOutpoint)
3✔
3302

3✔
3303
        var deliveryScript []byte
3✔
3304

3✔
3305
        shutdownInfo, err := c.ShutdownInfo()
3✔
3306
        switch {
3✔
3307
        // We have previously stored the delivery script that we need to use
3308
        // in the shutdown message. Re-use this script.
3309
        case err == nil:
3✔
3310
                shutdownInfo.WhenSome(func(info channeldb.ShutdownInfo) {
6✔
3311
                        deliveryScript = info.DeliveryScript.Val
3✔
3312
                })
3✔
3313

3314
        // An error other than ErrNoShutdownInfo was returned
3315
        case !errors.Is(err, channeldb.ErrNoShutdownInfo):
×
3316
                return nil, err
×
3317

3318
        case errors.Is(err, channeldb.ErrNoShutdownInfo):
×
3319
                deliveryScript = c.LocalShutdownScript
×
3320
                if len(deliveryScript) == 0 {
×
3321
                        var err error
×
3322
                        deliveryScript, err = p.genDeliveryScript()
×
3323
                        if err != nil {
×
3324
                                p.log.Errorf("unable to gen delivery script: "+
×
3325
                                        "%v", err)
×
3326

×
3327
                                return nil, fmt.Errorf("close addr unavailable")
×
3328
                        }
×
3329
                }
3330
        }
3331

3332
        // If the new RBF co-op close is negotiated, then we'll init and start
3333
        // that state machine, skipping the steps for the negotiate machine
3334
        // below. We don't support this close type for taproot channels though.
3335
        if p.rbfCoopCloseAllowed() && !isTaprootChan {
6✔
3336
                _, err := p.initRbfChanCloser(lnChan)
3✔
3337
                if err != nil {
3✔
3338
                        return nil, fmt.Errorf("unable to init rbf chan "+
×
3339
                                "closer during restart: %w", err)
×
3340
                }
×
3341

3342
                shutdownDesc := fn.MapOption(
3✔
3343
                        newRestartShutdownInit,
3✔
3344
                )(shutdownInfo)
3✔
3345

3✔
3346
                err = p.startRbfChanCloser(
3✔
3347
                        fn.FlattenOption(shutdownDesc), lnChan.ChannelPoint(),
3✔
3348
                )
3✔
3349

3✔
3350
                return nil, err
3✔
3351
        }
3352

3353
        // Compute an ideal fee.
3354
        feePerKw, err := p.cfg.FeeEstimator.EstimateFeePerKW(
×
3355
                p.cfg.CoopCloseTargetConfs,
×
3356
        )
×
3357
        if err != nil {
×
3358
                p.log.Errorf("unable to query fee estimator: %v", err)
×
3359
                return nil, fmt.Errorf("unable to estimate fee")
×
3360
        }
×
3361

3362
        // Determine whether we or the peer are the initiator of the coop
3363
        // close attempt by looking at the channel's status.
3364
        closingParty := lntypes.Remote
×
3365
        if c.HasChanStatus(channeldb.ChanStatusLocalCloseInitiator) {
×
3366
                closingParty = lntypes.Local
×
3367
        }
×
3368

3369
        addr, err := p.addrWithInternalKey(deliveryScript)
×
3370
        if err != nil {
×
3371
                return nil, fmt.Errorf("unable to parse addr: %w", err)
×
3372
        }
×
3373
        chanCloser, err := p.createChanCloser(
×
3374
                lnChan, addr, feePerKw, nil, closingParty,
×
3375
        )
×
3376
        if err != nil {
×
3377
                p.log.Errorf("unable to create chan closer: %v", err)
×
3378
                return nil, fmt.Errorf("unable to create chan closer")
×
3379
        }
×
3380

3381
        p.activeChanCloses.Store(chanID, makeNegotiateCloser(chanCloser))
×
3382

×
3383
        // Create the Shutdown message.
×
3384
        shutdownMsg, err := chanCloser.ShutdownChan()
×
3385
        if err != nil {
×
3386
                p.log.Errorf("unable to create shutdown message: %v", err)
×
3387
                p.activeChanCloses.Delete(chanID)
×
3388
                return nil, err
×
3389
        }
×
3390

3391
        return shutdownMsg, nil
×
3392
}
3393

3394
// createChanCloser constructs a ChanCloser from the passed parameters and is
3395
// used to de-duplicate code.
3396
func (p *Brontide) createChanCloser(channel *lnwallet.LightningChannel,
3397
        deliveryScript *chancloser.DeliveryAddrWithKey,
3398
        fee chainfee.SatPerKWeight, req *htlcswitch.ChanClose,
3399
        closer lntypes.ChannelParty) (*chancloser.ChanCloser, error) {
12✔
3400

12✔
3401
        _, startingHeight, err := p.cfg.ChainIO.GetBestBlock()
12✔
3402
        if err != nil {
12✔
3403
                p.log.Errorf("unable to obtain best block: %v", err)
×
3404
                return nil, fmt.Errorf("cannot obtain best block")
×
3405
        }
×
3406

3407
        // The req will only be set if we initiated the co-op closing flow.
3408
        var maxFee chainfee.SatPerKWeight
12✔
3409
        if req != nil {
21✔
3410
                maxFee = req.MaxFee
9✔
3411
        }
9✔
3412

3413
        chanCloser := chancloser.NewChanCloser(
12✔
3414
                chancloser.ChanCloseCfg{
12✔
3415
                        Channel:      channel,
12✔
3416
                        MusigSession: NewMusigChanCloser(channel),
12✔
3417
                        FeeEstimator: &chancloser.SimpleCoopFeeEstimator{},
12✔
3418
                        BroadcastTx:  p.cfg.Wallet.PublishTransaction,
12✔
3419
                        AuxCloser:    p.cfg.AuxChanCloser,
12✔
3420
                        DisableChannel: func(op wire.OutPoint) error {
24✔
3421
                                return p.cfg.ChanStatusMgr.RequestDisable(
12✔
3422
                                        op, false,
12✔
3423
                                )
12✔
3424
                        },
12✔
3425
                        MaxFee: maxFee,
3426
                        Disconnect: func() error {
×
3427
                                return p.cfg.DisconnectPeer(p.IdentityKey())
×
3428
                        },
×
3429
                        ChainParams: &p.cfg.Wallet.Cfg.NetParams,
3430
                },
3431
                *deliveryScript,
3432
                fee,
3433
                uint32(startingHeight),
3434
                req,
3435
                closer,
3436
        )
3437

3438
        return chanCloser, nil
12✔
3439
}
3440

3441
// initNegotiateChanCloser initializes the channel closer for a channel that is
3442
// using the original "negotiation" based protocol. This path is used when
3443
// we're the one initiating the channel close.
3444
//
3445
// TODO(roasbeef): can make a MsgEndpoint for existing handling logic to
3446
// further abstract.
3447
func (p *Brontide) initNegotiateChanCloser(req *htlcswitch.ChanClose,
3448
        channel *lnwallet.LightningChannel) error {
10✔
3449

10✔
3450
        // First, we'll choose a delivery address that we'll use to send the
10✔
3451
        // funds to in the case of a successful negotiation.
10✔
3452

10✔
3453
        // An upfront shutdown and user provided script are both optional, but
10✔
3454
        // must be equal if both set  (because we cannot serve a request to
10✔
3455
        // close out to a script which violates upfront shutdown). Get the
10✔
3456
        // appropriate address to close out to (which may be nil if neither are
10✔
3457
        // set) and error if they are both set and do not match.
10✔
3458
        deliveryScript, err := chooseDeliveryScript(
10✔
3459
                channel.LocalUpfrontShutdownScript(), req.DeliveryScript,
10✔
3460
                p.genDeliveryScript,
10✔
3461
        )
10✔
3462
        if err != nil {
11✔
3463
                return fmt.Errorf("cannot close channel %v: %w",
1✔
3464
                        req.ChanPoint, err)
1✔
3465
        }
1✔
3466

3467
        addr, err := p.addrWithInternalKey(deliveryScript)
9✔
3468
        if err != nil {
9✔
3469
                return fmt.Errorf("unable to parse addr for channel "+
×
3470
                        "%v: %w", req.ChanPoint, err)
×
3471
        }
×
3472

3473
        chanCloser, err := p.createChanCloser(
9✔
3474
                channel, addr, req.TargetFeePerKw, req, lntypes.Local,
9✔
3475
        )
9✔
3476
        if err != nil {
9✔
3477
                return fmt.Errorf("unable to make chan closer: %w", err)
×
3478
        }
×
3479

3480
        chanID := lnwire.NewChanIDFromOutPoint(channel.ChannelPoint())
9✔
3481
        p.activeChanCloses.Store(chanID, makeNegotiateCloser(chanCloser))
9✔
3482

9✔
3483
        // Finally, we'll initiate the channel shutdown within the
9✔
3484
        // chanCloser, and send the shutdown message to the remote
9✔
3485
        // party to kick things off.
9✔
3486
        shutdownMsg, err := chanCloser.ShutdownChan()
9✔
3487
        if err != nil {
9✔
3488
                // As we were unable to shutdown the channel, we'll return it
×
3489
                // back to its normal state.
×
3490
                defer channel.ResetState()
×
3491

×
3492
                p.activeChanCloses.Delete(chanID)
×
3493

×
3494
                return fmt.Errorf("unable to shutdown channel: %w", err)
×
3495
        }
×
3496

3497
        link := p.fetchLinkFromKeyAndCid(chanID)
9✔
3498
        if link == nil {
9✔
3499
                // If the link is nil then it means it was already removed from
×
3500
                // the switch or it never existed in the first place. The
×
3501
                // latter case is handled at the beginning of this function, so
×
3502
                // in the case where it has already been removed, we can skip
×
3503
                // adding the commit hook to queue a Shutdown message.
×
3504
                p.log.Warnf("link not found during attempted closure: "+
×
3505
                        "%v", chanID)
×
3506
                return nil
×
3507
        }
×
3508

3509
        if !link.DisableAdds(htlcswitch.Outgoing) {
9✔
3510
                p.log.Warnf("Outgoing link adds already "+
×
3511
                        "disabled: %v", link.ChanID())
×
3512
        }
×
3513

3514
        link.OnCommitOnce(htlcswitch.Outgoing, func() {
18✔
3515
                p.queueMsg(shutdownMsg, nil)
9✔
3516
        })
9✔
3517

3518
        return nil
9✔
3519
}
3520

3521
// chooseAddr returns the provided address if it is non-zero length, otherwise
3522
// None.
3523
func chooseAddr(addr lnwire.DeliveryAddress) fn.Option[lnwire.DeliveryAddress] {
3✔
3524
        if len(addr) == 0 {
6✔
3525
                return fn.None[lnwire.DeliveryAddress]()
3✔
3526
        }
3✔
3527

3528
        return fn.Some(addr)
×
3529
}
3530

3531
// observeRbfCloseUpdates observes the channel for any updates that may
3532
// indicate that a new txid has been broadcasted, or the channel fully closed
3533
// on chain.
3534
func (p *Brontide) observeRbfCloseUpdates(chanCloser *chancloser.RbfChanCloser,
3535
        closeReq *htlcswitch.ChanClose,
3536
        coopCloseStates chancloser.RbfStateSub) {
3✔
3537

3✔
3538
        newStateChan := coopCloseStates.NewItemCreated.ChanOut()
3✔
3539
        defer chanCloser.RemoveStateSub(coopCloseStates)
3✔
3540

3✔
3541
        var (
3✔
3542
                lastTxids    lntypes.Dual[chainhash.Hash]
3✔
3543
                lastFeeRates lntypes.Dual[chainfee.SatPerVByte]
3✔
3544
        )
3✔
3545

3✔
3546
        maybeNotifyTxBroadcast := func(state chancloser.AsymmetricPeerState,
3✔
3547
                party lntypes.ChannelParty) {
6✔
3548

3✔
3549
                // First, check to see if we have an error to report to the
3✔
3550
                // caller. If so, then we''ll return that error and exit, as the
3✔
3551
                // stream will exit as well.
3✔
3552
                if closeErr, ok := state.(*chancloser.CloseErr); ok {
6✔
3553
                        // We hit an error during the last state transition, so
3✔
3554
                        // we'll extract the error then send it to the
3✔
3555
                        // user.
3✔
3556
                        err := closeErr.Err()
3✔
3557

3✔
3558
                        peerLog.Warnf("ChannelPoint(%v): encountered close "+
3✔
3559
                                "err: %v", closeReq.ChanPoint, err)
3✔
3560

3✔
3561
                        select {
3✔
3562
                        case closeReq.Err <- err:
3✔
3563
                        case <-closeReq.Ctx.Done():
×
3564
                        case <-p.cg.Done():
×
3565
                        }
3566

3567
                        return
3✔
3568
                }
3569

3570
                closePending, ok := state.(*chancloser.ClosePending)
3✔
3571

3✔
3572
                // If this isn't the close pending state, we aren't at the
3✔
3573
                // terminal state yet.
3✔
3574
                if !ok {
6✔
3575
                        return
3✔
3576
                }
3✔
3577

3578
                // Only notify if the fee rate is greater.
3579
                newFeeRate := closePending.FeeRate
3✔
3580
                lastFeeRate := lastFeeRates.GetForParty(party)
3✔
3581
                if newFeeRate <= lastFeeRate {
6✔
3582
                        peerLog.Debugf("ChannelPoint(%v): remote party made "+
3✔
3583
                                "update for fee rate %v, but we already have "+
3✔
3584
                                "a higher fee rate of %v", closeReq.ChanPoint,
3✔
3585
                                newFeeRate, lastFeeRate)
3✔
3586

3✔
3587
                        return
3✔
3588
                }
3✔
3589

3590
                feeRate := closePending.FeeRate
3✔
3591
                lastFeeRates.SetForParty(party, feeRate)
3✔
3592

3✔
3593
                // At this point, we'll have a txid that we can use to notify
3✔
3594
                // the client, but only if it's different from the last one we
3✔
3595
                // sent. If the user attempted to bump, but was rejected due to
3✔
3596
                // RBF, then we'll send a redundant update.
3✔
3597
                closingTxid := closePending.CloseTx.TxHash()
3✔
3598
                lastTxid := lastTxids.GetForParty(party)
3✔
3599
                if closeReq != nil && closingTxid != lastTxid {
6✔
3600
                        select {
3✔
3601
                        case closeReq.Updates <- &PendingUpdate{
3602
                                Txid:        closingTxid[:],
3603
                                FeePerVbyte: fn.Some(closePending.FeeRate),
3604
                                IsLocalCloseTx: fn.Some(
3605
                                        party == lntypes.Local,
3606
                                ),
3607
                        }:
3✔
3608

3609
                        case <-closeReq.Ctx.Done():
×
3610
                                return
×
3611

3612
                        case <-p.cg.Done():
×
3613
                                return
×
3614
                        }
3615
                }
3616

3617
                lastTxids.SetForParty(party, closingTxid)
3✔
3618
        }
3619

3620
        peerLog.Infof("Observing RBF close updates for channel %v",
3✔
3621
                closeReq.ChanPoint)
3✔
3622

3✔
3623
        // We'll consume each new incoming state to send out the appropriate
3✔
3624
        // RPC update.
3✔
3625
        for {
6✔
3626
                select {
3✔
3627
                case newState := <-newStateChan:
3✔
3628

3✔
3629
                        switch closeState := newState.(type) {
3✔
3630
                        // Once we've reached the state of pending close, we
3631
                        // have a txid that we broadcasted.
3632
                        case *chancloser.ClosingNegotiation:
3✔
3633
                                peerState := closeState.PeerState
3✔
3634

3✔
3635
                                // Each side may have gained a new co-op close
3✔
3636
                                // tx, so we'll examine both to see if they've
3✔
3637
                                // changed.
3✔
3638
                                maybeNotifyTxBroadcast(
3✔
3639
                                        peerState.GetForParty(lntypes.Local),
3✔
3640
                                        lntypes.Local,
3✔
3641
                                )
3✔
3642
                                maybeNotifyTxBroadcast(
3✔
3643
                                        peerState.GetForParty(lntypes.Remote),
3✔
3644
                                        lntypes.Remote,
3✔
3645
                                )
3✔
3646

3647
                        // Otherwise, if we're transition to CloseFin, then we
3648
                        // know that we're done.
3649
                        case *chancloser.CloseFin:
3✔
3650
                                // To clean up, we'll remove the chan closer
3✔
3651
                                // from the active map, and send the final
3✔
3652
                                // update to the client.
3✔
3653
                                closingTxid := closeState.ConfirmedTx.TxHash()
3✔
3654
                                if closeReq != nil {
6✔
3655
                                        closeReq.Updates <- &ChannelCloseUpdate{
3✔
3656
                                                ClosingTxid: closingTxid[:],
3✔
3657
                                                Success:     true,
3✔
3658
                                        }
3✔
3659
                                }
3✔
3660
                                chanID := lnwire.NewChanIDFromOutPoint(
3✔
3661
                                        *closeReq.ChanPoint,
3✔
3662
                                )
3✔
3663
                                p.activeChanCloses.Delete(chanID)
3✔
3664

3✔
3665
                                return
3✔
3666
                        }
3667

3668
                case <-closeReq.Ctx.Done():
3✔
3669
                        return
3✔
3670

3671
                case <-p.cg.Done():
3✔
3672
                        return
3✔
3673
                }
3674
        }
3675
}
3676

3677
// chanErrorReporter is a simple implementation of the
3678
// chancloser.ErrorReporter. This is bound to a single channel by the channel
3679
// ID.
3680
type chanErrorReporter struct {
3681
        chanID lnwire.ChannelID
3682
        peer   *Brontide
3683
}
3684

3685
// newChanErrorReporter creates a new instance of the chanErrorReporter.
3686
func newChanErrorReporter(chanID lnwire.ChannelID,
3687
        peer *Brontide) *chanErrorReporter {
3✔
3688

3✔
3689
        return &chanErrorReporter{
3✔
3690
                chanID: chanID,
3✔
3691
                peer:   peer,
3✔
3692
        }
3✔
3693
}
3✔
3694

3695
// ReportError is a method that's used to report an error that occurred during
3696
// state machine execution. This is used by the RBF close state machine to
3697
// terminate the state machine and send an error to the remote peer.
3698
//
3699
// This is a part of the chancloser.ErrorReporter interface.
3700
func (c *chanErrorReporter) ReportError(chanErr error) {
×
3701
        c.peer.log.Errorf("coop close error for channel %v: %v",
×
3702
                c.chanID, chanErr)
×
3703

×
3704
        var errMsg []byte
×
3705
        if errors.Is(chanErr, chancloser.ErrInvalidStateTransition) {
×
3706
                errMsg = []byte("unexpected protocol message")
×
3707
        } else {
×
3708
                errMsg = []byte(chanErr.Error())
×
3709
        }
×
3710

3711
        err := c.peer.SendMessageLazy(false, &lnwire.Error{
×
3712
                ChanID: c.chanID,
×
3713
                Data:   errMsg,
×
3714
        })
×
3715
        if err != nil {
×
3716
                c.peer.log.Warnf("unable to send error message to peer: %v",
×
3717
                        err)
×
3718
        }
×
3719

3720
        // After we send the error message to the peer, we'll re-initialize the
3721
        // coop close state machine as they may send a shutdown message to
3722
        // retry the coop close.
3723
        lnChan, ok := c.peer.activeChannels.Load(c.chanID)
×
3724
        if !ok {
×
3725
                return
×
3726
        }
×
3727

3728
        if lnChan == nil {
×
3729
                c.peer.log.Debugf("channel %v is pending, not "+
×
3730
                        "re-initializing coop close state machine",
×
3731
                        c.chanID)
×
3732

×
3733
                return
×
3734
        }
×
3735

3736
        if _, err := c.peer.initRbfChanCloser(lnChan); err != nil {
×
3737
                c.peer.activeChanCloses.Delete(c.chanID)
×
3738

×
3739
                c.peer.log.Errorf("unable to init RBF chan closer after "+
×
3740
                        "error case: %v", err)
×
3741
        }
×
3742
}
3743

3744
// chanFlushEventSentinel is used to send the RBF coop close state machine the
3745
// channel flushed event. We'll wait until the state machine enters the
3746
// ChannelFlushing state, then request the link to send the event once flushed.
3747
//
3748
// NOTE: This MUST be run as a goroutine.
3749
func (p *Brontide) chanFlushEventSentinel(chanCloser *chancloser.RbfChanCloser,
3750
        link htlcswitch.ChannelUpdateHandler,
3751
        channel *lnwallet.LightningChannel) {
3✔
3752

3✔
3753
        defer p.cg.WgDone()
3✔
3754

3✔
3755
        // If there's no link, then the channel has already been flushed, so we
3✔
3756
        // don't need to continue.
3✔
3757
        if link == nil {
6✔
3758
                return
3✔
3759
        }
3✔
3760

3761
        coopCloseStates := chanCloser.RegisterStateEvents()
3✔
3762
        defer chanCloser.RemoveStateSub(coopCloseStates)
3✔
3763

3✔
3764
        newStateChan := coopCloseStates.NewItemCreated.ChanOut()
3✔
3765

3✔
3766
        sendChanFlushed := func() {
6✔
3767
                chanState := channel.StateSnapshot()
3✔
3768

3✔
3769
                peerLog.Infof("ChannelPoint(%v) has been flushed for co-op "+
3✔
3770
                        "close, sending event to chan closer",
3✔
3771
                        channel.ChannelPoint())
3✔
3772

3✔
3773
                chanBalances := chancloser.ShutdownBalances{
3✔
3774
                        LocalBalance:  chanState.LocalBalance,
3✔
3775
                        RemoteBalance: chanState.RemoteBalance,
3✔
3776
                }
3✔
3777
                ctx := context.Background()
3✔
3778
                chanCloser.SendEvent(ctx, &chancloser.ChannelFlushed{
3✔
3779
                        ShutdownBalances: chanBalances,
3✔
3780
                        FreshFlush:       true,
3✔
3781
                })
3✔
3782
        }
3✔
3783

3784
        // We'll wait until the channel enters the ChannelFlushing state. We
3785
        // exit after a success loop. As after the first RBF iteration, the
3786
        // channel will always be flushed.
3787
        for newState := range newStateChan {
6✔
3788
                if _, ok := newState.(*chancloser.ChannelFlushing); ok {
6✔
3789
                        peerLog.Infof("ChannelPoint(%v): rbf coop "+
3✔
3790
                                "close is awaiting a flushed state, "+
3✔
3791
                                "registering with link..., ",
3✔
3792
                                channel.ChannelPoint())
3✔
3793

3✔
3794
                        // Request the link to send the event once the channel
3✔
3795
                        // is flushed. We only need this event sent once, so we
3✔
3796
                        // can exit now.
3✔
3797
                        link.OnFlushedOnce(sendChanFlushed)
3✔
3798

3✔
3799
                        return
3✔
3800
                }
3✔
3801
        }
3802
}
3803

3804
// initRbfChanCloser initializes the channel closer for a channel that
3805
// is using the new RBF based co-op close protocol. This only creates the chan
3806
// closer, but doesn't attempt to trigger any manual state transitions.
3807
func (p *Brontide) initRbfChanCloser(
3808
        channel *lnwallet.LightningChannel) (*chancloser.RbfChanCloser, error) {
3✔
3809

3✔
3810
        chanID := lnwire.NewChanIDFromOutPoint(channel.ChannelPoint())
3✔
3811

3✔
3812
        link := p.fetchLinkFromKeyAndCid(chanID)
3✔
3813

3✔
3814
        _, startingHeight, err := p.cfg.ChainIO.GetBestBlock()
3✔
3815
        if err != nil {
3✔
3816
                return nil, fmt.Errorf("cannot obtain best block: %w", err)
×
3817
        }
×
3818

3819
        defaultFeePerKw, err := p.cfg.FeeEstimator.EstimateFeePerKW(
3✔
3820
                p.cfg.CoopCloseTargetConfs,
3✔
3821
        )
3✔
3822
        if err != nil {
3✔
3823
                return nil, fmt.Errorf("unable to estimate fee: %w", err)
×
3824
        }
×
3825

3826
        thawHeight, err := channel.AbsoluteThawHeight()
3✔
3827
        if err != nil {
3✔
3828
                return nil, fmt.Errorf("unable to get thaw height: %w", err)
×
3829
        }
×
3830

3831
        peerPub := *p.IdentityKey()
3✔
3832

3✔
3833
        msgMapper := chancloser.NewRbfMsgMapper(
3✔
3834
                uint32(startingHeight), chanID, peerPub,
3✔
3835
        )
3✔
3836

3✔
3837
        initialState := chancloser.ChannelActive{}
3✔
3838

3✔
3839
        scid := channel.ZeroConfRealScid().UnwrapOr(
3✔
3840
                channel.ShortChanID(),
3✔
3841
        )
3✔
3842

3✔
3843
        env := chancloser.Environment{
3✔
3844
                ChainParams:    p.cfg.Wallet.Cfg.NetParams,
3✔
3845
                ChanPeer:       peerPub,
3✔
3846
                ChanPoint:      channel.ChannelPoint(),
3✔
3847
                ChanID:         chanID,
3✔
3848
                Scid:           scid,
3✔
3849
                ChanType:       channel.ChanType(),
3✔
3850
                DefaultFeeRate: defaultFeePerKw.FeePerVByte(),
3✔
3851
                ThawHeight:     fn.Some(thawHeight),
3✔
3852
                RemoteUpfrontShutdown: chooseAddr(
3✔
3853
                        channel.RemoteUpfrontShutdownScript(),
3✔
3854
                ),
3✔
3855
                LocalUpfrontShutdown: chooseAddr(
3✔
3856
                        channel.LocalUpfrontShutdownScript(),
3✔
3857
                ),
3✔
3858
                NewDeliveryScript: func() (lnwire.DeliveryAddress, error) {
6✔
3859
                        return p.genDeliveryScript()
3✔
3860
                },
3✔
3861
                FeeEstimator: &chancloser.SimpleCoopFeeEstimator{},
3862
                CloseSigner:  channel,
3863
                ChanObserver: newChanObserver(
3864
                        channel, link, p.cfg.ChanStatusMgr,
3865
                ),
3866
        }
3867

3868
        spendEvent := protofsm.RegisterSpend[chancloser.ProtocolEvent]{
3✔
3869
                OutPoint:   channel.ChannelPoint(),
3✔
3870
                PkScript:   channel.FundingTxOut().PkScript,
3✔
3871
                HeightHint: channel.DeriveHeightHint(),
3✔
3872
                PostSpendEvent: fn.Some[chancloser.RbfSpendMapper](
3✔
3873
                        chancloser.SpendMapper,
3✔
3874
                ),
3✔
3875
        }
3✔
3876

3✔
3877
        daemonAdapters := NewLndDaemonAdapters(LndAdapterCfg{
3✔
3878
                MsgSender:     newPeerMsgSender(peerPub, p),
3✔
3879
                TxBroadcaster: p.cfg.Wallet,
3✔
3880
                ChainNotifier: p.cfg.ChainNotifier,
3✔
3881
        })
3✔
3882

3✔
3883
        protoCfg := chancloser.RbfChanCloserCfg{
3✔
3884
                Daemon:        daemonAdapters,
3✔
3885
                InitialState:  &initialState,
3✔
3886
                Env:           &env,
3✔
3887
                InitEvent:     fn.Some[protofsm.DaemonEvent](&spendEvent),
3✔
3888
                ErrorReporter: newChanErrorReporter(chanID, p),
3✔
3889
                MsgMapper: fn.Some[protofsm.MsgMapper[chancloser.ProtocolEvent]]( //nolint:ll
3✔
3890
                        msgMapper,
3✔
3891
                ),
3✔
3892
        }
3✔
3893

3✔
3894
        ctx := context.Background()
3✔
3895
        chanCloser := protofsm.NewStateMachine(protoCfg)
3✔
3896
        chanCloser.Start(ctx)
3✔
3897

3✔
3898
        // Finally, we'll register this new endpoint with the message router so
3✔
3899
        // future co-op close messages are handled by this state machine.
3✔
3900
        err = fn.MapOptionZ(p.msgRouter, func(r msgmux.Router) error {
6✔
3901
                _ = r.UnregisterEndpoint(chanCloser.Name())
3✔
3902

3✔
3903
                return r.RegisterEndpoint(&chanCloser)
3✔
3904
        })
3✔
3905
        if err != nil {
3✔
3906
                chanCloser.Stop()
×
3907

×
3908
                return nil, fmt.Errorf("unable to register endpoint for co-op "+
×
3909
                        "close: %w", err)
×
3910
        }
×
3911

3912
        p.activeChanCloses.Store(chanID, makeRbfCloser(&chanCloser))
3✔
3913

3✔
3914
        // Now that we've created the rbf closer state machine, we'll launch a
3✔
3915
        // new goroutine to eventually send in the ChannelFlushed event once
3✔
3916
        // needed.
3✔
3917
        p.cg.WgAdd(1)
3✔
3918
        go p.chanFlushEventSentinel(&chanCloser, link, channel)
3✔
3919

3✔
3920
        return &chanCloser, nil
3✔
3921
}
3922

3923
// shutdownInit describes the two ways we can initiate a new shutdown. Either we
3924
// got an RPC request to do so (left), or we sent a shutdown message to the
3925
// party (for w/e reason), but crashed before the close was complete.
3926
//
3927
//nolint:ll
3928
type shutdownInit = fn.Option[fn.Either[*htlcswitch.ChanClose, channeldb.ShutdownInfo]]
3929

3930
// shutdownStartFeeRate returns the fee rate that should be used for the
3931
// shutdown.  This returns a doubly wrapped option as the shutdown info might
3932
// be none, and the fee rate is only defined for the user initiated shutdown.
3933
func shutdownStartFeeRate(s shutdownInit) fn.Option[chainfee.SatPerKWeight] {
3✔
3934
        feeRateOpt := fn.MapOption(func(init fn.Either[*htlcswitch.ChanClose,
3✔
3935
                channeldb.ShutdownInfo]) fn.Option[chainfee.SatPerKWeight] {
6✔
3936

3✔
3937
                var feeRate fn.Option[chainfee.SatPerKWeight]
3✔
3938
                init.WhenLeft(func(req *htlcswitch.ChanClose) {
6✔
3939
                        feeRate = fn.Some(req.TargetFeePerKw)
3✔
3940
                })
3✔
3941

3942
                return feeRate
3✔
3943
        })(s)
3944

3945
        return fn.FlattenOption(feeRateOpt)
3✔
3946
}
3947

3948
// shutdownStartAddr returns the delivery address that should be used when
3949
// restarting the shutdown process.  If we didn't send a shutdown before we
3950
// restarted, and the user didn't initiate one either, then None is returned.
3951
func shutdownStartAddr(s shutdownInit) fn.Option[lnwire.DeliveryAddress] {
3✔
3952
        addrOpt := fn.MapOption(func(init fn.Either[*htlcswitch.ChanClose,
3✔
3953
                channeldb.ShutdownInfo]) fn.Option[lnwire.DeliveryAddress] {
6✔
3954

3✔
3955
                var addr fn.Option[lnwire.DeliveryAddress]
3✔
3956
                init.WhenLeft(func(req *htlcswitch.ChanClose) {
6✔
3957
                        if len(req.DeliveryScript) != 0 {
6✔
3958
                                addr = fn.Some(req.DeliveryScript)
3✔
3959
                        }
3✔
3960
                })
3961
                init.WhenRight(func(info channeldb.ShutdownInfo) {
6✔
3962
                        addr = fn.Some(info.DeliveryScript.Val)
3✔
3963
                })
3✔
3964

3965
                return addr
3✔
3966
        })(s)
3967

3968
        return fn.FlattenOption(addrOpt)
3✔
3969
}
3970

3971
// whenRPCShutdown registers a callback to be executed when the shutdown init
3972
// type is and RPC request.
3973
func whenRPCShutdown(s shutdownInit, f func(r *htlcswitch.ChanClose)) {
3✔
3974
        s.WhenSome(func(init fn.Either[*htlcswitch.ChanClose,
3✔
3975
                channeldb.ShutdownInfo]) {
6✔
3976

3✔
3977
                init.WhenLeft(f)
3✔
3978
        })
3✔
3979
}
3980

3981
// newRestartShutdownInit creates a new shutdownInit for the case where we need
3982
// to restart the shutdown flow after a restart.
3983
func newRestartShutdownInit(info channeldb.ShutdownInfo) shutdownInit {
3✔
3984
        return fn.Some(fn.NewRight[*htlcswitch.ChanClose](info))
3✔
3985
}
3✔
3986

3987
// newRPCShutdownInit creates a new shutdownInit for the case where we
3988
// initiated the shutdown via an RPC client.
3989
func newRPCShutdownInit(req *htlcswitch.ChanClose) shutdownInit {
3✔
3990
        return fn.Some(
3✔
3991
                fn.NewLeft[*htlcswitch.ChanClose, channeldb.ShutdownInfo](req),
3✔
3992
        )
3✔
3993
}
3✔
3994

3995
// waitUntilRbfCoastClear waits until the RBF co-op close state machine has
3996
// advanced to a terminal state before attempting another fee bump.
3997
func waitUntilRbfCoastClear(ctx context.Context,
3998
        rbfCloser *chancloser.RbfChanCloser) error {
3✔
3999

3✔
4000
        coopCloseStates := rbfCloser.RegisterStateEvents()
3✔
4001
        newStateChan := coopCloseStates.NewItemCreated.ChanOut()
3✔
4002
        defer rbfCloser.RemoveStateSub(coopCloseStates)
3✔
4003

3✔
4004
        isTerminalState := func(newState chancloser.RbfState) bool {
6✔
4005
                // If we're not in the negotiation sub-state, then we aren't at
3✔
4006
                // the terminal state yet.
3✔
4007
                state, ok := newState.(*chancloser.ClosingNegotiation)
3✔
4008
                if !ok {
3✔
4009
                        return false
×
4010
                }
×
4011

4012
                localState := state.PeerState.GetForParty(lntypes.Local)
3✔
4013

3✔
4014
                // If this isn't the close pending state, we aren't at the
3✔
4015
                // terminal state yet.
3✔
4016
                _, ok = localState.(*chancloser.ClosePending)
3✔
4017

3✔
4018
                return ok
3✔
4019
        }
4020

4021
        // Before we enter the subscription loop below, check to see if we're
4022
        // already in the terminal state.
4023
        rbfState, err := rbfCloser.CurrentState()
3✔
4024
        if err != nil {
3✔
4025
                return err
×
4026
        }
×
4027
        if isTerminalState(rbfState) {
6✔
4028
                return nil
3✔
4029
        }
3✔
4030

4031
        peerLog.Debugf("Waiting for RBF iteration to complete...")
×
4032

×
4033
        for {
×
4034
                select {
×
4035
                case newState := <-newStateChan:
×
4036
                        if isTerminalState(newState) {
×
4037
                                return nil
×
4038
                        }
×
4039

4040
                case <-ctx.Done():
×
4041
                        return fmt.Errorf("context canceled")
×
4042
                }
4043
        }
4044
}
4045

4046
// startRbfChanCloser kicks off the co-op close process using the new RBF based
4047
// co-op close protocol. This is called when we're the one that's initiating
4048
// the cooperative channel close.
4049
//
4050
// TODO(roasbeef): just accept the two shutdown pointer params instead??
4051
func (p *Brontide) startRbfChanCloser(shutdown shutdownInit,
4052
        chanPoint wire.OutPoint) error {
3✔
4053

3✔
4054
        // Unlike the old negotiate chan closer, we'll always create the RBF
3✔
4055
        // chan closer on startup, so we can skip init here.
3✔
4056
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
3✔
4057
        chanCloser, found := p.activeChanCloses.Load(chanID)
3✔
4058
        if !found {
3✔
4059
                return fmt.Errorf("rbf chan closer not found for channel %v",
×
4060
                        chanPoint)
×
4061
        }
×
4062

4063
        defaultFeePerKw, err := shutdownStartFeeRate(
3✔
4064
                shutdown,
3✔
4065
        ).UnwrapOrFuncErr(func() (chainfee.SatPerKWeight, error) {
6✔
4066
                return p.cfg.FeeEstimator.EstimateFeePerKW(
3✔
4067
                        p.cfg.CoopCloseTargetConfs,
3✔
4068
                )
3✔
4069
        })
3✔
4070
        if err != nil {
3✔
4071
                return fmt.Errorf("unable to estimate fee: %w", err)
×
4072
        }
×
4073

4074
        chanCloser.WhenRight(func(rbfCloser *chancloser.RbfChanCloser) {
6✔
4075
                peerLog.Infof("ChannelPoint(%v): rbf-coop close requested, "+
3✔
4076
                        "sending shutdown", chanPoint)
3✔
4077

3✔
4078
                rbfState, err := rbfCloser.CurrentState()
3✔
4079
                if err != nil {
3✔
4080
                        peerLog.Warnf("ChannelPoint(%v): unable to get "+
×
4081
                                "current state for rbf-coop close: %v",
×
4082
                                chanPoint, err)
×
4083

×
4084
                        return
×
4085
                }
×
4086

4087
                coopCloseStates := rbfCloser.RegisterStateEvents()
3✔
4088

3✔
4089
                // Before we send our event below, we'll launch a goroutine to
3✔
4090
                // watch for the final terminal state to send updates to the RPC
3✔
4091
                // client. We only need to do this if there's an RPC caller.
3✔
4092
                var rpcShutdown bool
3✔
4093
                whenRPCShutdown(shutdown, func(req *htlcswitch.ChanClose) {
6✔
4094
                        rpcShutdown = true
3✔
4095

3✔
4096
                        p.cg.WgAdd(1)
3✔
4097
                        go func() {
6✔
4098
                                defer p.cg.WgDone()
3✔
4099

3✔
4100
                                p.observeRbfCloseUpdates(
3✔
4101
                                        rbfCloser, req, coopCloseStates,
3✔
4102
                                )
3✔
4103
                        }()
3✔
4104
                })
4105

4106
                if !rpcShutdown {
6✔
4107
                        defer rbfCloser.RemoveStateSub(coopCloseStates)
3✔
4108
                }
3✔
4109

4110
                ctx, _ := p.cg.Create(context.Background())
3✔
4111
                feeRate := defaultFeePerKw.FeePerVByte()
3✔
4112

3✔
4113
                // Depending on the state of the state machine, we'll either
3✔
4114
                // kick things off by sending shutdown, or attempt to send a new
3✔
4115
                // offer to the remote party.
3✔
4116
                switch rbfState.(type) {
3✔
4117
                // The channel is still active, so we'll now kick off the co-op
4118
                // close process by instructing it to send a shutdown message to
4119
                // the remote party.
4120
                case *chancloser.ChannelActive:
3✔
4121
                        rbfCloser.SendEvent(
3✔
4122
                                context.Background(),
3✔
4123
                                &chancloser.SendShutdown{
3✔
4124
                                        IdealFeeRate: feeRate,
3✔
4125
                                        DeliveryAddr: shutdownStartAddr(
3✔
4126
                                                shutdown,
3✔
4127
                                        ),
3✔
4128
                                },
3✔
4129
                        )
3✔
4130

4131
                // If we haven't yet sent an offer (didn't have enough funds at
4132
                // the prior fee rate), or we've sent an offer, then we'll
4133
                // trigger a new offer event.
4134
                case *chancloser.ClosingNegotiation:
3✔
4135
                        // Before we send the event below, we'll wait until
3✔
4136
                        // we're in a semi-terminal state.
3✔
4137
                        err := waitUntilRbfCoastClear(ctx, rbfCloser)
3✔
4138
                        if err != nil {
3✔
4139
                                peerLog.Warnf("ChannelPoint(%v): unable to "+
×
4140
                                        "wait for coast to clear: %v",
×
4141
                                        chanPoint, err)
×
4142

×
4143
                                return
×
4144
                        }
×
4145

4146
                        event := chancloser.ProtocolEvent(
3✔
4147
                                &chancloser.SendOfferEvent{
3✔
4148
                                        TargetFeeRate: feeRate,
3✔
4149
                                },
3✔
4150
                        )
3✔
4151
                        rbfCloser.SendEvent(ctx, event)
3✔
4152

4153
                default:
×
4154
                        peerLog.Warnf("ChannelPoint(%v): unexpected state "+
×
4155
                                "for rbf-coop close: %T", chanPoint, rbfState)
×
4156
                }
4157
        })
4158

4159
        return nil
3✔
4160
}
4161

4162
// handleLocalCloseReq kicks-off the workflow to execute a cooperative or
4163
// forced unilateral closure of the channel initiated by a local subsystem.
4164
func (p *Brontide) handleLocalCloseReq(req *htlcswitch.ChanClose) {
10✔
4165
        chanID := lnwire.NewChanIDFromOutPoint(*req.ChanPoint)
10✔
4166

10✔
4167
        channel, ok := p.activeChannels.Load(chanID)
10✔
4168

10✔
4169
        // Though this function can't be called for pending channels, we still
10✔
4170
        // check whether channel is nil for safety.
10✔
4171
        if !ok || channel == nil {
10✔
4172
                err := fmt.Errorf("unable to close channel, ChannelID(%v) is "+
×
4173
                        "unknown", chanID)
×
4174
                p.log.Errorf(err.Error())
×
4175
                req.Err <- err
×
4176
                return
×
4177
        }
×
4178

4179
        isTaprootChan := channel.ChanType().IsTaproot()
10✔
4180

10✔
4181
        switch req.CloseType {
10✔
4182
        // A type of CloseRegular indicates that the user has opted to close
4183
        // out this channel on-chain, so we execute the cooperative channel
4184
        // closure workflow.
4185
        case contractcourt.CloseRegular:
10✔
4186
                var err error
10✔
4187
                switch {
10✔
4188
                // If this is the RBF coop state machine, then we'll instruct
4189
                // it to send the shutdown message. This also might be an RBF
4190
                // iteration, in which case we'll be obtaining a new
4191
                // transaction w/ a higher fee rate.
4192
                //
4193
                // We don't support this close type for taproot channels yet
4194
                // however.
4195
                case !isTaprootChan && p.rbfCoopCloseAllowed():
3✔
4196
                        err = p.startRbfChanCloser(
3✔
4197
                                newRPCShutdownInit(req), channel.ChannelPoint(),
3✔
4198
                        )
3✔
4199
                default:
10✔
4200
                        err = p.initNegotiateChanCloser(req, channel)
10✔
4201
                }
4202

4203
                if err != nil {
11✔
4204
                        p.log.Errorf(err.Error())
1✔
4205
                        req.Err <- err
1✔
4206
                }
1✔
4207

4208
        // A type of CloseBreach indicates that the counterparty has breached
4209
        // the channel therefore we need to clean up our local state.
4210
        case contractcourt.CloseBreach:
1✔
4211
                // TODO(roasbeef): no longer need with newer beach logic?
1✔
4212
                p.log.Infof("ChannelPoint(%v) has been breached, wiping "+
1✔
4213
                        "channel", req.ChanPoint)
1✔
4214
                p.WipeChannel(req.ChanPoint)
1✔
4215
        }
4216
}
4217

4218
// linkFailureReport is sent to the channelManager whenever a link reports a
4219
// link failure, and is forced to exit. The report houses the necessary
4220
// information to clean up the channel state, send back the error message, and
4221
// force close if necessary.
4222
type linkFailureReport struct {
4223
        chanPoint   wire.OutPoint
4224
        chanID      lnwire.ChannelID
4225
        shortChanID lnwire.ShortChannelID
4226
        linkErr     htlcswitch.LinkFailureError
4227
}
4228

4229
// handleLinkFailure processes a link failure report when a link in the switch
4230
// fails. It facilitates the removal of all channel state within the peer,
4231
// force closing the channel depending on severity, and sending the error
4232
// message back to the remote party.
4233
func (p *Brontide) handleLinkFailure(failure linkFailureReport) {
3✔
4234
        // Retrieve the channel from the map of active channels. We do this to
3✔
4235
        // have access to it even after WipeChannel remove it from the map.
3✔
4236
        chanID := lnwire.NewChanIDFromOutPoint(failure.chanPoint)
3✔
4237
        lnChan, _ := p.activeChannels.Load(chanID)
3✔
4238

3✔
4239
        // We begin by wiping the link, which will remove it from the switch,
3✔
4240
        // such that it won't be attempted used for any more updates.
3✔
4241
        //
3✔
4242
        // TODO(halseth): should introduce a way to atomically stop/pause the
3✔
4243
        // link and cancel back any adds in its mailboxes such that we can
3✔
4244
        // safely force close without the link being added again and updates
3✔
4245
        // being applied.
3✔
4246
        p.WipeChannel(&failure.chanPoint)
3✔
4247

3✔
4248
        // If the error encountered was severe enough, we'll now force close
3✔
4249
        // the channel to prevent reading it to the switch in the future.
3✔
4250
        if failure.linkErr.FailureAction == htlcswitch.LinkFailureForceClose {
6✔
4251
                p.log.Warnf("Force closing link(%v)", failure.shortChanID)
3✔
4252

3✔
4253
                closeTx, err := p.cfg.ChainArb.ForceCloseContract(
3✔
4254
                        failure.chanPoint,
3✔
4255
                )
3✔
4256
                if err != nil {
6✔
4257
                        p.log.Errorf("unable to force close "+
3✔
4258
                                "link(%v): %v", failure.shortChanID, err)
3✔
4259
                } else {
6✔
4260
                        p.log.Infof("channel(%v) force "+
3✔
4261
                                "closed with txid %v",
3✔
4262
                                failure.shortChanID, closeTx.TxHash())
3✔
4263
                }
3✔
4264
        }
4265

4266
        // If this is a permanent failure, we will mark the channel borked.
4267
        if failure.linkErr.PermanentFailure && lnChan != nil {
3✔
4268
                p.log.Warnf("Marking link(%v) borked due to permanent "+
×
4269
                        "failure", failure.shortChanID)
×
4270

×
4271
                if err := lnChan.State().MarkBorked(); err != nil {
×
4272
                        p.log.Errorf("Unable to mark channel %v borked: %v",
×
4273
                                failure.shortChanID, err)
×
4274
                }
×
4275
        }
4276

4277
        // Send an error to the peer, why we failed the channel.
4278
        if failure.linkErr.ShouldSendToPeer() {
6✔
4279
                // If SendData is set, send it to the peer. If not, we'll use
3✔
4280
                // the standard error messages in the payload. We only include
3✔
4281
                // sendData in the cases where the error data does not contain
3✔
4282
                // sensitive information.
3✔
4283
                data := []byte(failure.linkErr.Error())
3✔
4284
                if failure.linkErr.SendData != nil {
3✔
4285
                        data = failure.linkErr.SendData
×
4286
                }
×
4287

4288
                var networkMsg lnwire.Message
3✔
4289
                if failure.linkErr.Warning {
3✔
4290
                        networkMsg = &lnwire.Warning{
×
4291
                                ChanID: failure.chanID,
×
4292
                                Data:   data,
×
4293
                        }
×
4294
                } else {
3✔
4295
                        networkMsg = &lnwire.Error{
3✔
4296
                                ChanID: failure.chanID,
3✔
4297
                                Data:   data,
3✔
4298
                        }
3✔
4299
                }
3✔
4300

4301
                err := p.SendMessage(true, networkMsg)
3✔
4302
                if err != nil {
3✔
4303
                        p.log.Errorf("unable to send msg to "+
×
4304
                                "remote peer: %v", err)
×
4305
                }
×
4306
        }
4307

4308
        // If the failure action is disconnect, then we'll execute that now. If
4309
        // we had to send an error above, it was a sync call, so we expect the
4310
        // message to be flushed on the wire by now.
4311
        if failure.linkErr.FailureAction == htlcswitch.LinkFailureDisconnect {
3✔
4312
                p.Disconnect(fmt.Errorf("link requested disconnect"))
×
4313
        }
×
4314
}
4315

4316
// fetchLinkFromKeyAndCid fetches a link from the switch via the remote's
4317
// public key and the channel id.
4318
func (p *Brontide) fetchLinkFromKeyAndCid(
4319
        cid lnwire.ChannelID) htlcswitch.ChannelUpdateHandler {
22✔
4320

22✔
4321
        var chanLink htlcswitch.ChannelUpdateHandler
22✔
4322

22✔
4323
        // We don't need to check the error here, and can instead just loop
22✔
4324
        // over the slice and return nil.
22✔
4325
        links, _ := p.cfg.Switch.GetLinksByInterface(p.cfg.PubKeyBytes)
22✔
4326
        for _, link := range links {
43✔
4327
                if link.ChanID() == cid {
42✔
4328
                        chanLink = link
21✔
4329
                        break
21✔
4330
                }
4331
        }
4332

4333
        return chanLink
22✔
4334
}
4335

4336
// finalizeChanClosure performs the final clean up steps once the cooperative
4337
// closure transaction has been fully broadcast. The finalized closing state
4338
// machine should be passed in. Once the transaction has been sufficiently
4339
// confirmed, the channel will be marked as fully closed within the database,
4340
// and any clients will be notified of updates to the closing state.
4341
func (p *Brontide) finalizeChanClosure(chanCloser *chancloser.ChanCloser) {
7✔
4342
        closeReq := chanCloser.CloseRequest()
7✔
4343

7✔
4344
        // First, we'll clear all indexes related to the channel in question.
7✔
4345
        chanPoint := chanCloser.Channel().ChannelPoint()
7✔
4346
        p.WipeChannel(&chanPoint)
7✔
4347

7✔
4348
        // Also clear the activeChanCloses map of this channel.
7✔
4349
        cid := lnwire.NewChanIDFromOutPoint(chanPoint)
7✔
4350
        p.activeChanCloses.Delete(cid) // TODO(roasbeef): existing race
7✔
4351

7✔
4352
        // Next, we'll launch a goroutine which will request to be notified by
7✔
4353
        // the ChainNotifier once the closure transaction obtains a single
7✔
4354
        // confirmation.
7✔
4355
        notifier := p.cfg.ChainNotifier
7✔
4356

7✔
4357
        // If any error happens during waitForChanToClose, forward it to
7✔
4358
        // closeReq. If this channel closure is not locally initiated, closeReq
7✔
4359
        // will be nil, so just ignore the error.
7✔
4360
        errChan := make(chan error, 1)
7✔
4361
        if closeReq != nil {
12✔
4362
                errChan = closeReq.Err
5✔
4363
        }
5✔
4364

4365
        closingTx, err := chanCloser.ClosingTx()
7✔
4366
        if err != nil {
7✔
4367
                if closeReq != nil {
×
4368
                        p.log.Error(err)
×
4369
                        closeReq.Err <- err
×
4370
                }
×
4371
        }
4372

4373
        closingTxid := closingTx.TxHash()
7✔
4374

7✔
4375
        // If this is a locally requested shutdown, update the caller with a
7✔
4376
        // new event detailing the current pending state of this request.
7✔
4377
        if closeReq != nil {
12✔
4378
                closeReq.Updates <- &PendingUpdate{
5✔
4379
                        Txid: closingTxid[:],
5✔
4380
                }
5✔
4381
        }
5✔
4382

4383
        localOut := chanCloser.LocalCloseOutput()
7✔
4384
        remoteOut := chanCloser.RemoteCloseOutput()
7✔
4385
        auxOut := chanCloser.AuxOutputs()
7✔
4386
        go WaitForChanToClose(
7✔
4387
                chanCloser.NegotiationHeight(), notifier, errChan,
7✔
4388
                &chanPoint, &closingTxid, closingTx.TxOut[0].PkScript, func() {
14✔
4389
                        // Respond to the local subsystem which requested the
7✔
4390
                        // channel closure.
7✔
4391
                        if closeReq != nil {
12✔
4392
                                closeReq.Updates <- &ChannelCloseUpdate{
5✔
4393
                                        ClosingTxid:       closingTxid[:],
5✔
4394
                                        Success:           true,
5✔
4395
                                        LocalCloseOutput:  localOut,
5✔
4396
                                        RemoteCloseOutput: remoteOut,
5✔
4397
                                        AuxOutputs:        auxOut,
5✔
4398
                                }
5✔
4399
                        }
5✔
4400
                },
4401
        )
4402
}
4403

4404
// WaitForChanToClose uses the passed notifier to wait until the channel has
4405
// been detected as closed on chain and then concludes by executing the
4406
// following actions: the channel point will be sent over the settleChan, and
4407
// finally the callback will be executed. If any error is encountered within
4408
// the function, then it will be sent over the errChan.
4409
func WaitForChanToClose(bestHeight uint32, notifier chainntnfs.ChainNotifier,
4410
        errChan chan error, chanPoint *wire.OutPoint,
4411
        closingTxID *chainhash.Hash, closeScript []byte, cb func()) {
7✔
4412

7✔
4413
        peerLog.Infof("Waiting for confirmation of close of ChannelPoint(%v) "+
7✔
4414
                "with txid: %v", chanPoint, closingTxID)
7✔
4415

7✔
4416
        // TODO(roasbeef): add param for num needed confs
7✔
4417
        confNtfn, err := notifier.RegisterConfirmationsNtfn(
7✔
4418
                closingTxID, closeScript, 1, bestHeight,
7✔
4419
        )
7✔
4420
        if err != nil {
7✔
4421
                if errChan != nil {
×
4422
                        errChan <- err
×
4423
                }
×
4424
                return
×
4425
        }
4426

4427
        // In the case that the ChainNotifier is shutting down, all subscriber
4428
        // notification channels will be closed, generating a nil receive.
4429
        height, ok := <-confNtfn.Confirmed
7✔
4430
        if !ok {
10✔
4431
                return
3✔
4432
        }
3✔
4433

4434
        // The channel has been closed, remove it from any active indexes, and
4435
        // the database state.
4436
        peerLog.Infof("ChannelPoint(%v) is now closed at "+
7✔
4437
                "height %v", chanPoint, height.BlockHeight)
7✔
4438

7✔
4439
        // Finally, execute the closure call back to mark the confirmation of
7✔
4440
        // the transaction closing the contract.
7✔
4441
        cb()
7✔
4442
}
4443

4444
// WipeChannel removes the passed channel point from all indexes associated with
4445
// the peer and the switch.
4446
func (p *Brontide) WipeChannel(chanPoint *wire.OutPoint) {
7✔
4447
        chanID := lnwire.NewChanIDFromOutPoint(*chanPoint)
7✔
4448

7✔
4449
        p.activeChannels.Delete(chanID)
7✔
4450

7✔
4451
        // Instruct the HtlcSwitch to close this link as the channel is no
7✔
4452
        // longer active.
7✔
4453
        p.cfg.Switch.RemoveLink(chanID)
7✔
4454
}
7✔
4455

4456
// handleInitMsg handles the incoming init message which contains global and
4457
// local feature vectors. If feature vectors are incompatible then disconnect.
4458
func (p *Brontide) handleInitMsg(msg *lnwire.Init) error {
6✔
4459
        // First, merge any features from the legacy global features field into
6✔
4460
        // those presented in the local features fields.
6✔
4461
        err := msg.Features.Merge(msg.GlobalFeatures)
6✔
4462
        if err != nil {
6✔
4463
                return fmt.Errorf("unable to merge legacy global features: %w",
×
4464
                        err)
×
4465
        }
×
4466

4467
        // Then, finalize the remote feature vector providing the flattened
4468
        // feature bit namespace.
4469
        p.remoteFeatures = lnwire.NewFeatureVector(
6✔
4470
                msg.Features, lnwire.Features,
6✔
4471
        )
6✔
4472

6✔
4473
        // Now that we have their features loaded, we'll ensure that they
6✔
4474
        // didn't set any required bits that we don't know of.
6✔
4475
        err = feature.ValidateRequired(p.remoteFeatures)
6✔
4476
        if err != nil {
6✔
4477
                return fmt.Errorf("invalid remote features: %w", err)
×
4478
        }
×
4479

4480
        // Ensure the remote party's feature vector contains all transitive
4481
        // dependencies. We know ours are correct since they are validated
4482
        // during the feature manager's instantiation.
4483
        err = feature.ValidateDeps(p.remoteFeatures)
6✔
4484
        if err != nil {
6✔
4485
                return fmt.Errorf("invalid remote features: %w", err)
×
4486
        }
×
4487

4488
        // Now that we know we understand their requirements, we'll check to
4489
        // see if they don't support anything that we deem to be mandatory.
4490
        if !p.remoteFeatures.HasFeature(lnwire.DataLossProtectRequired) {
6✔
4491
                return fmt.Errorf("data loss protection required")
×
4492
        }
×
4493

4494
        return nil
6✔
4495
}
4496

4497
// LocalFeatures returns the set of global features that has been advertised by
4498
// the local node. This allows sub-systems that use this interface to gate their
4499
// behavior off the set of negotiated feature bits.
4500
//
4501
// NOTE: Part of the lnpeer.Peer interface.
4502
func (p *Brontide) LocalFeatures() *lnwire.FeatureVector {
3✔
4503
        return p.cfg.Features
3✔
4504
}
3✔
4505

4506
// RemoteFeatures returns the set of global features that has been advertised by
4507
// the remote node. This allows sub-systems that use this interface to gate
4508
// their behavior off the set of negotiated feature bits.
4509
//
4510
// NOTE: Part of the lnpeer.Peer interface.
4511
func (p *Brontide) RemoteFeatures() *lnwire.FeatureVector {
16✔
4512
        return p.remoteFeatures
16✔
4513
}
16✔
4514

4515
// hasNegotiatedScidAlias returns true if we've negotiated the
4516
// option-scid-alias feature bit with the peer.
4517
func (p *Brontide) hasNegotiatedScidAlias() bool {
6✔
4518
        peerHas := p.remoteFeatures.HasFeature(lnwire.ScidAliasOptional)
6✔
4519
        localHas := p.cfg.Features.HasFeature(lnwire.ScidAliasOptional)
6✔
4520
        return peerHas && localHas
6✔
4521
}
6✔
4522

4523
// sendInitMsg sends the Init message to the remote peer. This message contains
4524
// our currently supported local and global features.
4525
func (p *Brontide) sendInitMsg(legacyChan bool) error {
10✔
4526
        features := p.cfg.Features.Clone()
10✔
4527
        legacyFeatures := p.cfg.LegacyFeatures.Clone()
10✔
4528

10✔
4529
        // If we have a legacy channel open with a peer, we downgrade static
10✔
4530
        // remote required to optional in case the peer does not understand the
10✔
4531
        // required feature bit. If we do not do this, the peer will reject our
10✔
4532
        // connection because it does not understand a required feature bit, and
10✔
4533
        // our channel will be unusable.
10✔
4534
        if legacyChan && features.RequiresFeature(lnwire.StaticRemoteKeyRequired) {
11✔
4535
                p.log.Infof("Legacy channel open with peer, " +
1✔
4536
                        "downgrading static remote required feature bit to " +
1✔
4537
                        "optional")
1✔
4538

1✔
4539
                // Unset and set in both the local and global features to
1✔
4540
                // ensure both sets are consistent and merge able by old and
1✔
4541
                // new nodes.
1✔
4542
                features.Unset(lnwire.StaticRemoteKeyRequired)
1✔
4543
                legacyFeatures.Unset(lnwire.StaticRemoteKeyRequired)
1✔
4544

1✔
4545
                features.Set(lnwire.StaticRemoteKeyOptional)
1✔
4546
                legacyFeatures.Set(lnwire.StaticRemoteKeyOptional)
1✔
4547
        }
1✔
4548

4549
        msg := lnwire.NewInitMessage(
10✔
4550
                legacyFeatures.RawFeatureVector,
10✔
4551
                features.RawFeatureVector,
10✔
4552
        )
10✔
4553

10✔
4554
        return p.writeMessage(msg)
10✔
4555
}
4556

4557
// resendChanSyncMsg will attempt to find a channel sync message for the closed
4558
// channel and resend it to our peer.
4559
func (p *Brontide) resendChanSyncMsg(cid lnwire.ChannelID) error {
3✔
4560
        // If we already re-sent the mssage for this channel, we won't do it
3✔
4561
        // again.
3✔
4562
        if _, ok := p.resentChanSyncMsg[cid]; ok {
4✔
4563
                return nil
1✔
4564
        }
1✔
4565

4566
        // Check if we have any channel sync messages stored for this channel.
4567
        c, err := p.cfg.ChannelDB.FetchClosedChannelForID(cid)
3✔
4568
        if err != nil {
6✔
4569
                return fmt.Errorf("unable to fetch channel sync messages for "+
3✔
4570
                        "peer %v: %v", p, err)
3✔
4571
        }
3✔
4572

4573
        if c.LastChanSyncMsg == nil {
3✔
4574
                return fmt.Errorf("no chan sync message stored for channel %v",
×
4575
                        cid)
×
4576
        }
×
4577

4578
        if !c.RemotePub.IsEqual(p.IdentityKey()) {
3✔
4579
                return fmt.Errorf("ignoring channel reestablish from "+
×
4580
                        "peer=%x", p.IdentityKey().SerializeCompressed())
×
4581
        }
×
4582

4583
        p.log.Debugf("Re-sending channel sync message for channel %v to "+
3✔
4584
                "peer", cid)
3✔
4585

3✔
4586
        if err := p.SendMessage(true, c.LastChanSyncMsg); err != nil {
3✔
4587
                return fmt.Errorf("failed resending channel sync "+
×
4588
                        "message to peer %v: %v", p, err)
×
4589
        }
×
4590

4591
        p.log.Debugf("Re-sent channel sync message for channel %v to peer ",
3✔
4592
                cid)
3✔
4593

3✔
4594
        // Note down that we sent the message, so we won't resend it again for
3✔
4595
        // this connection.
3✔
4596
        p.resentChanSyncMsg[cid] = struct{}{}
3✔
4597

3✔
4598
        return nil
3✔
4599
}
4600

4601
// SendMessage sends a variadic number of high-priority messages to the remote
4602
// peer. The first argument denotes if the method should block until the
4603
// messages have been sent to the remote peer or an error is returned,
4604
// otherwise it returns immediately after queuing.
4605
//
4606
// NOTE: Part of the lnpeer.Peer interface.
4607
func (p *Brontide) SendMessage(sync bool, msgs ...lnwire.Message) error {
6✔
4608
        return p.sendMessage(sync, true, msgs...)
6✔
4609
}
6✔
4610

4611
// SendMessageLazy sends a variadic number of low-priority messages to the
4612
// remote peer. The first argument denotes if the method should block until
4613
// the messages have been sent to the remote peer or an error is returned,
4614
// otherwise it returns immediately after queueing.
4615
//
4616
// NOTE: Part of the lnpeer.Peer interface.
4617
func (p *Brontide) SendMessageLazy(sync bool, msgs ...lnwire.Message) error {
4✔
4618
        return p.sendMessage(sync, false, msgs...)
4✔
4619
}
4✔
4620

4621
// sendMessage queues a variadic number of messages using the passed priority
4622
// to the remote peer. If sync is true, this method will block until the
4623
// messages have been sent to the remote peer or an error is returned, otherwise
4624
// it returns immediately after queueing.
4625
func (p *Brontide) sendMessage(sync, priority bool, msgs ...lnwire.Message) error {
7✔
4626
        // Add all incoming messages to the outgoing queue. A list of error
7✔
4627
        // chans is populated for each message if the caller requested a sync
7✔
4628
        // send.
7✔
4629
        var errChans []chan error
7✔
4630
        if sync {
11✔
4631
                errChans = make([]chan error, 0, len(msgs))
4✔
4632
        }
4✔
4633
        for _, msg := range msgs {
14✔
4634
                // If a sync send was requested, create an error chan to listen
7✔
4635
                // for an ack from the writeHandler.
7✔
4636
                var errChan chan error
7✔
4637
                if sync {
11✔
4638
                        errChan = make(chan error, 1)
4✔
4639
                        errChans = append(errChans, errChan)
4✔
4640
                }
4✔
4641

4642
                if priority {
13✔
4643
                        p.queueMsg(msg, errChan)
6✔
4644
                } else {
10✔
4645
                        p.queueMsgLazy(msg, errChan)
4✔
4646
                }
4✔
4647
        }
4648

4649
        // Wait for all replies from the writeHandler. For async sends, this
4650
        // will be a NOP as the list of error chans is nil.
4651
        for _, errChan := range errChans {
11✔
4652
                select {
4✔
4653
                case err := <-errChan:
4✔
4654
                        return err
4✔
4655
                case <-p.cg.Done():
×
4656
                        return lnpeer.ErrPeerExiting
×
4657
                case <-p.cfg.Quit:
×
4658
                        return lnpeer.ErrPeerExiting
×
4659
                }
4660
        }
4661

4662
        return nil
6✔
4663
}
4664

4665
// PubKey returns the pubkey of the peer in compressed serialized format.
4666
//
4667
// NOTE: Part of the lnpeer.Peer interface.
4668
func (p *Brontide) PubKey() [33]byte {
5✔
4669
        return p.cfg.PubKeyBytes
5✔
4670
}
5✔
4671

4672
// IdentityKey returns the public key of the remote peer.
4673
//
4674
// NOTE: Part of the lnpeer.Peer interface.
4675
func (p *Brontide) IdentityKey() *btcec.PublicKey {
18✔
4676
        return p.cfg.Addr.IdentityKey
18✔
4677
}
18✔
4678

4679
// Address returns the network address of the remote peer.
4680
//
4681
// NOTE: Part of the lnpeer.Peer interface.
4682
func (p *Brontide) Address() net.Addr {
3✔
4683
        return p.cfg.Addr.Address
3✔
4684
}
3✔
4685

4686
// AddNewChannel adds a new channel to the peer. The channel should fail to be
4687
// added if the cancel channel is closed.
4688
//
4689
// NOTE: Part of the lnpeer.Peer interface.
4690
func (p *Brontide) AddNewChannel(newChan *lnpeer.NewChannel,
4691
        cancel <-chan struct{}) error {
3✔
4692

3✔
4693
        errChan := make(chan error, 1)
3✔
4694
        newChanMsg := &newChannelMsg{
3✔
4695
                channel: newChan,
3✔
4696
                err:     errChan,
3✔
4697
        }
3✔
4698

3✔
4699
        select {
3✔
4700
        case p.newActiveChannel <- newChanMsg:
3✔
4701
        case <-cancel:
×
4702
                return errors.New("canceled adding new channel")
×
4703
        case <-p.cg.Done():
×
4704
                return lnpeer.ErrPeerExiting
×
4705
        }
4706

4707
        // We pause here to wait for the peer to recognize the new channel
4708
        // before we close the channel barrier corresponding to the channel.
4709
        select {
3✔
4710
        case err := <-errChan:
3✔
4711
                return err
3✔
4712
        case <-p.cg.Done():
×
4713
                return lnpeer.ErrPeerExiting
×
4714
        }
4715
}
4716

4717
// AddPendingChannel adds a pending open channel to the peer. The channel
4718
// should fail to be added if the cancel channel is closed.
4719
//
4720
// NOTE: Part of the lnpeer.Peer interface.
4721
func (p *Brontide) AddPendingChannel(cid lnwire.ChannelID,
4722
        cancel <-chan struct{}) error {
3✔
4723

3✔
4724
        errChan := make(chan error, 1)
3✔
4725
        newChanMsg := &newChannelMsg{
3✔
4726
                channelID: cid,
3✔
4727
                err:       errChan,
3✔
4728
        }
3✔
4729

3✔
4730
        select {
3✔
4731
        case p.newPendingChannel <- newChanMsg:
3✔
4732

4733
        case <-cancel:
×
4734
                return errors.New("canceled adding pending channel")
×
4735

4736
        case <-p.cg.Done():
×
4737
                return lnpeer.ErrPeerExiting
×
4738
        }
4739

4740
        // We pause here to wait for the peer to recognize the new pending
4741
        // channel before we close the channel barrier corresponding to the
4742
        // channel.
4743
        select {
3✔
4744
        case err := <-errChan:
3✔
4745
                return err
3✔
4746

4747
        case <-cancel:
×
4748
                return errors.New("canceled adding pending channel")
×
4749

4750
        case <-p.cg.Done():
×
4751
                return lnpeer.ErrPeerExiting
×
4752
        }
4753
}
4754

4755
// RemovePendingChannel removes a pending open channel from the peer.
4756
//
4757
// NOTE: Part of the lnpeer.Peer interface.
4758
func (p *Brontide) RemovePendingChannel(cid lnwire.ChannelID) error {
3✔
4759
        errChan := make(chan error, 1)
3✔
4760
        newChanMsg := &newChannelMsg{
3✔
4761
                channelID: cid,
3✔
4762
                err:       errChan,
3✔
4763
        }
3✔
4764

3✔
4765
        select {
3✔
4766
        case p.removePendingChannel <- newChanMsg:
3✔
4767
        case <-p.cg.Done():
×
4768
                return lnpeer.ErrPeerExiting
×
4769
        }
4770

4771
        // We pause here to wait for the peer to respond to the cancellation of
4772
        // the pending channel before we close the channel barrier
4773
        // corresponding to the channel.
4774
        select {
3✔
4775
        case err := <-errChan:
3✔
4776
                return err
3✔
4777

4778
        case <-p.cg.Done():
×
4779
                return lnpeer.ErrPeerExiting
×
4780
        }
4781
}
4782

4783
// StartTime returns the time at which the connection was established if the
4784
// peer started successfully, and zero otherwise.
4785
func (p *Brontide) StartTime() time.Time {
3✔
4786
        return p.startTime
3✔
4787
}
3✔
4788

4789
// handleCloseMsg is called when a new cooperative channel closure related
4790
// message is received from the remote peer. We'll use this message to advance
4791
// the chan closer state machine.
4792
func (p *Brontide) handleCloseMsg(msg *closeMsg) {
16✔
4793
        link := p.fetchLinkFromKeyAndCid(msg.cid)
16✔
4794

16✔
4795
        // We'll now fetch the matching closing state machine in order to
16✔
4796
        // continue, or finalize the channel closure process.
16✔
4797
        chanCloserE, err := p.fetchActiveChanCloser(msg.cid)
16✔
4798
        if err != nil {
19✔
4799
                // If the channel is not known to us, we'll simply ignore this
3✔
4800
                // message.
3✔
4801
                if err == ErrChannelNotFound {
6✔
4802
                        return
3✔
4803
                }
3✔
4804

4805
                p.log.Errorf("Unable to respond to remote close msg: %v", err)
×
4806

×
4807
                errMsg := &lnwire.Error{
×
4808
                        ChanID: msg.cid,
×
4809
                        Data:   lnwire.ErrorData(err.Error()),
×
4810
                }
×
4811
                p.queueMsg(errMsg, nil)
×
4812
                return
×
4813
        }
4814

4815
        if chanCloserE.IsRight() {
16✔
4816
                // TODO(roasbeef): assert?
×
4817
                return
×
4818
        }
×
4819

4820
        // At this point, we'll only enter this call path if a negotiate chan
4821
        // closer was used. So we'll extract that from the either now.
4822
        //
4823
        // TODO(roabeef): need extra helper func for either to make cleaner
4824
        var chanCloser *chancloser.ChanCloser
16✔
4825
        chanCloserE.WhenLeft(func(c *chancloser.ChanCloser) {
32✔
4826
                chanCloser = c
16✔
4827
        })
16✔
4828

4829
        handleErr := func(err error) {
17✔
4830
                err = fmt.Errorf("unable to process close msg: %w", err)
1✔
4831
                p.log.Error(err)
1✔
4832

1✔
4833
                // As the negotiations failed, we'll reset the channel state
1✔
4834
                // machine to ensure we act to on-chain events as normal.
1✔
4835
                chanCloser.Channel().ResetState()
1✔
4836
                if chanCloser.CloseRequest() != nil {
1✔
4837
                        chanCloser.CloseRequest().Err <- err
×
4838
                }
×
4839

4840
                p.activeChanCloses.Delete(msg.cid)
1✔
4841

1✔
4842
                p.Disconnect(err)
1✔
4843
        }
4844

4845
        // Next, we'll process the next message using the target state machine.
4846
        // We'll either continue negotiation, or halt.
4847
        switch typed := msg.msg.(type) {
16✔
4848
        case *lnwire.Shutdown:
8✔
4849
                // Disable incoming adds immediately.
8✔
4850
                if link != nil && !link.DisableAdds(htlcswitch.Incoming) {
8✔
4851
                        p.log.Warnf("Incoming link adds already disabled: %v",
×
4852
                                link.ChanID())
×
4853
                }
×
4854

4855
                oShutdown, err := chanCloser.ReceiveShutdown(*typed)
8✔
4856
                if err != nil {
8✔
4857
                        handleErr(err)
×
4858
                        return
×
4859
                }
×
4860

4861
                oShutdown.WhenSome(func(msg lnwire.Shutdown) {
14✔
4862
                        // If the link is nil it means we can immediately queue
6✔
4863
                        // the Shutdown message since we don't have to wait for
6✔
4864
                        // commitment transaction synchronization.
6✔
4865
                        if link == nil {
7✔
4866
                                p.queueMsg(&msg, nil)
1✔
4867
                                return
1✔
4868
                        }
1✔
4869

4870
                        // Immediately disallow any new HTLC's from being added
4871
                        // in the outgoing direction.
4872
                        if !link.DisableAdds(htlcswitch.Outgoing) {
5✔
4873
                                p.log.Warnf("Outgoing link adds already "+
×
4874
                                        "disabled: %v", link.ChanID())
×
4875
                        }
×
4876

4877
                        // When we have a Shutdown to send, we defer it till the
4878
                        // next time we send a CommitSig to remain spec
4879
                        // compliant.
4880
                        link.OnCommitOnce(htlcswitch.Outgoing, func() {
10✔
4881
                                p.queueMsg(&msg, nil)
5✔
4882
                        })
5✔
4883
                })
4884

4885
                beginNegotiation := func() {
16✔
4886
                        oClosingSigned, err := chanCloser.BeginNegotiation()
8✔
4887
                        if err != nil {
8✔
4888
                                handleErr(err)
×
4889
                                return
×
4890
                        }
×
4891

4892
                        oClosingSigned.WhenSome(func(msg lnwire.ClosingSigned) {
16✔
4893
                                p.queueMsg(&msg, nil)
8✔
4894
                        })
8✔
4895
                }
4896

4897
                if link == nil {
9✔
4898
                        beginNegotiation()
1✔
4899
                } else {
8✔
4900
                        // Now we register a flush hook to advance the
7✔
4901
                        // ChanCloser and possibly send out a ClosingSigned
7✔
4902
                        // when the link finishes draining.
7✔
4903
                        link.OnFlushedOnce(func() {
14✔
4904
                                // Remove link in goroutine to prevent deadlock.
7✔
4905
                                go p.cfg.Switch.RemoveLink(msg.cid)
7✔
4906
                                beginNegotiation()
7✔
4907
                        })
7✔
4908
                }
4909

4910
        case *lnwire.ClosingSigned:
11✔
4911
                oClosingSigned, err := chanCloser.ReceiveClosingSigned(*typed)
11✔
4912
                if err != nil {
12✔
4913
                        handleErr(err)
1✔
4914
                        return
1✔
4915
                }
1✔
4916

4917
                oClosingSigned.WhenSome(func(msg lnwire.ClosingSigned) {
22✔
4918
                        p.queueMsg(&msg, nil)
11✔
4919
                })
11✔
4920

4921
        default:
×
4922
                panic("impossible closeMsg type")
×
4923
        }
4924

4925
        // If we haven't finished close negotiations, then we'll continue as we
4926
        // can't yet finalize the closure.
4927
        if _, err := chanCloser.ClosingTx(); err != nil {
26✔
4928
                return
11✔
4929
        }
11✔
4930

4931
        // Otherwise, we've agreed on a closing fee! In this case, we'll wrap up
4932
        // the channel closure by notifying relevant sub-systems and launching a
4933
        // goroutine to wait for close tx conf.
4934
        p.finalizeChanClosure(chanCloser)
7✔
4935
}
4936

4937
// HandleLocalCloseChanReqs accepts a *htlcswitch.ChanClose and passes it onto
4938
// the channelManager goroutine, which will shut down the link and possibly
4939
// close the channel.
4940
func (p *Brontide) HandleLocalCloseChanReqs(req *htlcswitch.ChanClose) {
3✔
4941
        select {
3✔
4942
        case p.localCloseChanReqs <- req:
3✔
4943
                p.log.Info("Local close channel request is going to be " +
3✔
4944
                        "delivered to the peer")
3✔
4945
        case <-p.cg.Done():
×
4946
                p.log.Info("Unable to deliver local close channel request " +
×
4947
                        "to peer")
×
4948
        }
4949
}
4950

4951
// NetAddress returns the network of the remote peer as an lnwire.NetAddress.
4952
func (p *Brontide) NetAddress() *lnwire.NetAddress {
3✔
4953
        return p.cfg.Addr
3✔
4954
}
3✔
4955

4956
// Inbound is a getter for the Brontide's Inbound boolean in cfg.
4957
func (p *Brontide) Inbound() bool {
3✔
4958
        return p.cfg.Inbound
3✔
4959
}
3✔
4960

4961
// ConnReq is a getter for the Brontide's connReq in cfg.
4962
func (p *Brontide) ConnReq() *connmgr.ConnReq {
3✔
4963
        return p.cfg.ConnReq
3✔
4964
}
3✔
4965

4966
// ErrorBuffer is a getter for the Brontide's errorBuffer in cfg.
4967
func (p *Brontide) ErrorBuffer() *queue.CircularBuffer {
3✔
4968
        return p.cfg.ErrorBuffer
3✔
4969
}
3✔
4970

4971
// SetAddress sets the remote peer's address given an address.
4972
func (p *Brontide) SetAddress(address net.Addr) {
×
4973
        p.cfg.Addr.Address = address
×
4974
}
×
4975

4976
// ActiveSignal returns the peer's active signal.
4977
func (p *Brontide) ActiveSignal() chan struct{} {
3✔
4978
        return p.activeSignal
3✔
4979
}
3✔
4980

4981
// Conn returns a pointer to the peer's connection struct.
4982
func (p *Brontide) Conn() net.Conn {
3✔
4983
        return p.cfg.Conn
3✔
4984
}
3✔
4985

4986
// BytesReceived returns the number of bytes received from the peer.
4987
func (p *Brontide) BytesReceived() uint64 {
3✔
4988
        return atomic.LoadUint64(&p.bytesReceived)
3✔
4989
}
3✔
4990

4991
// BytesSent returns the number of bytes sent to the peer.
4992
func (p *Brontide) BytesSent() uint64 {
3✔
4993
        return atomic.LoadUint64(&p.bytesSent)
3✔
4994
}
3✔
4995

4996
// LastRemotePingPayload returns the last payload the remote party sent as part
4997
// of their ping.
4998
func (p *Brontide) LastRemotePingPayload() []byte {
3✔
4999
        pingPayload := p.lastPingPayload.Load()
3✔
5000
        if pingPayload == nil {
6✔
5001
                return []byte{}
3✔
5002
        }
3✔
5003

5004
        pingBytes, ok := pingPayload.(lnwire.PingPayload)
×
5005
        if !ok {
×
5006
                return nil
×
5007
        }
×
5008

5009
        return pingBytes
×
5010
}
5011

5012
// attachChannelEventSubscription creates a channel event subscription and
5013
// attaches to client to Brontide if the reenableTimeout is no greater than 1
5014
// minute.
5015
func (p *Brontide) attachChannelEventSubscription() error {
6✔
5016
        // If the timeout is greater than 1 minute, it's unlikely that the link
6✔
5017
        // hasn't yet finished its reestablishment. Return a nil without
6✔
5018
        // creating the client to specify that we don't want to retry.
6✔
5019
        if p.cfg.ChanActiveTimeout > 1*time.Minute {
9✔
5020
                return nil
3✔
5021
        }
3✔
5022

5023
        // When the reenable timeout is less than 1 minute, it's likely the
5024
        // channel link hasn't finished its reestablishment yet. In that case,
5025
        // we'll give it a second chance by subscribing to the channel update
5026
        // events. Upon receiving the `ActiveLinkEvent`, we'll then request
5027
        // enabling the channel again.
5028
        sub, err := p.cfg.ChannelNotifier.SubscribeChannelEvents()
6✔
5029
        if err != nil {
6✔
5030
                return fmt.Errorf("SubscribeChannelEvents failed: %w", err)
×
5031
        }
×
5032

5033
        p.channelEventClient = sub
6✔
5034

6✔
5035
        return nil
6✔
5036
}
5037

5038
// updateNextRevocation updates the existing channel's next revocation if it's
5039
// nil.
5040
func (p *Brontide) updateNextRevocation(c *channeldb.OpenChannel) error {
6✔
5041
        chanPoint := c.FundingOutpoint
6✔
5042
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
6✔
5043

6✔
5044
        // Read the current channel.
6✔
5045
        currentChan, loaded := p.activeChannels.Load(chanID)
6✔
5046

6✔
5047
        // currentChan should exist, but we perform a check anyway to avoid nil
6✔
5048
        // pointer dereference.
6✔
5049
        if !loaded {
7✔
5050
                return fmt.Errorf("missing active channel with chanID=%v",
1✔
5051
                        chanID)
1✔
5052
        }
1✔
5053

5054
        // currentChan should not be nil, but we perform a check anyway to
5055
        // avoid nil pointer dereference.
5056
        if currentChan == nil {
6✔
5057
                return fmt.Errorf("found nil active channel with chanID=%v",
1✔
5058
                        chanID)
1✔
5059
        }
1✔
5060

5061
        // If we're being sent a new channel, and our existing channel doesn't
5062
        // have the next revocation, then we need to update the current
5063
        // existing channel.
5064
        if currentChan.RemoteNextRevocation() != nil {
4✔
5065
                return nil
×
5066
        }
×
5067

5068
        p.log.Infof("Processing retransmitted ChannelReady for "+
4✔
5069
                "ChannelPoint(%v)", chanPoint)
4✔
5070

4✔
5071
        nextRevoke := c.RemoteNextRevocation
4✔
5072

4✔
5073
        err := currentChan.InitNextRevocation(nextRevoke)
4✔
5074
        if err != nil {
4✔
5075
                return fmt.Errorf("unable to init next revocation: %w", err)
×
5076
        }
×
5077

5078
        return nil
4✔
5079
}
5080

5081
// addActiveChannel adds a new active channel to the `activeChannels` map. It
5082
// takes a `channeldb.OpenChannel`, creates a `lnwallet.LightningChannel` from
5083
// it and assembles it with a channel link.
5084
func (p *Brontide) addActiveChannel(c *lnpeer.NewChannel) error {
3✔
5085
        chanPoint := c.FundingOutpoint
3✔
5086
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
3✔
5087

3✔
5088
        // If we've reached this point, there are two possible scenarios.  If
3✔
5089
        // the channel was in the active channels map as nil, then it was
3✔
5090
        // loaded from disk and we need to send reestablish. Else, it was not
3✔
5091
        // loaded from disk and we don't need to send reestablish as this is a
3✔
5092
        // fresh channel.
3✔
5093
        shouldReestablish := p.isLoadedFromDisk(chanID)
3✔
5094

3✔
5095
        chanOpts := c.ChanOpts
3✔
5096
        if shouldReestablish {
6✔
5097
                // If we have to do the reestablish dance for this channel,
3✔
5098
                // ensure that we don't try to call InitRemoteMusigNonces twice
3✔
5099
                // by calling SkipNonceInit.
3✔
5100
                chanOpts = append(chanOpts, lnwallet.WithSkipNonceInit())
3✔
5101
        }
3✔
5102

5103
        p.cfg.AuxLeafStore.WhenSome(func(s lnwallet.AuxLeafStore) {
3✔
5104
                chanOpts = append(chanOpts, lnwallet.WithLeafStore(s))
×
5105
        })
×
5106
        p.cfg.AuxSigner.WhenSome(func(s lnwallet.AuxSigner) {
3✔
5107
                chanOpts = append(chanOpts, lnwallet.WithAuxSigner(s))
×
5108
        })
×
5109
        p.cfg.AuxResolver.WhenSome(func(s lnwallet.AuxContractResolver) {
3✔
5110
                chanOpts = append(chanOpts, lnwallet.WithAuxResolver(s))
×
5111
        })
×
5112

5113
        // If not already active, we'll add this channel to the set of active
5114
        // channels, so we can look it up later easily according to its channel
5115
        // ID.
5116
        lnChan, err := lnwallet.NewLightningChannel(
3✔
5117
                p.cfg.Signer, c.OpenChannel, p.cfg.SigPool, chanOpts...,
3✔
5118
        )
3✔
5119
        if err != nil {
3✔
5120
                return fmt.Errorf("unable to create LightningChannel: %w", err)
×
5121
        }
×
5122

5123
        // Store the channel in the activeChannels map.
5124
        p.activeChannels.Store(chanID, lnChan)
3✔
5125

3✔
5126
        p.log.Infof("New channel active ChannelPoint(%v) with peer", chanPoint)
3✔
5127

3✔
5128
        // Next, we'll assemble a ChannelLink along with the necessary items it
3✔
5129
        // needs to function.
3✔
5130
        chainEvents, err := p.cfg.ChainArb.SubscribeChannelEvents(chanPoint)
3✔
5131
        if err != nil {
3✔
5132
                return fmt.Errorf("unable to subscribe to chain events: %w",
×
5133
                        err)
×
5134
        }
×
5135

5136
        // We'll query the channel DB for the new channel's initial forwarding
5137
        // policies to determine the policy we start out with.
5138
        initialPolicy, err := p.cfg.ChannelDB.GetInitialForwardingPolicy(chanID)
3✔
5139
        if err != nil {
3✔
5140
                return fmt.Errorf("unable to query for initial forwarding "+
×
5141
                        "policy: %v", err)
×
5142
        }
×
5143

5144
        // Create the link and add it to the switch.
5145
        err = p.addLink(
3✔
5146
                &chanPoint, lnChan, initialPolicy, chainEvents,
3✔
5147
                shouldReestablish, fn.None[lnwire.Shutdown](),
3✔
5148
        )
3✔
5149
        if err != nil {
3✔
5150
                return fmt.Errorf("can't register new channel link(%v) with "+
×
5151
                        "peer", chanPoint)
×
5152
        }
×
5153

5154
        isTaprootChan := c.ChanType.IsTaproot()
3✔
5155

3✔
5156
        // We're using the old co-op close, so we don't need to init the new RBF
3✔
5157
        // chan closer. If this is a taproot channel, then we'll also fall
3✔
5158
        // through, as we don't support this type yet w/ rbf close.
3✔
5159
        if !p.rbfCoopCloseAllowed() || isTaprootChan {
6✔
5160
                return nil
3✔
5161
        }
3✔
5162

5163
        // Now that the link has been added above, we'll also init an RBF chan
5164
        // closer for this channel, but only if the new close feature is
5165
        // negotiated.
5166
        //
5167
        // Creating this here ensures that any shutdown messages sent will be
5168
        // automatically routed by the msg router.
5169
        if _, err := p.initRbfChanCloser(lnChan); err != nil {
3✔
5170
                p.activeChanCloses.Delete(chanID)
×
5171

×
5172
                return fmt.Errorf("unable to init RBF chan closer for new "+
×
5173
                        "chan: %w", err)
×
5174
        }
×
5175

5176
        return nil
3✔
5177
}
5178

5179
// handleNewActiveChannel handles a `newChannelMsg` request. Depending on we
5180
// know this channel ID or not, we'll either add it to the `activeChannels` map
5181
// or init the next revocation for it.
5182
func (p *Brontide) handleNewActiveChannel(req *newChannelMsg) {
3✔
5183
        newChan := req.channel
3✔
5184
        chanPoint := newChan.FundingOutpoint
3✔
5185
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
3✔
5186

3✔
5187
        // Only update RemoteNextRevocation if the channel is in the
3✔
5188
        // activeChannels map and if we added the link to the switch. Only
3✔
5189
        // active channels will be added to the switch.
3✔
5190
        if p.isActiveChannel(chanID) {
6✔
5191
                p.log.Infof("Already have ChannelPoint(%v), ignoring",
3✔
5192
                        chanPoint)
3✔
5193

3✔
5194
                // Handle it and close the err chan on the request.
3✔
5195
                close(req.err)
3✔
5196

3✔
5197
                // Update the next revocation point.
3✔
5198
                err := p.updateNextRevocation(newChan.OpenChannel)
3✔
5199
                if err != nil {
3✔
5200
                        p.log.Errorf(err.Error())
×
5201
                }
×
5202

5203
                return
3✔
5204
        }
5205

5206
        // This is a new channel, we now add it to the map.
5207
        if err := p.addActiveChannel(req.channel); err != nil {
3✔
5208
                // Log and send back the error to the request.
×
5209
                p.log.Errorf(err.Error())
×
5210
                req.err <- err
×
5211

×
5212
                return
×
5213
        }
×
5214

5215
        // Close the err chan if everything went fine.
5216
        close(req.err)
3✔
5217
}
5218

5219
// handleNewPendingChannel takes a `newChannelMsg` request and add it to
5220
// `activeChannels` map with nil value. This pending channel will be saved as
5221
// it may become active in the future. Once active, the funding manager will
5222
// send it again via `AddNewChannel`, and we'd handle the link creation there.
5223
func (p *Brontide) handleNewPendingChannel(req *newChannelMsg) {
7✔
5224
        defer close(req.err)
7✔
5225

7✔
5226
        chanID := req.channelID
7✔
5227

7✔
5228
        // If we already have this channel, something is wrong with the funding
7✔
5229
        // flow as it will only be marked as active after `ChannelReady` is
7✔
5230
        // handled. In this case, we will do nothing but log an error, just in
7✔
5231
        // case this is a legit channel.
7✔
5232
        if p.isActiveChannel(chanID) {
8✔
5233
                p.log.Errorf("Channel(%v) is already active, ignoring "+
1✔
5234
                        "pending channel request", chanID)
1✔
5235

1✔
5236
                return
1✔
5237
        }
1✔
5238

5239
        // The channel has already been added, we will do nothing and return.
5240
        if p.isPendingChannel(chanID) {
7✔
5241
                p.log.Infof("Channel(%v) is already added, ignoring "+
1✔
5242
                        "pending channel request", chanID)
1✔
5243

1✔
5244
                return
1✔
5245
        }
1✔
5246

5247
        // This is a new channel, we now add it to the map `activeChannels`
5248
        // with nil value and mark it as a newly added channel in
5249
        // `addedChannels`.
5250
        p.activeChannels.Store(chanID, nil)
5✔
5251
        p.addedChannels.Store(chanID, struct{}{})
5✔
5252
}
5253

5254
// handleRemovePendingChannel takes a `newChannelMsg` request and removes it
5255
// from `activeChannels` map. The request will be ignored if the channel is
5256
// considered active by Brontide. Noop if the channel ID cannot be found.
5257
func (p *Brontide) handleRemovePendingChannel(req *newChannelMsg) {
7✔
5258
        defer close(req.err)
7✔
5259

7✔
5260
        chanID := req.channelID
7✔
5261

7✔
5262
        // If we already have this channel, something is wrong with the funding
7✔
5263
        // flow as it will only be marked as active after `ChannelReady` is
7✔
5264
        // handled. In this case, we will log an error and exit.
7✔
5265
        if p.isActiveChannel(chanID) {
8✔
5266
                p.log.Errorf("Channel(%v) is active, ignoring remove request",
1✔
5267
                        chanID)
1✔
5268
                return
1✔
5269
        }
1✔
5270

5271
        // The channel has not been added yet, we will log a warning as there
5272
        // is an unexpected call from funding manager.
5273
        if !p.isPendingChannel(chanID) {
10✔
5274
                p.log.Warnf("Channel(%v) not found, removing it anyway", chanID)
4✔
5275
        }
4✔
5276

5277
        // Remove the record of this pending channel.
5278
        p.activeChannels.Delete(chanID)
6✔
5279
        p.addedChannels.Delete(chanID)
6✔
5280
}
5281

5282
// sendLinkUpdateMsg sends a message that updates the channel to the
5283
// channel's message stream.
5284
func (p *Brontide) sendLinkUpdateMsg(cid lnwire.ChannelID, msg lnwire.Message) {
3✔
5285
        p.log.Tracef("Sending link update msg=%v", msg.MsgType())
3✔
5286

3✔
5287
        chanStream, ok := p.activeMsgStreams[cid]
3✔
5288
        if !ok {
6✔
5289
                // If a stream hasn't yet been created, then we'll do so, add
3✔
5290
                // it to the map, and finally start it.
3✔
5291
                chanStream = newChanMsgStream(p, cid)
3✔
5292
                p.activeMsgStreams[cid] = chanStream
3✔
5293
                chanStream.Start()
3✔
5294

3✔
5295
                // Stop the stream when quit.
3✔
5296
                go func() {
6✔
5297
                        <-p.cg.Done()
3✔
5298
                        chanStream.Stop()
3✔
5299
                }()
3✔
5300
        }
5301

5302
        // With the stream obtained, add the message to the stream so we can
5303
        // continue processing message.
5304
        chanStream.AddMsg(msg)
3✔
5305
}
5306

5307
// scaleTimeout multiplies the argument duration by a constant factor depending
5308
// on variious heuristics. Currently this is only used to check whether our peer
5309
// appears to be connected over Tor and relaxes the timout deadline. However,
5310
// this is subject to change and should be treated as opaque.
5311
func (p *Brontide) scaleTimeout(timeout time.Duration) time.Duration {
70✔
5312
        if p.isTorConnection {
73✔
5313
                return timeout * time.Duration(torTimeoutMultiplier)
3✔
5314
        }
3✔
5315

5316
        return timeout
67✔
5317
}
5318

5319
// CoopCloseUpdates is a struct used to communicate updates for an active close
5320
// to the caller.
5321
type CoopCloseUpdates struct {
5322
        UpdateChan chan interface{}
5323

5324
        ErrChan chan error
5325
}
5326

5327
// ChanHasRbfCoopCloser returns true if the channel as identifier by the channel
5328
// point has an active RBF chan closer.
5329
func (p *Brontide) ChanHasRbfCoopCloser(chanPoint wire.OutPoint) bool {
3✔
5330
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
3✔
5331
        chanCloser, found := p.activeChanCloses.Load(chanID)
3✔
5332
        if !found {
6✔
5333
                return false
3✔
5334
        }
3✔
5335

5336
        return chanCloser.IsRight()
3✔
5337
}
5338

5339
// TriggerCoopCloseRbfBump given a chan ID, and the params needed to trigger a
5340
// new RBF co-op close update, a bump is attempted. A channel used for updates,
5341
// along with one used to o=communicate any errors is returned. If no chan
5342
// closer is found, then false is returned for the second argument.
5343
func (p *Brontide) TriggerCoopCloseRbfBump(ctx context.Context,
5344
        chanPoint wire.OutPoint, feeRate chainfee.SatPerKWeight,
5345
        deliveryScript lnwire.DeliveryAddress) (*CoopCloseUpdates, error) {
3✔
5346

3✔
5347
        // If RBF coop close isn't permitted, then we'll an error.
3✔
5348
        if !p.rbfCoopCloseAllowed() {
3✔
5349
                return nil, fmt.Errorf("rbf coop close not enabled for " +
×
5350
                        "channel")
×
5351
        }
×
5352

5353
        closeUpdates := &CoopCloseUpdates{
3✔
5354
                UpdateChan: make(chan interface{}, 1),
3✔
5355
                ErrChan:    make(chan error, 1),
3✔
5356
        }
3✔
5357

3✔
5358
        // We'll re-use the existing switch struct here, even though we're
3✔
5359
        // bypassing the switch entirely.
3✔
5360
        closeReq := htlcswitch.ChanClose{
3✔
5361
                CloseType:      contractcourt.CloseRegular,
3✔
5362
                ChanPoint:      &chanPoint,
3✔
5363
                TargetFeePerKw: feeRate,
3✔
5364
                DeliveryScript: deliveryScript,
3✔
5365
                Updates:        closeUpdates.UpdateChan,
3✔
5366
                Err:            closeUpdates.ErrChan,
3✔
5367
                Ctx:            ctx,
3✔
5368
        }
3✔
5369

3✔
5370
        err := p.startRbfChanCloser(newRPCShutdownInit(&closeReq), chanPoint)
3✔
5371
        if err != nil {
3✔
5372
                return nil, err
×
5373
        }
×
5374

5375
        return closeUpdates, nil
3✔
5376
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc