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

lightningnetwork / lnd / 18197857992

02 Oct 2025 03:32PM UTC coverage: 66.622% (-0.02%) from 66.646%
18197857992

Pull #10267

github

web-flow
Merge 0d9bfccfe into 1c2ff4a7e
Pull Request #10267: [g175] multi: small G175 preparations

24 of 141 new or added lines in 12 files covered. (17.02%)

64 existing lines in 20 files now uncovered.

137216 of 205963 relevant lines covered (66.62%)

21302.01 hits per line

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

78.46
/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/lightningnetwork/lnd/aliasmgr"
23
        "github.com/lightningnetwork/lnd/brontide"
24
        "github.com/lightningnetwork/lnd/buffer"
25
        "github.com/lightningnetwork/lnd/chainntnfs"
26
        "github.com/lightningnetwork/lnd/channeldb"
27
        "github.com/lightningnetwork/lnd/channelnotifier"
28
        "github.com/lightningnetwork/lnd/contractcourt"
29
        "github.com/lightningnetwork/lnd/discovery"
30
        "github.com/lightningnetwork/lnd/feature"
31
        "github.com/lightningnetwork/lnd/fn/v2"
32
        "github.com/lightningnetwork/lnd/funding"
33
        graphdb "github.com/lightningnetwork/lnd/graph/db"
34
        "github.com/lightningnetwork/lnd/graph/db/models"
35
        "github.com/lightningnetwork/lnd/htlcswitch"
36
        "github.com/lightningnetwork/lnd/htlcswitch/hodl"
37
        "github.com/lightningnetwork/lnd/htlcswitch/hop"
38
        "github.com/lightningnetwork/lnd/input"
39
        "github.com/lightningnetwork/lnd/invoices"
40
        "github.com/lightningnetwork/lnd/keychain"
41
        "github.com/lightningnetwork/lnd/lnpeer"
42
        "github.com/lightningnetwork/lnd/lntypes"
43
        "github.com/lightningnetwork/lnd/lnutils"
44
        "github.com/lightningnetwork/lnd/lnwallet"
45
        "github.com/lightningnetwork/lnd/lnwallet/chainfee"
46
        "github.com/lightningnetwork/lnd/lnwallet/chancloser"
47
        "github.com/lightningnetwork/lnd/lnwire"
48
        "github.com/lightningnetwork/lnd/msgmux"
49
        "github.com/lightningnetwork/lnd/netann"
50
        "github.com/lightningnetwork/lnd/pool"
51
        "github.com/lightningnetwork/lnd/protofsm"
52
        "github.com/lightningnetwork/lnd/queue"
53
        "github.com/lightningnetwork/lnd/subscribe"
54
        "github.com/lightningnetwork/lnd/ticker"
55
        "github.com/lightningnetwork/lnd/tlv"
56
        "github.com/lightningnetwork/lnd/watchtower/wtclient"
57
)
58

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

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

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

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

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

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

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

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

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

97
        // msgStreamSize is the size of the message streams.
98
        msgStreamSize = 50
99
)
100

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

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

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

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

127
        err chan error
128
}
129

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

442
        // QuiescenceTimeout is the max duration that the channel can be
443
        // quiesced. Any dependent protocols (dynamic commitments, splicing,
444
        // etc.) must finish their operations under this timeout value,
445
        // otherwise the node will disconnect.
446
        QuiescenceTimeout time.Duration
447

448
        // MaxFeeExposure limits the number of outstanding fees in a channel.
449
        // This value will be passed to created links.
450
        MaxFeeExposure lnwire.MilliSatoshi
451

452
        // MsgRouter is an optional instance of the main message router that
453
        // the peer will use. If None, then a new default version will be used
454
        // in place.
455
        MsgRouter fn.Option[msgmux.Router]
456

457
        // AuxChanCloser is an optional instance of an abstraction that can be
458
        // used to modify the way the co-op close transaction is constructed.
459
        AuxChanCloser fn.Option[chancloser.AuxChanCloser]
460

461
        // AuxChannelNegotiator is an optional interface that allows aux channel
462
        // implementations to inject and process custom records over channel
463
        // related wire messages.
464
        AuxChannelNegotiator fn.Option[lnwallet.AuxChannelNegotiator]
465

466
        // ShouldFwdExpEndorsement is a closure that indicates whether
467
        // experimental endorsement signals should be set.
468
        ShouldFwdExpEndorsement func() bool
469

470
        // NoDisconnectOnPongFailure indicates whether the peer should *not* be
471
        // disconnected if a pong is not received in time or is mismatched.
472
        NoDisconnectOnPongFailure bool
473

474
        // Quit is the server's quit channel. If this is closed, we halt operation.
475
        Quit chan struct{}
476
}
477

478
// chanCloserFsm is a union-like type that can hold the two versions of co-op
479
// close we support: negotiation, and RBF based.
480
//
481
// TODO(roasbeef): rename to chancloser.Negotiator and chancloser.RBF?
482
type chanCloserFsm = fn.Either[*chancloser.ChanCloser, *chancloser.RbfChanCloser] //nolint:ll
483

484
// makeNegotiateCloser creates a new negotiate closer from a
485
// chancloser.ChanCloser.
486
func makeNegotiateCloser(chanCloser *chancloser.ChanCloser) chanCloserFsm {
12✔
487
        return fn.NewLeft[*chancloser.ChanCloser, *chancloser.RbfChanCloser](
12✔
488
                chanCloser,
12✔
489
        )
12✔
490
}
12✔
491

492
// makeRbfCloser creates a new RBF closer from a chancloser.RbfChanCloser.
493
func makeRbfCloser(rbfCloser *chancloser.RbfChanCloser) chanCloserFsm {
3✔
494
        return fn.NewRight[*chancloser.ChanCloser](
3✔
495
                rbfCloser,
3✔
496
        )
3✔
497
}
3✔
498

499
// Brontide is an active peer on the Lightning Network. This struct is responsible
500
// for managing any channel state related to this peer. To do so, it has
501
// several helper goroutines to handle events such as HTLC timeouts, new
502
// funding workflow, and detecting an uncooperative closure of any active
503
// channels.
504
type Brontide struct {
505
        // MUST be used atomically.
506
        started    int32
507
        disconnect int32
508

509
        // MUST be used atomically.
510
        bytesReceived uint64
511
        bytesSent     uint64
512

513
        // isTorConnection is a flag that indicates whether or not we believe
514
        // the remote peer is a tor connection. It is not always possible to
515
        // know this with certainty but we have heuristics we use that should
516
        // catch most cases.
517
        //
518
        // NOTE: We judge the tor-ness of a connection by if the remote peer has
519
        // ".onion" in the address OR if it's connected over localhost.
520
        // This will miss cases where our peer is connected to our clearnet
521
        // address over the tor network (via exit nodes). It will also misjudge
522
        // actual localhost connections as tor. We need to include this because
523
        // inbound connections to our tor address will appear to come from the
524
        // local socks5 proxy. This heuristic is only used to expand the timeout
525
        // window for peers so it is OK to misjudge this. If you use this field
526
        // for any other purpose you should seriously consider whether or not
527
        // this heuristic is good enough for your use case.
528
        isTorConnection bool
529

530
        pingManager *PingManager
531

532
        // lastPingPayload stores an unsafe pointer wrapped as an atomic
533
        // variable which points to the last payload the remote party sent us
534
        // as their ping.
535
        //
536
        // MUST be used atomically.
537
        lastPingPayload atomic.Value
538

539
        cfg Config
540

541
        // activeSignal when closed signals that the peer is now active and
542
        // ready to process messages.
543
        activeSignal chan struct{}
544

545
        // startTime is the time this peer connection was successfully established.
546
        // It will be zero for peers that did not successfully call Start().
547
        startTime time.Time
548

549
        // sendQueue is the channel which is used to queue outgoing messages to be
550
        // written onto the wire. Note that this channel is unbuffered.
551
        sendQueue chan outgoingMsg
552

553
        // outgoingQueue is a buffered channel which allows second/third party
554
        // objects to queue messages to be sent out on the wire.
555
        outgoingQueue chan outgoingMsg
556

557
        // activeChannels is a map which stores the state machines of all
558
        // active channels. Channels are indexed into the map by the txid of
559
        // the funding transaction which opened the channel.
560
        //
561
        // NOTE: On startup, pending channels are stored as nil in this map.
562
        // Confirmed channels have channel data populated in the map. This means
563
        // that accesses to this map should nil-check the LightningChannel to
564
        // see if this is a pending channel or not. The tradeoff here is either
565
        // having two maps everywhere (one for pending, one for confirmed chans)
566
        // or having an extra nil-check per access.
567
        activeChannels *lnutils.SyncMap[
568
                lnwire.ChannelID, *lnwallet.LightningChannel]
569

570
        // addedChannels tracks any new channels opened during this peer's
571
        // lifecycle. We use this to filter out these new channels when the time
572
        // comes to request a reenable for active channels, since they will have
573
        // waited a shorter duration.
574
        addedChannels *lnutils.SyncMap[lnwire.ChannelID, struct{}]
575

576
        // newActiveChannel is used by the fundingManager to send fully opened
577
        // channels to the source peer which handled the funding workflow.
578
        newActiveChannel chan *newChannelMsg
579

580
        // newPendingChannel is used by the fundingManager to send pending open
581
        // channels to the source peer which handled the funding workflow.
582
        newPendingChannel chan *newChannelMsg
583

584
        // removePendingChannel is used by the fundingManager to cancel pending
585
        // open channels to the source peer when the funding flow is failed.
586
        removePendingChannel chan *newChannelMsg
587

588
        // activeMsgStreams is a map from channel id to the channel streams that
589
        // proxy messages to individual, active links.
590
        activeMsgStreams map[lnwire.ChannelID]*msgStream
591

592
        // activeChanCloses is a map that keeps track of all the active
593
        // cooperative channel closures. Any channel closing messages are directed
594
        // to one of these active state machines. Once the channel has been closed,
595
        // the state machine will be deleted from the map.
596
        activeChanCloses *lnutils.SyncMap[lnwire.ChannelID, chanCloserFsm]
597

598
        // localCloseChanReqs is a channel in which any local requests to close
599
        // a particular channel are sent over.
600
        localCloseChanReqs chan *htlcswitch.ChanClose
601

602
        // linkFailures receives all reported channel failures from the switch,
603
        // and instructs the channelManager to clean remaining channel state.
604
        linkFailures chan linkFailureReport
605

606
        // chanCloseMsgs is a channel that any message related to channel
607
        // closures are sent over. This includes lnwire.Shutdown message as
608
        // well as lnwire.ClosingSigned messages.
609
        chanCloseMsgs chan *closeMsg
610

611
        // remoteFeatures is the feature vector received from the peer during
612
        // the connection handshake.
613
        remoteFeatures *lnwire.FeatureVector
614

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

621
        // channelEventClient is the channel event subscription client that's
622
        // used to assist retry enabling the channels. This client is only
623
        // created when the reenableTimeout is no greater than 1 minute. Once
624
        // created, it is canceled once the reenabling has been finished.
625
        //
626
        // NOTE: we choose to create the client conditionally to avoid
627
        // potentially holding lots of un-consumed events.
628
        channelEventClient *subscribe.Client
629

630
        // msgRouter is an instance of the msgmux.Router which is used to send
631
        // off new wire messages for handing.
632
        msgRouter fn.Option[msgmux.Router]
633

634
        // globalMsgRouter is a flag that indicates whether we have a global
635
        // msg router. If so, then we don't worry about stopping the msg router
636
        // when a peer disconnects.
637
        globalMsgRouter bool
638

639
        startReady chan struct{}
640

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

646
        // log is a peer-specific logging instance.
647
        log btclog.Logger
648
}
649

650
// A compile-time check to ensure that Brontide satisfies the lnpeer.Peer
651
// interface.
652
var _ lnpeer.Peer = (*Brontide)(nil)
653

654
// NewBrontide creates a new Brontide from a peer.Config struct.
655
func NewBrontide(cfg Config) *Brontide {
28✔
656
        logPrefix := fmt.Sprintf("Peer(%x):", cfg.PubKeyBytes)
28✔
657

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

28✔
663
        // We'll either use the msg router instance passed in, or create a new
28✔
664
        // blank instance.
28✔
665
        msgRouter := cfg.MsgRouter.Alt(fn.Some[msgmux.Router](
28✔
666
                msgmux.NewMultiMsgRouter(),
28✔
667
        ))
28✔
668

28✔
669
        p := &Brontide{
28✔
670
                cfg:           cfg,
28✔
671
                activeSignal:  make(chan struct{}),
28✔
672
                sendQueue:     make(chan outgoingMsg),
28✔
673
                outgoingQueue: make(chan outgoingMsg),
28✔
674
                addedChannels: &lnutils.SyncMap[lnwire.ChannelID, struct{}]{},
28✔
675
                activeChannels: &lnutils.SyncMap[
28✔
676
                        lnwire.ChannelID, *lnwallet.LightningChannel,
28✔
677
                ]{},
28✔
678
                newActiveChannel:     make(chan *newChannelMsg, 1),
28✔
679
                newPendingChannel:    make(chan *newChannelMsg, 1),
28✔
680
                removePendingChannel: make(chan *newChannelMsg),
28✔
681

28✔
682
                activeMsgStreams: make(map[lnwire.ChannelID]*msgStream),
28✔
683
                activeChanCloses: &lnutils.SyncMap[
28✔
684
                        lnwire.ChannelID, chanCloserFsm,
28✔
685
                ]{},
28✔
686
                localCloseChanReqs: make(chan *htlcswitch.ChanClose),
28✔
687
                linkFailures:       make(chan linkFailureReport),
28✔
688
                chanCloseMsgs:      make(chan *closeMsg),
28✔
689
                resentChanSyncMsg:  make(map[lnwire.ChannelID]struct{}),
28✔
690
                startReady:         make(chan struct{}),
28✔
691
                log:                peerLog.WithPrefix(logPrefix),
28✔
692
                msgRouter:          msgRouter,
28✔
693
                globalMsgRouter:    globalMsgRouter,
28✔
694
                cg:                 fn.NewContextGuard(),
28✔
695
        }
28✔
696

28✔
697
        if cfg.Conn != nil && cfg.Conn.RemoteAddr() != nil {
31✔
698
                remoteAddr := cfg.Conn.RemoteAddr().String()
3✔
699
                p.isTorConnection = strings.Contains(remoteAddr, ".onion") ||
3✔
700
                        strings.Contains(remoteAddr, "127.0.0.1")
3✔
701
        }
3✔
702

703
        var (
28✔
704
                lastBlockHeader           *wire.BlockHeader
28✔
705
                lastSerializedBlockHeader [wire.MaxBlockHeaderPayload]byte
28✔
706
        )
28✔
707
        newPingPayload := func() []byte {
28✔
708
                // We query the BestBlockHeader from our BestBlockView each time
×
709
                // this is called, and update our serialized block header if
×
710
                // they differ.  Over time, we'll use this to disseminate the
×
711
                // latest block header between all our peers, which can later be
×
712
                // used to cross-check our own view of the network to mitigate
×
713
                // various types of eclipse attacks.
×
714
                header, err := p.cfg.BestBlockView.BestBlockHeader()
×
715
                if err != nil && header == lastBlockHeader {
×
716
                        return lastSerializedBlockHeader[:]
×
717
                }
×
718

719
                buf := bytes.NewBuffer(lastSerializedBlockHeader[0:0])
×
720
                err = header.Serialize(buf)
×
721
                if err == nil {
×
722
                        lastBlockHeader = header
×
723
                } else {
×
724
                        p.log.Warn("unable to serialize current block" +
×
725
                                "header for ping payload generation." +
×
726
                                "This should be impossible and means" +
×
727
                                "there is an implementation bug.")
×
728
                }
×
729

730
                return lastSerializedBlockHeader[:]
×
731
        }
732

733
        // TODO(roasbeef): make dynamic in order to create fake cover traffic.
734
        //
735
        // NOTE(proofofkeags): this was changed to be dynamic to allow better
736
        // pong identification, however, more thought is needed to make this
737
        // actually usable as a traffic decoy.
738
        randPongSize := func() uint16 {
28✔
739
                return uint16(
×
740
                        // We don't need cryptographic randomness here.
×
741
                        /* #nosec */
×
742
                        rand.Intn(pongSizeCeiling) + 1,
×
743
                )
×
744
        }
×
745

746
        p.pingManager = NewPingManager(&PingManagerConfig{
28✔
747
                NewPingPayload:   newPingPayload,
28✔
748
                NewPongSize:      randPongSize,
28✔
749
                IntervalDuration: p.scaleTimeout(pingInterval),
28✔
750
                TimeoutDuration:  p.scaleTimeout(pingTimeout),
28✔
751
                SendPing: func(ping *lnwire.Ping) {
28✔
752
                        p.queueMsg(ping, nil)
×
753
                },
×
754
                OnPongFailure: func(reason error,
755
                        timeWaitedForPong time.Duration,
756
                        lastKnownRTT time.Duration) {
×
757

×
758
                        logMsg := fmt.Sprintf("pong response "+
×
759
                                "failure for %s: %v. Time waited for this "+
×
760
                                "pong: %v. Last successful RTT: %v.",
×
761
                                p, reason, timeWaitedForPong, lastKnownRTT)
×
762

×
763
                        // If NoDisconnectOnPongFailure is true, we don't
×
764
                        // disconnect. Otherwise (if it's false, the default),
×
765
                        // we disconnect.
×
766
                        if p.cfg.NoDisconnectOnPongFailure {
×
767
                                p.log.Warnf("%s -- not disconnecting "+
×
768
                                        "due to config", logMsg)
×
769
                                return
×
770
                        }
×
771

772
                        p.log.Warnf("%s -- disconnecting", logMsg)
×
773

×
774
                        go p.Disconnect(fmt.Errorf("pong failure: %w", reason))
×
775
                },
776
        })
777

778
        return p
28✔
779
}
780

781
// Start starts all helper goroutines the peer needs for normal operations.  In
782
// the case this peer has already been started, then this function is a noop.
783
func (p *Brontide) Start() error {
6✔
784
        if atomic.AddInt32(&p.started, 1) != 1 {
6✔
785
                return nil
×
786
        }
×
787

788
        // Once we've finished starting up the peer, we'll signal to other
789
        // goroutines that the they can move forward to tear down the peer, or
790
        // carry out other relevant changes.
791
        defer close(p.startReady)
6✔
792

6✔
793
        p.log.Tracef("starting with conn[%v->%v]",
6✔
794
                p.cfg.Conn.LocalAddr(), p.cfg.Conn.RemoteAddr())
6✔
795

6✔
796
        // Fetch and then load all the active channels we have with this remote
6✔
797
        // peer from the database.
6✔
798
        activeChans, err := p.cfg.ChannelDB.FetchOpenChannels(
6✔
799
                p.cfg.Addr.IdentityKey,
6✔
800
        )
6✔
801
        if err != nil {
6✔
802
                p.log.Errorf("Unable to fetch active chans "+
×
803
                        "for peer: %v", err)
×
804
                return err
×
805
        }
×
806

807
        if len(activeChans) == 0 {
10✔
808
                go p.cfg.PrunePersistentPeerConnection(p.cfg.PubKeyBytes)
4✔
809
        }
4✔
810

811
        // Quickly check if we have any existing legacy channels with this
812
        // peer.
813
        haveLegacyChan := false
6✔
814
        for _, c := range activeChans {
11✔
815
                if c.ChanType.IsTweakless() {
10✔
816
                        continue
5✔
817
                }
818

819
                haveLegacyChan = true
3✔
820
                break
3✔
821
        }
822

823
        // Exchange local and global features, the init message should be very
824
        // first between two nodes.
825
        if err := p.sendInitMsg(haveLegacyChan); err != nil {
9✔
826
                return fmt.Errorf("unable to send init msg: %w", err)
3✔
827
        }
3✔
828

829
        // Before we launch any of the helper goroutines off the peer struct,
830
        // we'll first ensure proper adherence to the p2p protocol. The init
831
        // message MUST be sent before any other message.
832
        readErr := make(chan error, 1)
6✔
833
        msgChan := make(chan lnwire.Message, 1)
6✔
834
        p.cg.WgAdd(1)
6✔
835
        go func() {
12✔
836
                defer p.cg.WgDone()
6✔
837

6✔
838
                msg, err := p.readNextMessage()
6✔
839
                if err != nil {
9✔
840
                        readErr <- err
3✔
841
                        msgChan <- nil
3✔
842
                        return
3✔
843
                }
3✔
844
                readErr <- nil
6✔
845
                msgChan <- msg
6✔
846
        }()
847

848
        select {
6✔
849
        // In order to avoid blocking indefinitely, we'll give the other peer
850
        // an upper timeout to respond before we bail out early.
851
        case <-time.After(handshakeTimeout):
×
852
                return fmt.Errorf("peer did not complete handshake within %v",
×
853
                        handshakeTimeout)
×
854
        case err := <-readErr:
6✔
855
                if err != nil {
9✔
856
                        return fmt.Errorf("unable to read init msg: %w", err)
3✔
857
                }
3✔
858
        }
859

860
        // Once the init message arrives, we can parse it so we can figure out
861
        // the negotiation of features for this session.
862
        msg := <-msgChan
6✔
863
        if msg, ok := msg.(*lnwire.Init); ok {
12✔
864
                if err := p.handleInitMsg(msg); err != nil {
6✔
865
                        p.storeError(err)
×
866
                        return err
×
867
                }
×
868
        } else {
×
869
                return errors.New("very first message between nodes " +
×
870
                        "must be init message")
×
871
        }
×
872

873
        // Next, load all the active channels we have with this peer,
874
        // registering them with the switch and launching the necessary
875
        // goroutines required to operate them.
876
        p.log.Debugf("Loaded %v active channels from database",
6✔
877
                len(activeChans))
6✔
878

6✔
879
        // Conditionally subscribe to channel events before loading channels so
6✔
880
        // we won't miss events. This subscription is used to listen to active
6✔
881
        // channel event when reenabling channels. Once the reenabling process
6✔
882
        // is finished, this subscription will be canceled.
6✔
883
        //
6✔
884
        // NOTE: ChannelNotifier must be started before subscribing events
6✔
885
        // otherwise we'd panic here.
6✔
886
        if err := p.attachChannelEventSubscription(); err != nil {
6✔
887
                return err
×
888
        }
×
889

890
        // Register the message router now as we may need to register some
891
        // endpoints while loading the channels below.
892
        p.msgRouter.WhenSome(func(router msgmux.Router) {
12✔
893
                router.Start(context.Background())
6✔
894
        })
6✔
895

896
        msgs, err := p.loadActiveChannels(activeChans)
6✔
897
        if err != nil {
6✔
898
                return fmt.Errorf("unable to load channels: %w", err)
×
899
        }
×
900

901
        p.startTime = time.Now()
6✔
902

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

5✔
910
                // Send the messages directly via writeMessage and bypass the
5✔
911
                // writeHandler goroutine.
5✔
912
                for _, msg := range msgs {
10✔
913
                        if err := p.writeMessage(msg); err != nil {
5✔
914
                                return fmt.Errorf("unable to send "+
×
915
                                        "reestablish msg: %v", err)
×
916
                        }
×
917
                }
918
        }
919

920
        err = p.pingManager.Start()
6✔
921
        if err != nil {
6✔
922
                return fmt.Errorf("could not start ping manager %w", err)
×
923
        }
×
924

925
        p.cg.WgAdd(4)
6✔
926
        go p.queueHandler()
6✔
927
        go p.writeHandler()
6✔
928
        go p.channelManager()
6✔
929
        go p.readHandler()
6✔
930

6✔
931
        // Signal to any external processes that the peer is now active.
6✔
932
        close(p.activeSignal)
6✔
933

6✔
934
        // Node announcements don't propagate very well throughout the network
6✔
935
        // as there isn't a way to efficiently query for them through their
6✔
936
        // timestamp, mostly affecting nodes that were offline during the time
6✔
937
        // of broadcast. We'll resend our node announcement to the remote peer
6✔
938
        // as a best-effort delivery such that it can also propagate to their
6✔
939
        // peers. To ensure they can successfully process it in most cases,
6✔
940
        // we'll only resend it as long as we have at least one confirmed
6✔
941
        // advertised channel with the remote peer.
6✔
942
        //
6✔
943
        // TODO(wilmer): Remove this once we're able to query for node
6✔
944
        // announcements through their timestamps.
6✔
945
        p.cg.WgAdd(2)
6✔
946
        go p.maybeSendNodeAnn(activeChans)
6✔
947
        go p.maybeSendChannelUpdates()
6✔
948

6✔
949
        return nil
6✔
950
}
951

952
// initGossipSync initializes either a gossip syncer or an initial routing
953
// dump, depending on the negotiated synchronization method.
954
func (p *Brontide) initGossipSync() {
6✔
955
        // If the remote peer knows of the new gossip queries feature, then
6✔
956
        // we'll create a new gossipSyncer in the AuthenticatedGossiper for it.
6✔
957
        if p.remoteFeatures.HasFeature(lnwire.GossipQueriesOptional) {
12✔
958
                p.log.Info("Negotiated chan series queries")
6✔
959

6✔
960
                if p.cfg.AuthGossiper == nil {
9✔
961
                        // This should only ever be hit in the unit tests.
3✔
962
                        p.log.Warn("No AuthGossiper configured. Abandoning " +
3✔
963
                                "gossip sync.")
3✔
964
                        return
3✔
965
                }
3✔
966

967
                // Register the peer's gossip syncer with the gossiper.
968
                // This blocks synchronously to ensure the gossip syncer is
969
                // registered with the gossiper before attempting to read
970
                // messages from the remote peer.
971
                //
972
                // TODO(wilmer): Only sync updates from non-channel peers. This
973
                // requires an improved version of the current network
974
                // bootstrapper to ensure we can find and connect to non-channel
975
                // peers.
976
                p.cfg.AuthGossiper.InitSyncState(p)
3✔
977
        }
978
}
979

980
// taprootShutdownAllowed returns true if both parties have negotiated the
981
// shutdown-any-segwit feature.
982
func (p *Brontide) taprootShutdownAllowed() bool {
9✔
983
        return p.RemoteFeatures().HasFeature(lnwire.ShutdownAnySegwitOptional) &&
9✔
984
                p.LocalFeatures().HasFeature(lnwire.ShutdownAnySegwitOptional)
9✔
985
}
9✔
986

987
// rbfCoopCloseAllowed returns true if both parties have negotiated the new RBF
988
// coop close feature.
989
func (p *Brontide) rbfCoopCloseAllowed() bool {
10✔
990
        bothHaveBit := func(bit lnwire.FeatureBit) bool {
27✔
991
                return p.RemoteFeatures().HasFeature(bit) &&
17✔
992
                        p.LocalFeatures().HasFeature(bit)
17✔
993
        }
17✔
994

995
        return bothHaveBit(lnwire.RbfCoopCloseOptional) ||
10✔
996
                bothHaveBit(lnwire.RbfCoopCloseOptionalStaging)
10✔
997
}
998

999
// QuitSignal is a method that should return a channel which will be sent upon
1000
// or closed once the backing peer exits. This allows callers using the
1001
// interface to cancel any processing in the event the backing implementation
1002
// exits.
1003
//
1004
// NOTE: Part of the lnpeer.Peer interface.
1005
func (p *Brontide) QuitSignal() <-chan struct{} {
3✔
1006
        return p.cg.Done()
3✔
1007
}
3✔
1008

1009
// addrWithInternalKey takes a delivery script, then attempts to supplement it
1010
// with information related to the internal key for the addr, but only if it's
1011
// a taproot addr.
1012
func (p *Brontide) addrWithInternalKey(
1013
        deliveryScript []byte) (*chancloser.DeliveryAddrWithKey, error) {
12✔
1014

12✔
1015
        // Currently, custom channels cannot be created with external upfront
12✔
1016
        // shutdown addresses, so this shouldn't be an issue. We only require
12✔
1017
        // the internal key for taproot addresses to be able to provide a non
12✔
1018
        // inclusion proof of any scripts.
12✔
1019
        internalKeyDesc, err := lnwallet.InternalKeyForAddr(
12✔
1020
                p.cfg.Wallet, &p.cfg.Wallet.Cfg.NetParams, deliveryScript,
12✔
1021
        )
12✔
1022
        if err != nil {
12✔
1023
                return nil, fmt.Errorf("unable to fetch internal key: %w", err)
×
1024
        }
×
1025

1026
        return &chancloser.DeliveryAddrWithKey{
12✔
1027
                DeliveryAddress: deliveryScript,
12✔
1028
                InternalKey: fn.MapOption(
12✔
1029
                        func(desc keychain.KeyDescriptor) btcec.PublicKey {
15✔
1030
                                return *desc.PubKey
3✔
1031
                        },
3✔
1032
                )(internalKeyDesc),
1033
        }, nil
1034
}
1035

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

6✔
1043
        // Return a slice of messages to send to the peers in case the channel
6✔
1044
        // cannot be loaded normally.
6✔
1045
        var msgs []lnwire.Message
6✔
1046

6✔
1047
        scidAliasNegotiated := p.hasNegotiatedScidAlias()
6✔
1048

6✔
1049
        for _, dbChan := range chans {
11✔
1050
                hasScidFeature := dbChan.ChanType.HasScidAliasFeature()
5✔
1051
                if scidAliasNegotiated && !hasScidFeature {
8✔
1052
                        // We'll request and store an alias, making sure that a
3✔
1053
                        // gossiper mapping is not created for the alias to the
3✔
1054
                        // real SCID. This is done because the peer and funding
3✔
1055
                        // manager are not aware of each other's states and if
3✔
1056
                        // we did not do this, we would accept alias channel
3✔
1057
                        // updates after 6 confirmations, which would be buggy.
3✔
1058
                        // We'll queue a channel_ready message with the new
3✔
1059
                        // alias. This should technically be done *after* the
3✔
1060
                        // reestablish, but this behavior is pre-existing since
3✔
1061
                        // the funding manager may already queue a
3✔
1062
                        // channel_ready before the channel_reestablish.
3✔
1063
                        if !dbChan.IsPending {
6✔
1064
                                aliasScid, err := p.cfg.RequestAlias()
3✔
1065
                                if err != nil {
3✔
1066
                                        return nil, err
×
1067
                                }
×
1068

1069
                                err = p.cfg.AddLocalAlias(
3✔
1070
                                        aliasScid, dbChan.ShortChanID(), false,
3✔
1071
                                        false,
3✔
1072
                                )
3✔
1073
                                if err != nil {
3✔
1074
                                        return nil, err
×
1075
                                }
×
1076

1077
                                chanID := lnwire.NewChanIDFromOutPoint(
3✔
1078
                                        dbChan.FundingOutpoint,
3✔
1079
                                )
3✔
1080

3✔
1081
                                // Fetch the second commitment point to send in
3✔
1082
                                // the channel_ready message.
3✔
1083
                                second, err := dbChan.SecondCommitmentPoint()
3✔
1084
                                if err != nil {
3✔
1085
                                        return nil, err
×
1086
                                }
×
1087

1088
                                channelReadyMsg := lnwire.NewChannelReady(
3✔
1089
                                        chanID, second,
3✔
1090
                                )
3✔
1091
                                channelReadyMsg.AliasScid = &aliasScid
3✔
1092

3✔
1093
                                msgs = append(msgs, channelReadyMsg)
3✔
1094
                        }
1095

1096
                        // If we've negotiated the option-scid-alias feature
1097
                        // and this channel does not have ScidAliasFeature set
1098
                        // to true due to an upgrade where the feature bit was
1099
                        // turned on, we'll update the channel's database
1100
                        // state.
1101
                        err := dbChan.MarkScidAliasNegotiated()
3✔
1102
                        if err != nil {
3✔
1103
                                return nil, err
×
1104
                        }
×
1105
                }
1106

1107
                var chanOpts []lnwallet.ChannelOpt
5✔
1108
                p.cfg.AuxLeafStore.WhenSome(func(s lnwallet.AuxLeafStore) {
5✔
1109
                        chanOpts = append(chanOpts, lnwallet.WithLeafStore(s))
×
1110
                })
×
1111
                p.cfg.AuxSigner.WhenSome(func(s lnwallet.AuxSigner) {
5✔
1112
                        chanOpts = append(chanOpts, lnwallet.WithAuxSigner(s))
×
1113
                })
×
1114
                p.cfg.AuxResolver.WhenSome(
5✔
1115
                        func(s lnwallet.AuxContractResolver) {
5✔
1116
                                chanOpts = append(
×
1117
                                        chanOpts, lnwallet.WithAuxResolver(s),
×
1118
                                )
×
1119
                        },
×
1120
                )
1121

1122
                lnChan, err := lnwallet.NewLightningChannel(
5✔
1123
                        p.cfg.Signer, dbChan, p.cfg.SigPool, chanOpts...,
5✔
1124
                )
5✔
1125
                if err != nil {
5✔
1126
                        return nil, fmt.Errorf("unable to create channel "+
×
1127
                                "state machine: %w", err)
×
1128
                }
×
1129

1130
                chanPoint := dbChan.FundingOutpoint
5✔
1131

5✔
1132
                chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
5✔
1133

5✔
1134
                p.log.Infof("Loading ChannelPoint(%v), isPending=%v",
5✔
1135
                        chanPoint, lnChan.IsPending())
5✔
1136

5✔
1137
                // Skip adding any permanently irreconcilable channels to the
5✔
1138
                // htlcswitch.
5✔
1139
                if !dbChan.HasChanStatus(channeldb.ChanStatusDefault) &&
5✔
1140
                        !dbChan.HasChanStatus(channeldb.ChanStatusRestored) {
10✔
1141

5✔
1142
                        p.log.Warnf("ChannelPoint(%v) has status %v, won't "+
5✔
1143
                                "start.", chanPoint, dbChan.ChanStatus())
5✔
1144

5✔
1145
                        // To help our peer recover from a potential data loss,
5✔
1146
                        // we resend our channel reestablish message if the
5✔
1147
                        // channel is in a borked state. We won't process any
5✔
1148
                        // channel reestablish message sent from the peer, but
5✔
1149
                        // that's okay since the assumption is that we did when
5✔
1150
                        // marking the channel borked.
5✔
1151
                        chanSync, err := dbChan.ChanSyncMsg()
5✔
1152
                        if err != nil {
5✔
1153
                                p.log.Errorf("Unable to create channel "+
×
1154
                                        "reestablish message for channel %v: "+
×
1155
                                        "%v", chanPoint, err)
×
1156
                                continue
×
1157
                        }
1158

1159
                        msgs = append(msgs, chanSync)
5✔
1160

5✔
1161
                        // Check if this channel needs to have the cooperative
5✔
1162
                        // close process restarted. If so, we'll need to send
5✔
1163
                        // the Shutdown message that is returned.
5✔
1164
                        if dbChan.HasChanStatus(
5✔
1165
                                channeldb.ChanStatusCoopBroadcasted,
5✔
1166
                        ) {
8✔
1167

3✔
1168
                                shutdownMsg, err := p.restartCoopClose(lnChan)
3✔
1169
                                if err != nil {
3✔
1170
                                        p.log.Errorf("Unable to restart "+
×
1171
                                                "coop close for channel: %v",
×
1172
                                                err)
×
1173
                                        continue
×
1174
                                }
1175

1176
                                if shutdownMsg == nil {
6✔
1177
                                        continue
3✔
1178
                                }
1179

1180
                                // Append the message to the set of messages to
1181
                                // send.
1182
                                msgs = append(msgs, shutdownMsg)
×
1183
                        }
1184

1185
                        continue
5✔
1186
                }
1187

1188
                // Before we register this new link with the HTLC Switch, we'll
1189
                // need to fetch its current link-layer forwarding policy from
1190
                // the database.
1191
                graph := p.cfg.ChannelGraph
3✔
1192
                info, p1, p2, err := graph.FetchChannelEdgesByOutpoint(
3✔
1193
                        &chanPoint,
3✔
1194
                )
3✔
1195
                if err != nil && !errors.Is(err, graphdb.ErrEdgeNotFound) {
3✔
1196
                        return nil, err
×
1197
                }
×
1198

1199
                // We'll filter out our policy from the directional channel
1200
                // edges based whom the edge connects to. If it doesn't connect
1201
                // to us, then we know that we were the one that advertised the
1202
                // policy.
1203
                //
1204
                // TODO(roasbeef): can add helper method to get policy for
1205
                // particular channel.
1206
                var selfPolicy *models.ChannelEdgePolicy
3✔
1207
                if info != nil && bytes.Equal(info.NodeKey1Bytes[:],
3✔
1208
                        p.cfg.ServerPubKey[:]) {
6✔
1209

3✔
1210
                        selfPolicy = p1
3✔
1211
                } else {
6✔
1212
                        selfPolicy = p2
3✔
1213
                }
3✔
1214

1215
                // If we don't yet have an advertised routing policy, then
1216
                // we'll use the current default, otherwise we'll translate the
1217
                // routing policy into a forwarding policy.
1218
                var forwardingPolicy *models.ForwardingPolicy
3✔
1219
                if selfPolicy != nil {
6✔
1220
                        forwardingPolicy = &models.ForwardingPolicy{
3✔
1221
                                MinHTLCOut:    selfPolicy.MinHTLC,
3✔
1222
                                MaxHTLC:       selfPolicy.MaxHTLC,
3✔
1223
                                BaseFee:       selfPolicy.FeeBaseMSat,
3✔
1224
                                FeeRate:       selfPolicy.FeeProportionalMillionths,
3✔
1225
                                TimeLockDelta: uint32(selfPolicy.TimeLockDelta),
3✔
1226
                        }
3✔
1227
                        selfPolicy.InboundFee.WhenSome(func(fee lnwire.Fee) {
3✔
1228
                                inboundFee := models.NewInboundFeeFromWire(fee)
×
1229
                                forwardingPolicy.InboundFee = inboundFee
×
1230
                        })
×
1231
                } else {
3✔
1232
                        p.log.Warnf("Unable to find our forwarding policy "+
3✔
1233
                                "for channel %v, using default values",
3✔
1234
                                chanPoint)
3✔
1235
                        forwardingPolicy = &p.cfg.RoutingPolicy
3✔
1236
                }
3✔
1237

1238
                p.log.Tracef("Using link policy of: %v",
3✔
1239
                        lnutils.SpewLogClosure(forwardingPolicy))
3✔
1240

3✔
1241
                // If the channel is pending, set the value to nil in the
3✔
1242
                // activeChannels map. This is done to signify that the channel
3✔
1243
                // is pending. We don't add the link to the switch here - it's
3✔
1244
                // the funding manager's responsibility to spin up pending
3✔
1245
                // channels. Adding them here would just be extra work as we'll
3✔
1246
                // tear them down when creating + adding the final link.
3✔
1247
                if lnChan.IsPending() {
6✔
1248
                        p.activeChannels.Store(chanID, nil)
3✔
1249

3✔
1250
                        continue
3✔
1251
                }
1252

1253
                shutdownInfo, err := lnChan.State().ShutdownInfo()
3✔
1254
                if err != nil && !errors.Is(err, channeldb.ErrNoShutdownInfo) {
3✔
1255
                        return nil, err
×
1256
                }
×
1257

1258
                isTaprootChan := lnChan.ChanType().IsTaproot()
3✔
1259

3✔
1260
                var (
3✔
1261
                        shutdownMsg     fn.Option[lnwire.Shutdown]
3✔
1262
                        shutdownInfoErr error
3✔
1263
                )
3✔
1264
                shutdownInfo.WhenSome(func(info channeldb.ShutdownInfo) {
6✔
1265
                        // If we can use the new RBF close feature, we don't
3✔
1266
                        // need to create the legacy closer. However for taproot
3✔
1267
                        // channels, we'll continue to use the legacy closer.
3✔
1268
                        if p.rbfCoopCloseAllowed() && !isTaprootChan {
6✔
1269
                                return
3✔
1270
                        }
3✔
1271

1272
                        // Compute an ideal fee.
1273
                        feePerKw, err := p.cfg.FeeEstimator.EstimateFeePerKW(
3✔
1274
                                p.cfg.CoopCloseTargetConfs,
3✔
1275
                        )
3✔
1276
                        if err != nil {
3✔
1277
                                shutdownInfoErr = fmt.Errorf("unable to "+
×
1278
                                        "estimate fee: %w", err)
×
1279

×
1280
                                return
×
1281
                        }
×
1282

1283
                        addr, err := p.addrWithInternalKey(
3✔
1284
                                info.DeliveryScript.Val,
3✔
1285
                        )
3✔
1286
                        if err != nil {
3✔
1287
                                shutdownInfoErr = fmt.Errorf("unable to make "+
×
1288
                                        "delivery addr: %w", err)
×
1289
                                return
×
1290
                        }
×
1291
                        negotiateChanCloser, err := p.createChanCloser(
3✔
1292
                                lnChan, addr, feePerKw, nil,
3✔
1293
                                info.Closer(),
3✔
1294
                        )
3✔
1295
                        if err != nil {
3✔
1296
                                shutdownInfoErr = fmt.Errorf("unable to "+
×
1297
                                        "create chan closer: %w", err)
×
1298

×
1299
                                return
×
1300
                        }
×
1301

1302
                        chanID := lnwire.NewChanIDFromOutPoint(
3✔
1303
                                lnChan.State().FundingOutpoint,
3✔
1304
                        )
3✔
1305

3✔
1306
                        p.activeChanCloses.Store(chanID, makeNegotiateCloser(
3✔
1307
                                negotiateChanCloser,
3✔
1308
                        ))
3✔
1309

3✔
1310
                        // Create the Shutdown message.
3✔
1311
                        shutdown, err := negotiateChanCloser.ShutdownChan()
3✔
1312
                        if err != nil {
3✔
1313
                                p.activeChanCloses.Delete(chanID)
×
1314
                                shutdownInfoErr = err
×
1315

×
1316
                                return
×
1317
                        }
×
1318

1319
                        shutdownMsg = fn.Some(*shutdown)
3✔
1320
                })
1321
                if shutdownInfoErr != nil {
3✔
1322
                        return nil, shutdownInfoErr
×
1323
                }
×
1324

1325
                // Subscribe to the set of on-chain events for this channel.
1326
                chainEvents, err := p.cfg.ChainArb.SubscribeChannelEvents(
3✔
1327
                        chanPoint,
3✔
1328
                )
3✔
1329
                if err != nil {
3✔
1330
                        return nil, err
×
1331
                }
×
1332

1333
                err = p.addLink(
3✔
1334
                        &chanPoint, lnChan, forwardingPolicy, chainEvents,
3✔
1335
                        true, shutdownMsg,
3✔
1336
                )
3✔
1337
                if err != nil {
3✔
1338
                        return nil, fmt.Errorf("unable to add link %v to "+
×
1339
                                "switch: %v", chanPoint, err)
×
1340
                }
×
1341

1342
                p.activeChannels.Store(chanID, lnChan)
3✔
1343

3✔
1344
                // We're using the old co-op close, so we don't need to init
3✔
1345
                // the new RBF chan closer. If we have a taproot chan, then
3✔
1346
                // we'll also use the legacy type, so we don't need to make the
3✔
1347
                // new closer.
3✔
1348
                if !p.rbfCoopCloseAllowed() || isTaprootChan {
6✔
1349
                        continue
3✔
1350
                }
1351

1352
                // Now that the link has been added above, we'll also init an
1353
                // RBF chan closer for this channel, but only if the new close
1354
                // feature is negotiated.
1355
                //
1356
                // Creating this here ensures that any shutdown messages sent
1357
                // will be automatically routed by the msg router.
1358
                if _, err := p.initRbfChanCloser(lnChan); err != nil {
3✔
1359
                        p.activeChanCloses.Delete(chanID)
×
1360

×
1361
                        return nil, fmt.Errorf("unable to init RBF chan "+
×
1362
                                "closer during peer connect: %w", err)
×
1363
                }
×
1364

1365
                // If the shutdown info isn't blank, then we should kick things
1366
                // off by sending a shutdown message to the remote party to
1367
                // continue the old shutdown flow.
1368
                restartShutdown := func(s channeldb.ShutdownInfo) error {
6✔
1369
                        return p.startRbfChanCloser(
3✔
1370
                                newRestartShutdownInit(s),
3✔
1371
                                lnChan.ChannelPoint(),
3✔
1372
                        )
3✔
1373
                }
3✔
1374
                err = fn.MapOptionZ(shutdownInfo, restartShutdown)
3✔
1375
                if err != nil {
3✔
1376
                        return nil, fmt.Errorf("unable to start RBF "+
×
1377
                                "chan closer: %w", err)
×
1378
                }
×
1379
        }
1380

1381
        return msgs, nil
6✔
1382
}
1383

1384
// addLink creates and adds a new ChannelLink from the specified channel.
1385
func (p *Brontide) addLink(chanPoint *wire.OutPoint,
1386
        lnChan *lnwallet.LightningChannel,
1387
        forwardingPolicy *models.ForwardingPolicy,
1388
        chainEvents *contractcourt.ChainEventSubscription,
1389
        syncStates bool, shutdownMsg fn.Option[lnwire.Shutdown]) error {
3✔
1390

3✔
1391
        // onChannelFailure will be called by the link in case the channel
3✔
1392
        // fails for some reason.
3✔
1393
        onChannelFailure := func(chanID lnwire.ChannelID,
3✔
1394
                shortChanID lnwire.ShortChannelID,
3✔
1395
                linkErr htlcswitch.LinkFailureError) {
6✔
1396

3✔
1397
                failure := linkFailureReport{
3✔
1398
                        chanPoint:   *chanPoint,
3✔
1399
                        chanID:      chanID,
3✔
1400
                        shortChanID: shortChanID,
3✔
1401
                        linkErr:     linkErr,
3✔
1402
                }
3✔
1403

3✔
1404
                select {
3✔
1405
                case p.linkFailures <- failure:
3✔
1406
                case <-p.cg.Done():
×
1407
                case <-p.cfg.Quit:
1✔
1408
                }
1409
        }
1410

1411
        updateContractSignals := func(signals *contractcourt.ContractSignals) error {
6✔
1412
                return p.cfg.ChainArb.UpdateContractSignals(*chanPoint, signals)
3✔
1413
        }
3✔
1414

1415
        notifyContractUpdate := func(update *contractcourt.ContractUpdate) error {
6✔
1416
                return p.cfg.ChainArb.NotifyContractUpdate(*chanPoint, update)
3✔
1417
        }
3✔
1418

1419
        //nolint:ll
1420
        linkCfg := htlcswitch.ChannelLinkConfig{
3✔
1421
                Peer:                   p,
3✔
1422
                DecodeHopIterators:     p.cfg.Sphinx.DecodeHopIterators,
3✔
1423
                ExtractErrorEncrypter:  p.cfg.Sphinx.ExtractErrorEncrypter,
3✔
1424
                FetchLastChannelUpdate: p.cfg.FetchLastChanUpdate,
3✔
1425
                HodlMask:               p.cfg.Hodl.Mask(),
3✔
1426
                Registry:               p.cfg.Invoices,
3✔
1427
                BestHeight:             p.cfg.Switch.BestHeight,
3✔
1428
                Circuits:               p.cfg.Switch.CircuitModifier(),
3✔
1429
                ForwardPackets:         p.cfg.InterceptSwitch.ForwardPackets,
3✔
1430
                FwrdingPolicy:          *forwardingPolicy,
3✔
1431
                FeeEstimator:           p.cfg.FeeEstimator,
3✔
1432
                PreimageCache:          p.cfg.WitnessBeacon,
3✔
1433
                ChainEvents:            chainEvents,
3✔
1434
                UpdateContractSignals:  updateContractSignals,
3✔
1435
                NotifyContractUpdate:   notifyContractUpdate,
3✔
1436
                OnChannelFailure:       onChannelFailure,
3✔
1437
                SyncStates:             syncStates,
3✔
1438
                BatchTicker:            ticker.New(p.cfg.ChannelCommitInterval),
3✔
1439
                FwdPkgGCTicker:         ticker.New(time.Hour),
3✔
1440
                PendingCommitTicker: ticker.New(
3✔
1441
                        p.cfg.PendingCommitInterval,
3✔
1442
                ),
3✔
1443
                BatchSize:               p.cfg.ChannelCommitBatchSize,
3✔
1444
                UnsafeReplay:            p.cfg.UnsafeReplay,
3✔
1445
                MinUpdateTimeout:        htlcswitch.DefaultMinLinkFeeUpdateTimeout,
3✔
1446
                MaxUpdateTimeout:        htlcswitch.DefaultMaxLinkFeeUpdateTimeout,
3✔
1447
                OutgoingCltvRejectDelta: p.cfg.OutgoingCltvRejectDelta,
3✔
1448
                TowerClient:             p.cfg.TowerClient,
3✔
1449
                MaxOutgoingCltvExpiry:   p.cfg.MaxOutgoingCltvExpiry,
3✔
1450
                MaxFeeAllocation:        p.cfg.MaxChannelFeeAllocation,
3✔
1451
                MaxAnchorsCommitFeeRate: p.cfg.MaxAnchorsCommitFeeRate,
3✔
1452
                NotifyActiveLink:        p.cfg.ChannelNotifier.NotifyActiveLinkEvent,
3✔
1453
                NotifyActiveChannel:     p.cfg.ChannelNotifier.NotifyActiveChannelEvent,
3✔
1454
                NotifyInactiveChannel:   p.cfg.ChannelNotifier.NotifyInactiveChannelEvent,
3✔
1455
                NotifyInactiveLinkEvent: p.cfg.ChannelNotifier.NotifyInactiveLinkEvent,
3✔
1456
                HtlcNotifier:            p.cfg.HtlcNotifier,
3✔
1457
                GetAliases:              p.cfg.GetAliases,
3✔
1458
                PreviouslySentShutdown:  shutdownMsg,
3✔
1459
                DisallowRouteBlinding:   p.cfg.DisallowRouteBlinding,
3✔
1460
                MaxFeeExposure:          p.cfg.MaxFeeExposure,
3✔
1461
                ShouldFwdExpEndorsement: p.cfg.ShouldFwdExpEndorsement,
3✔
1462
                DisallowQuiescence: p.cfg.DisallowQuiescence ||
3✔
1463
                        !p.remoteFeatures.HasFeature(lnwire.QuiescenceOptional),
3✔
1464
                AuxTrafficShaper:     p.cfg.AuxTrafficShaper,
3✔
1465
                AuxChannelNegotiator: p.cfg.AuxChannelNegotiator,
3✔
1466
                QuiescenceTimeout:    p.cfg.QuiescenceTimeout,
3✔
1467
        }
3✔
1468

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

3✔
1476
        // With the channel link created, we'll now notify the htlc switch so
3✔
1477
        // this channel can be used to dispatch local payments and also
3✔
1478
        // passively forward payments.
3✔
1479
        return p.cfg.Switch.CreateAndAddLink(linkCfg, lnChan)
3✔
1480
}
1481

1482
// maybeSendNodeAnn sends our node announcement to the remote peer if at least
1483
// one confirmed public channel exists with them.
1484
func (p *Brontide) maybeSendNodeAnn(channels []*channeldb.OpenChannel) {
6✔
1485
        defer p.cg.WgDone()
6✔
1486

6✔
1487
        hasConfirmedPublicChan := false
6✔
1488
        for _, channel := range channels {
11✔
1489
                if channel.IsPending {
8✔
1490
                        continue
3✔
1491
                }
1492
                if channel.ChannelFlags&lnwire.FFAnnounceChannel == 0 {
10✔
1493
                        continue
5✔
1494
                }
1495

1496
                hasConfirmedPublicChan = true
3✔
1497
                break
3✔
1498
        }
1499
        if !hasConfirmedPublicChan {
12✔
1500
                return
6✔
1501
        }
6✔
1502

1503
        ourNodeAnn, err := p.cfg.GenNodeAnnouncement()
3✔
1504
        if err != nil {
3✔
1505
                p.log.Debugf("Unable to retrieve node announcement: %v", err)
×
1506
                return
×
1507
        }
×
1508

1509
        if err := p.SendMessageLazy(false, &ourNodeAnn); err != nil {
3✔
1510
                p.log.Debugf("Unable to resend node announcement: %v", err)
×
1511
        }
×
1512
}
1513

1514
// maybeSendChannelUpdates sends our channel updates to the remote peer if we
1515
// have any active channels with them.
1516
func (p *Brontide) maybeSendChannelUpdates() {
6✔
1517
        defer p.cg.WgDone()
6✔
1518

6✔
1519
        // If we don't have any active channels, then we can exit early.
6✔
1520
        if p.activeChannels.Len() == 0 {
10✔
1521
                return
4✔
1522
        }
4✔
1523

1524
        maybeSendUpd := func(cid lnwire.ChannelID,
5✔
1525
                lnChan *lnwallet.LightningChannel) error {
10✔
1526

5✔
1527
                // Nil channels are pending, so we'll skip them.
5✔
1528
                if lnChan == nil {
8✔
1529
                        return nil
3✔
1530
                }
3✔
1531

1532
                dbChan := lnChan.State()
5✔
1533
                scid := func() lnwire.ShortChannelID {
10✔
1534
                        switch {
5✔
1535
                        // Otherwise if it's a zero conf channel and confirmed,
1536
                        // then we need to use the "real" scid.
1537
                        case dbChan.IsZeroConf() && dbChan.ZeroConfConfirmed():
3✔
1538
                                return dbChan.ZeroConfRealScid()
3✔
1539

1540
                        // Otherwise, we can use the normal scid.
1541
                        default:
5✔
1542
                                return dbChan.ShortChanID()
5✔
1543
                        }
1544
                }()
1545

1546
                // Now that we know the channel is in a good state, we'll try
1547
                // to fetch the update to send to the remote peer. If the
1548
                // channel is pending, and not a zero conf channel, we'll get
1549
                // an error here which we'll ignore.
1550
                chanUpd, err := p.cfg.FetchLastChanUpdate(scid)
5✔
1551
                if err != nil {
8✔
1552
                        p.log.Debugf("Unable to fetch channel update for "+
3✔
1553
                                "ChannelPoint(%v), scid=%v: %v",
3✔
1554
                                dbChan.FundingOutpoint, dbChan.ShortChanID, err)
3✔
1555

3✔
1556
                        return nil
3✔
1557
                }
3✔
1558

1559
                p.log.Debugf("Sending channel update for ChannelPoint(%v), "+
5✔
1560
                        "scid=%v", dbChan.FundingOutpoint, dbChan.ShortChanID)
5✔
1561

5✔
1562
                // We'll send it as a normal message instead of using the lazy
5✔
1563
                // queue to prioritize transmission of the fresh update.
5✔
1564
                if err := p.SendMessage(false, chanUpd); err != nil {
5✔
1565
                        err := fmt.Errorf("unable to send channel update for "+
×
1566
                                "ChannelPoint(%v), scid=%v: %w",
×
1567
                                dbChan.FundingOutpoint, dbChan.ShortChanID(),
×
1568
                                err)
×
1569
                        p.log.Errorf(err.Error())
×
1570

×
1571
                        return err
×
1572
                }
×
1573

1574
                return nil
5✔
1575
        }
1576

1577
        p.activeChannels.ForEach(maybeSendUpd)
5✔
1578
}
1579

1580
// WaitForDisconnect waits until the peer has disconnected. A peer may be
1581
// disconnected if the local or remote side terminates the connection, or an
1582
// irrecoverable protocol error has been encountered. This method will only
1583
// begin watching the peer's waitgroup after the ready channel or the peer's
1584
// quit channel are signaled. The ready channel should only be signaled if a
1585
// call to Start returns no error. Otherwise, if the peer fails to start,
1586
// calling Disconnect will signal the quit channel and the method will not
1587
// block, since no goroutines were spawned.
1588
func (p *Brontide) WaitForDisconnect(ready chan struct{}) {
3✔
1589
        // Before we try to call the `Wait` goroutine, we'll make sure the main
3✔
1590
        // set of goroutines are already active.
3✔
1591
        select {
3✔
1592
        case <-p.startReady:
3✔
UNCOV
1593
        case <-p.cg.Done():
×
UNCOV
1594
                return
×
1595
        }
1596

1597
        select {
3✔
1598
        case <-ready:
3✔
1599
        case <-p.cg.Done():
3✔
1600
        }
1601

1602
        p.cg.WgWait()
3✔
1603
}
1604

1605
// Disconnect terminates the connection with the remote peer. Additionally, a
1606
// signal is sent to the server and htlcSwitch indicating the resources
1607
// allocated to the peer can now be cleaned up.
1608
//
1609
// NOTE: Be aware that this method will block if the peer is still starting up.
1610
// Therefore consider starting it in a goroutine if you cannot guarantee that
1611
// the peer has finished starting up before calling this method.
1612
func (p *Brontide) Disconnect(reason error) {
3✔
1613
        if !atomic.CompareAndSwapInt32(&p.disconnect, 0, 1) {
6✔
1614
                return
3✔
1615
        }
3✔
1616

1617
        // Make sure initialization has completed before we try to tear things
1618
        // down.
1619
        //
1620
        // NOTE: We only read the `startReady` chan if the peer has been
1621
        // started, otherwise we will skip reading it as this chan won't be
1622
        // closed, hence blocks forever.
1623
        if atomic.LoadInt32(&p.started) == 1 {
6✔
1624
                p.log.Debugf("Peer hasn't finished starting up yet, waiting " +
3✔
1625
                        "on startReady signal before closing connection")
3✔
1626

3✔
1627
                select {
3✔
1628
                case <-p.startReady:
3✔
1629
                case <-p.cg.Done():
×
1630
                        return
×
1631
                }
1632
        }
1633

1634
        err := fmt.Errorf("disconnecting %s, reason: %v", p, reason)
3✔
1635
        p.storeError(err)
3✔
1636

3✔
1637
        p.log.Infof(err.Error())
3✔
1638

3✔
1639
        // Stop PingManager before closing TCP connection.
3✔
1640
        p.pingManager.Stop()
3✔
1641

3✔
1642
        // Ensure that the TCP connection is properly closed before continuing.
3✔
1643
        p.cfg.Conn.Close()
3✔
1644

3✔
1645
        p.cg.Quit()
3✔
1646

3✔
1647
        // If our msg router isn't global (local to this instance), then we'll
3✔
1648
        // stop it. Otherwise, we'll leave it running.
3✔
1649
        if !p.globalMsgRouter {
6✔
1650
                p.msgRouter.WhenSome(func(router msgmux.Router) {
6✔
1651
                        router.Stop()
3✔
1652
                })
3✔
1653
        }
1654
}
1655

1656
// String returns the string representation of this peer.
1657
func (p *Brontide) String() string {
3✔
1658
        return fmt.Sprintf("%x@%s", p.cfg.PubKeyBytes, p.cfg.Conn.RemoteAddr())
3✔
1659
}
3✔
1660

1661
// readNextMessage reads, and returns the next message on the wire along with
1662
// any additional raw payload.
1663
func (p *Brontide) readNextMessage() (lnwire.Message, error) {
10✔
1664
        noiseConn := p.cfg.Conn
10✔
1665
        err := noiseConn.SetReadDeadline(time.Time{})
10✔
1666
        if err != nil {
10✔
1667
                return nil, err
×
1668
        }
×
1669

1670
        pktLen, err := noiseConn.ReadNextHeader()
10✔
1671
        if err != nil {
13✔
1672
                return nil, fmt.Errorf("read next header: %w", err)
3✔
1673
        }
3✔
1674

1675
        // First we'll read the next _full_ message. We do this rather than
1676
        // reading incrementally from the stream as the Lightning wire protocol
1677
        // is message oriented and allows nodes to pad on additional data to
1678
        // the message stream.
1679
        var (
7✔
1680
                nextMsg lnwire.Message
7✔
1681
                msgLen  uint64
7✔
1682
        )
7✔
1683
        err = p.cfg.ReadPool.Submit(func(buf *buffer.Read) error {
14✔
1684
                // Before reading the body of the message, set the read timeout
7✔
1685
                // accordingly to ensure we don't block other readers using the
7✔
1686
                // pool. We do so only after the task has been scheduled to
7✔
1687
                // ensure the deadline doesn't expire while the message is in
7✔
1688
                // the process of being scheduled.
7✔
1689
                readDeadline := time.Now().Add(
7✔
1690
                        p.scaleTimeout(readMessageTimeout),
7✔
1691
                )
7✔
1692
                readErr := noiseConn.SetReadDeadline(readDeadline)
7✔
1693
                if readErr != nil {
7✔
1694
                        return readErr
×
1695
                }
×
1696

1697
                // The ReadNextBody method will actually end up re-using the
1698
                // buffer, so within this closure, we can continue to use
1699
                // rawMsg as it's just a slice into the buf from the buffer
1700
                // pool.
1701
                rawMsg, readErr := noiseConn.ReadNextBody(buf[:pktLen])
7✔
1702
                if readErr != nil {
7✔
1703
                        return fmt.Errorf("read next body: %w", readErr)
×
1704
                }
×
1705
                msgLen = uint64(len(rawMsg))
7✔
1706

7✔
1707
                // Next, create a new io.Reader implementation from the raw
7✔
1708
                // message, and use this to decode the message directly from.
7✔
1709
                msgReader := bytes.NewReader(rawMsg)
7✔
1710
                nextMsg, err = lnwire.ReadMessage(msgReader, 0)
7✔
1711
                if err != nil {
10✔
1712
                        return err
3✔
1713
                }
3✔
1714

1715
                // At this point, rawMsg and buf will be returned back to the
1716
                // buffer pool for re-use.
1717
                return nil
7✔
1718
        })
1719
        atomic.AddUint64(&p.bytesReceived, msgLen)
7✔
1720
        if err != nil {
10✔
1721
                return nil, err
3✔
1722
        }
3✔
1723

1724
        p.logWireMessage(nextMsg, true)
7✔
1725

7✔
1726
        return nextMsg, nil
7✔
1727
}
1728

1729
// msgStream implements a goroutine-safe, in-order stream of messages to be
1730
// delivered via closure to a receiver. These messages MUST be in order due to
1731
// the nature of the lightning channel commitment and gossiper state machines.
1732
// TODO(conner): use stream handler interface to abstract out stream
1733
// state/logging.
1734
type msgStream struct {
1735
        streamShutdown int32 // To be used atomically.
1736

1737
        peer *Brontide
1738

1739
        apply func(lnwire.Message)
1740

1741
        startMsg string
1742
        stopMsg  string
1743

1744
        msgCond *sync.Cond
1745
        msgs    []lnwire.Message
1746

1747
        mtx sync.Mutex
1748

1749
        producerSema chan struct{}
1750

1751
        wg   sync.WaitGroup
1752
        quit chan struct{}
1753
}
1754

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

6✔
1763
        stream := &msgStream{
6✔
1764
                peer:         p,
6✔
1765
                apply:        apply,
6✔
1766
                startMsg:     startMsg,
6✔
1767
                stopMsg:      stopMsg,
6✔
1768
                producerSema: make(chan struct{}, bufSize),
6✔
1769
                quit:         make(chan struct{}),
6✔
1770
        }
6✔
1771
        stream.msgCond = sync.NewCond(&stream.mtx)
6✔
1772

6✔
1773
        // Before we return the active stream, we'll populate the producer's
6✔
1774
        // semaphore channel. We'll use this to ensure that the producer won't
6✔
1775
        // attempt to allocate memory in the queue for an item until it has
6✔
1776
        // sufficient extra space.
6✔
1777
        for i := uint32(0); i < bufSize; i++ {
159✔
1778
                stream.producerSema <- struct{}{}
153✔
1779
        }
153✔
1780

1781
        return stream
6✔
1782
}
1783

1784
// Start starts the chanMsgStream.
1785
func (ms *msgStream) Start() {
6✔
1786
        ms.wg.Add(1)
6✔
1787
        go ms.msgConsumer()
6✔
1788
}
6✔
1789

1790
// Stop stops the chanMsgStream.
1791
func (ms *msgStream) Stop() {
3✔
1792
        // TODO(roasbeef): signal too?
3✔
1793

3✔
1794
        close(ms.quit)
3✔
1795

3✔
1796
        // Now that we've closed the channel, we'll repeatedly signal the msg
3✔
1797
        // consumer until we've detected that it has exited.
3✔
1798
        for atomic.LoadInt32(&ms.streamShutdown) == 0 {
6✔
1799
                ms.msgCond.Signal()
3✔
1800
                time.Sleep(time.Millisecond * 100)
3✔
1801
        }
3✔
1802

1803
        ms.wg.Wait()
3✔
1804
}
1805

1806
// msgConsumer is the main goroutine that streams messages from the peer's
1807
// readHandler directly to the target channel.
1808
func (ms *msgStream) msgConsumer() {
6✔
1809
        defer ms.wg.Done()
6✔
1810
        defer peerLog.Tracef(ms.stopMsg)
6✔
1811
        defer atomic.StoreInt32(&ms.streamShutdown, 1)
6✔
1812

6✔
1813
        peerLog.Tracef(ms.startMsg)
6✔
1814

6✔
1815
        for {
12✔
1816
                // First, we'll check our condition. If the queue of messages
6✔
1817
                // is empty, then we'll wait until a new item is added.
6✔
1818
                ms.msgCond.L.Lock()
6✔
1819
                for len(ms.msgs) == 0 {
12✔
1820
                        ms.msgCond.Wait()
6✔
1821

6✔
1822
                        // If we woke up in order to exit, then we'll do so.
6✔
1823
                        // Otherwise, we'll check the message queue for any new
6✔
1824
                        // items.
6✔
1825
                        select {
6✔
1826
                        case <-ms.peer.cg.Done():
3✔
1827
                                ms.msgCond.L.Unlock()
3✔
1828
                                return
3✔
1829
                        case <-ms.quit:
3✔
1830
                                ms.msgCond.L.Unlock()
3✔
1831
                                return
3✔
1832
                        default:
3✔
1833
                        }
1834
                }
1835

1836
                // Grab the message off the front of the queue, shifting the
1837
                // slice's reference down one in order to remove the message
1838
                // from the queue.
1839
                msg := ms.msgs[0]
3✔
1840
                ms.msgs[0] = nil // Set to nil to prevent GC leak.
3✔
1841
                ms.msgs = ms.msgs[1:]
3✔
1842

3✔
1843
                ms.msgCond.L.Unlock()
3✔
1844

3✔
1845
                ms.apply(msg)
3✔
1846

3✔
1847
                // We've just successfully processed an item, so we'll signal
3✔
1848
                // to the producer that a new slot in the buffer. We'll use
3✔
1849
                // this to bound the size of the buffer to avoid allowing it to
3✔
1850
                // grow indefinitely.
3✔
1851
                select {
3✔
1852
                case ms.producerSema <- struct{}{}:
3✔
1853
                case <-ms.peer.cg.Done():
3✔
1854
                        return
3✔
1855
                case <-ms.quit:
3✔
1856
                        return
3✔
1857
                }
1858
        }
1859
}
1860

1861
// AddMsg adds a new message to the msgStream. This function is safe for
1862
// concurrent access.
1863
func (ms *msgStream) AddMsg(msg lnwire.Message) {
3✔
1864
        // First, we'll attempt to receive from the producerSema struct. This
3✔
1865
        // acts as a semaphore to prevent us from indefinitely buffering
3✔
1866
        // incoming items from the wire. Either the msg queue isn't full, and
3✔
1867
        // we'll not block, or the queue is full, and we'll block until either
3✔
1868
        // we're signalled to quit, or a slot is freed up.
3✔
1869
        select {
3✔
1870
        case <-ms.producerSema:
3✔
1871
        case <-ms.peer.cg.Done():
×
1872
                return
×
1873
        case <-ms.quit:
×
1874
                return
×
1875
        }
1876

1877
        // Next, we'll lock the condition, and add the message to the end of
1878
        // the message queue.
1879
        ms.msgCond.L.Lock()
3✔
1880
        ms.msgs = append(ms.msgs, msg)
3✔
1881
        ms.msgCond.L.Unlock()
3✔
1882

3✔
1883
        // With the message added, we signal to the msgConsumer that there are
3✔
1884
        // additional messages to consume.
3✔
1885
        ms.msgCond.Signal()
3✔
1886
}
1887

1888
// waitUntilLinkActive waits until the target link is active and returns a
1889
// ChannelLink to pass messages to. It accomplishes this by subscribing to
1890
// an ActiveLinkEvent which is emitted by the link when it first starts up.
1891
func waitUntilLinkActive(p *Brontide,
1892
        cid lnwire.ChannelID) htlcswitch.ChannelUpdateHandler {
3✔
1893

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

3✔
1896
        // Subscribe to receive channel events.
3✔
1897
        //
3✔
1898
        // NOTE: If the link is already active by SubscribeChannelEvents, then
3✔
1899
        // GetLink will retrieve the link and we can send messages. If the link
3✔
1900
        // becomes active between SubscribeChannelEvents and GetLink, then GetLink
3✔
1901
        // will retrieve the link. If the link becomes active after GetLink, then
3✔
1902
        // we will get an ActiveLinkEvent notification and retrieve the link. If
3✔
1903
        // the call to GetLink is before SubscribeChannelEvents, however, there
3✔
1904
        // will be a race condition.
3✔
1905
        sub, err := p.cfg.ChannelNotifier.SubscribeChannelEvents()
3✔
1906
        if err != nil {
6✔
1907
                // If we have a non-nil error, then the server is shutting down and we
3✔
1908
                // can exit here and return nil. This means no message will be delivered
3✔
1909
                // to the link.
3✔
1910
                return nil
3✔
1911
        }
3✔
1912
        defer sub.Cancel()
3✔
1913

3✔
1914
        // The link may already be active by this point, and we may have missed the
3✔
1915
        // ActiveLinkEvent. Check if the link exists.
3✔
1916
        link := p.fetchLinkFromKeyAndCid(cid)
3✔
1917
        if link != nil {
6✔
1918
                return link
3✔
1919
        }
3✔
1920

1921
        // If the link is nil, we must wait for it to be active.
1922
        for {
6✔
1923
                select {
3✔
1924
                // A new event has been sent by the ChannelNotifier. We first check
1925
                // whether the event is an ActiveLinkEvent. If it is, we'll check
1926
                // that the event is for this channel. Otherwise, we discard the
1927
                // message.
1928
                case e := <-sub.Updates():
3✔
1929
                        event, ok := e.(channelnotifier.ActiveLinkEvent)
3✔
1930
                        if !ok {
6✔
1931
                                // Ignore this notification.
3✔
1932
                                continue
3✔
1933
                        }
1934

1935
                        chanPoint := event.ChannelPoint
3✔
1936

3✔
1937
                        // Check whether the retrieved chanPoint matches the target
3✔
1938
                        // channel id.
3✔
1939
                        if !cid.IsChanPoint(chanPoint) {
3✔
1940
                                continue
×
1941
                        }
1942

1943
                        // The link shouldn't be nil as we received an
1944
                        // ActiveLinkEvent. If it is nil, we return nil and the
1945
                        // calling function should catch it.
1946
                        return p.fetchLinkFromKeyAndCid(cid)
3✔
1947

1948
                case <-p.cg.Done():
3✔
1949
                        return nil
3✔
1950
                }
1951
        }
1952
}
1953

1954
// newChanMsgStream is used to create a msgStream between the peer and
1955
// particular channel link in the htlcswitch. We utilize additional
1956
// synchronization with the fundingManager to ensure we don't attempt to
1957
// dispatch a message to a channel before it is fully active. A reference to the
1958
// channel this stream forwards to is held in scope to prevent unnecessary
1959
// lookups.
1960
func newChanMsgStream(p *Brontide, cid lnwire.ChannelID) *msgStream {
3✔
1961
        var chanLink htlcswitch.ChannelUpdateHandler
3✔
1962

3✔
1963
        apply := func(msg lnwire.Message) {
6✔
1964
                // This check is fine because if the link no longer exists, it will
3✔
1965
                // be removed from the activeChannels map and subsequent messages
3✔
1966
                // shouldn't reach the chan msg stream.
3✔
1967
                if chanLink == nil {
6✔
1968
                        chanLink = waitUntilLinkActive(p, cid)
3✔
1969

3✔
1970
                        // If the link is still not active and the calling function
3✔
1971
                        // errored out, just return.
3✔
1972
                        if chanLink == nil {
6✔
1973
                                p.log.Warnf("Link=%v is not active", cid)
3✔
1974
                                return
3✔
1975
                        }
3✔
1976
                }
1977

1978
                // In order to avoid unnecessarily delivering message
1979
                // as the peer is exiting, we'll check quickly to see
1980
                // if we need to exit.
1981
                select {
3✔
1982
                case <-p.cg.Done():
×
1983
                        return
×
1984
                default:
3✔
1985
                }
1986

1987
                chanLink.HandleChannelUpdate(msg)
3✔
1988
        }
1989

1990
        return newMsgStream(p,
3✔
1991
                fmt.Sprintf("Update stream for ChannelID(%x) created", cid[:]),
3✔
1992
                fmt.Sprintf("Update stream for ChannelID(%x) exiting", cid[:]),
3✔
1993
                msgStreamSize,
3✔
1994
                apply,
3✔
1995
        )
3✔
1996
}
1997

1998
// newDiscMsgStream is used to setup a msgStream between the peer and the
1999
// authenticated gossiper. This stream should be used to forward all remote
2000
// channel announcements.
2001
func newDiscMsgStream(p *Brontide) *msgStream {
6✔
2002
        apply := func(msg lnwire.Message) {
9✔
2003
                // TODO(elle): thread contexts through the peer system properly
3✔
2004
                // so that a parent context can be passed in here.
3✔
2005
                ctx := context.TODO()
3✔
2006

3✔
2007
                // Processing here means we send it to the gossiper which then
3✔
2008
                // decides whether this message is processed immediately or
3✔
2009
                // waits for dependent messages to be processed. It can also
3✔
2010
                // happen that the message is not processed at all if it is
3✔
2011
                // premature and the LRU cache fills up and the message is
3✔
2012
                // deleted.
3✔
2013
                p.log.Debugf("Processing remote msg %T", msg)
3✔
2014

3✔
2015
                // TODO(ziggie): ProcessRemoteAnnouncement returns an error
3✔
2016
                // channel, but we cannot rely on it being written to.
3✔
2017
                // Because some messages might never be processed (e.g.
3✔
2018
                // premature channel updates). We should change the design here
3✔
2019
                // and use the actor model pattern as soon as it is available.
3✔
2020
                // So for now we should NOT use the error channel.
3✔
2021
                // See https://github.com/lightningnetwork/lnd/pull/9820.
3✔
2022
                p.cfg.AuthGossiper.ProcessRemoteAnnouncement(ctx, msg, p)
3✔
2023
        }
3✔
2024

2025
        return newMsgStream(
6✔
2026
                p,
6✔
2027
                "Update stream for gossiper created",
6✔
2028
                "Update stream for gossiper exited",
6✔
2029
                msgStreamSize,
6✔
2030
                apply,
6✔
2031
        )
6✔
2032
}
2033

2034
// readHandler is responsible for reading messages off the wire in series, then
2035
// properly dispatching the handling of the message to the proper subsystem.
2036
//
2037
// NOTE: This method MUST be run as a goroutine.
2038
func (p *Brontide) readHandler() {
6✔
2039
        defer p.cg.WgDone()
6✔
2040

6✔
2041
        // We'll stop the timer after a new messages is received, and also
6✔
2042
        // reset it after we process the next message.
6✔
2043
        idleTimer := time.AfterFunc(idleTimeout, func() {
6✔
2044
                err := fmt.Errorf("peer %s no answer for %s -- disconnecting",
×
2045
                        p, idleTimeout)
×
2046
                p.Disconnect(err)
×
2047
        })
×
2048

2049
        // Initialize our negotiated gossip sync method before reading messages
2050
        // off the wire. When using gossip queries, this ensures a gossip
2051
        // syncer is active by the time query messages arrive.
2052
        //
2053
        // TODO(conner): have peer store gossip syncer directly and bypass
2054
        // gossiper?
2055
        p.initGossipSync()
6✔
2056

6✔
2057
        discStream := newDiscMsgStream(p)
6✔
2058
        discStream.Start()
6✔
2059
        defer discStream.Stop()
6✔
2060
out:
6✔
2061
        for atomic.LoadInt32(&p.disconnect) == 0 {
13✔
2062
                nextMsg, err := p.readNextMessage()
7✔
2063
                if !idleTimer.Stop() {
10✔
2064
                        select {
3✔
2065
                        case <-idleTimer.C:
×
2066
                        default:
3✔
2067
                        }
2068
                }
2069
                if err != nil {
7✔
2070
                        p.log.Infof("unable to read message from peer: %v", err)
3✔
2071

3✔
2072
                        // If we could not read our peer's message due to an
3✔
2073
                        // unknown type or invalid alias, we continue processing
3✔
2074
                        // as normal. We store unknown message and address
3✔
2075
                        // types, as they may provide debugging insight.
3✔
2076
                        switch e := err.(type) {
3✔
2077
                        // If this is just a message we don't yet recognize,
2078
                        // we'll continue processing as normal as this allows
2079
                        // us to introduce new messages in a forwards
2080
                        // compatible manner.
2081
                        case *lnwire.UnknownMessage:
3✔
2082
                                p.storeError(e)
3✔
2083
                                idleTimer.Reset(idleTimeout)
3✔
2084
                                continue
3✔
2085

2086
                        // If they sent us an address type that we don't yet
2087
                        // know of, then this isn't a wire error, so we'll
2088
                        // simply continue parsing the remainder of their
2089
                        // messages.
2090
                        case *lnwire.ErrUnknownAddrType:
×
2091
                                p.storeError(e)
×
2092
                                idleTimer.Reset(idleTimeout)
×
2093
                                continue
×
2094

2095
                        // If the NodeAnnouncement1 has an invalid alias, then
2096
                        // we'll log that error above and continue so we can
2097
                        // continue to read messages from the peer. We do not
2098
                        // store this error because it is of little debugging
2099
                        // value.
2100
                        case *lnwire.ErrInvalidNodeAlias:
×
2101
                                idleTimer.Reset(idleTimeout)
×
2102
                                continue
×
2103

2104
                        // If the error we encountered wasn't just a message we
2105
                        // didn't recognize, then we'll stop all processing as
2106
                        // this is a fatal error.
2107
                        default:
3✔
2108
                                break out
3✔
2109
                        }
2110
                }
2111

2112
                // If a message router is active, then we'll try to have it
2113
                // handle this message. If it can, then we're able to skip the
2114
                // rest of the message handling logic.
2115
                err = fn.MapOptionZ(p.msgRouter, func(r msgmux.Router) error {
8✔
2116
                        return r.RouteMsg(msgmux.PeerMsg{
4✔
2117
                                PeerPub: *p.IdentityKey(),
4✔
2118
                                Message: nextMsg,
4✔
2119
                        })
4✔
2120
                })
4✔
2121

2122
                // No error occurred, and the message was handled by the
2123
                // router.
2124
                if err == nil {
7✔
2125
                        continue
3✔
2126
                }
2127

2128
                var (
4✔
2129
                        targetChan   lnwire.ChannelID
4✔
2130
                        isLinkUpdate bool
4✔
2131
                )
4✔
2132

4✔
2133
                switch msg := nextMsg.(type) {
4✔
2134
                case *lnwire.Pong:
×
2135
                        // When we receive a Pong message in response to our
×
2136
                        // last ping message, we send it to the pingManager
×
2137
                        p.pingManager.ReceivedPong(msg)
×
2138

2139
                case *lnwire.Ping:
×
2140
                        // First, we'll store their latest ping payload within
×
2141
                        // the relevant atomic variable.
×
2142
                        p.lastPingPayload.Store(msg.PaddingBytes[:])
×
2143

×
2144
                        // Next, we'll send over the amount of specified pong
×
2145
                        // bytes.
×
2146
                        pong := lnwire.NewPong(p.cfg.PongBuf[0:msg.NumPongBytes])
×
2147
                        p.queueMsg(pong, nil)
×
2148

2149
                case *lnwire.OpenChannel,
2150
                        *lnwire.AcceptChannel,
2151
                        *lnwire.FundingCreated,
2152
                        *lnwire.FundingSigned,
2153
                        *lnwire.ChannelReady:
3✔
2154

3✔
2155
                        p.cfg.FundingManager.ProcessFundingMsg(msg, p)
3✔
2156

2157
                case *lnwire.Shutdown:
3✔
2158
                        select {
3✔
2159
                        case p.chanCloseMsgs <- &closeMsg{msg.ChannelID, msg}:
3✔
2160
                        case <-p.cg.Done():
×
2161
                                break out
×
2162
                        }
2163
                case *lnwire.ClosingSigned:
3✔
2164
                        select {
3✔
2165
                        case p.chanCloseMsgs <- &closeMsg{msg.ChannelID, msg}:
3✔
2166
                        case <-p.cg.Done():
×
2167
                                break out
×
2168
                        }
2169

2170
                case *lnwire.Warning:
×
2171
                        targetChan = msg.ChanID
×
2172
                        isLinkUpdate = p.handleWarningOrError(targetChan, msg)
×
2173

2174
                case *lnwire.Error:
3✔
2175
                        targetChan = msg.ChanID
3✔
2176
                        isLinkUpdate = p.handleWarningOrError(targetChan, msg)
3✔
2177

2178
                case *lnwire.ChannelReestablish:
3✔
2179
                        targetChan = msg.ChanID
3✔
2180
                        isLinkUpdate = p.hasChannel(targetChan)
3✔
2181

3✔
2182
                        // If we failed to find the link in question, and the
3✔
2183
                        // message received was a channel sync message, then
3✔
2184
                        // this might be a peer trying to resync closed channel.
3✔
2185
                        // In this case we'll try to resend our last channel
3✔
2186
                        // sync message, such that the peer can recover funds
3✔
2187
                        // from the closed channel.
3✔
2188
                        if !isLinkUpdate {
6✔
2189
                                err := p.resendChanSyncMsg(targetChan)
3✔
2190
                                if err != nil {
6✔
2191
                                        // TODO(halseth): send error to peer?
3✔
2192
                                        p.log.Errorf("resend failed: %v",
3✔
2193
                                                err)
3✔
2194
                                }
3✔
2195
                        }
2196

2197
                // For messages that implement the LinkUpdater interface, we
2198
                // will consider them as link updates and send them to
2199
                // chanStream. These messages will be queued inside chanStream
2200
                // if the channel is not active yet.
2201
                case lnwire.LinkUpdater:
3✔
2202
                        targetChan = msg.TargetChanID()
3✔
2203
                        isLinkUpdate = p.hasChannel(targetChan)
3✔
2204

3✔
2205
                        // Log an error if we don't have this channel. This
3✔
2206
                        // means the peer has sent us a message with unknown
3✔
2207
                        // channel ID.
3✔
2208
                        if !isLinkUpdate {
6✔
2209
                                p.log.Errorf("Unknown channel ID: %v found "+
3✔
2210
                                        "in received msg=%s", targetChan,
3✔
2211
                                        nextMsg.MsgType())
3✔
2212
                        }
3✔
2213

2214
                case *lnwire.ChannelUpdate1,
2215
                        *lnwire.ChannelAnnouncement1,
2216
                        *lnwire.NodeAnnouncement1,
2217
                        *lnwire.AnnounceSignatures1,
2218
                        *lnwire.GossipTimestampRange,
2219
                        *lnwire.QueryShortChanIDs,
2220
                        *lnwire.QueryChannelRange,
2221
                        *lnwire.ReplyChannelRange,
2222
                        *lnwire.ReplyShortChanIDsEnd:
3✔
2223

3✔
2224
                        discStream.AddMsg(msg)
3✔
2225

2226
                case *lnwire.Custom:
4✔
2227
                        err := p.handleCustomMessage(msg)
4✔
2228
                        if err != nil {
4✔
2229
                                p.storeError(err)
×
2230
                                p.log.Errorf("%v", err)
×
2231
                        }
×
2232

2233
                default:
×
2234
                        // If the message we received is unknown to us, store
×
2235
                        // the type to track the failure.
×
2236
                        err := fmt.Errorf("unknown message type %v received",
×
2237
                                uint16(msg.MsgType()))
×
2238
                        p.storeError(err)
×
2239

×
2240
                        p.log.Errorf("%v", err)
×
2241
                }
2242

2243
                if isLinkUpdate {
7✔
2244
                        // If this is a channel update, then we need to feed it
3✔
2245
                        // into the channel's in-order message stream.
3✔
2246
                        p.sendLinkUpdateMsg(targetChan, nextMsg)
3✔
2247
                }
3✔
2248

2249
                idleTimer.Reset(idleTimeout)
4✔
2250
        }
2251

2252
        p.Disconnect(errors.New("read handler closed"))
3✔
2253

3✔
2254
        p.log.Trace("readHandler for peer done")
3✔
2255
}
2256

2257
// handleCustomMessage handles the given custom message if a handler is
2258
// registered.
2259
func (p *Brontide) handleCustomMessage(msg *lnwire.Custom) error {
4✔
2260
        if p.cfg.HandleCustomMessage == nil {
4✔
2261
                return fmt.Errorf("no custom message handler for "+
×
2262
                        "message type %v", uint16(msg.MsgType()))
×
2263
        }
×
2264

2265
        return p.cfg.HandleCustomMessage(p.PubKey(), msg)
4✔
2266
}
2267

2268
// isLoadedFromDisk returns true if the provided channel ID is loaded from
2269
// disk.
2270
//
2271
// NOTE: only returns true for pending channels.
2272
func (p *Brontide) isLoadedFromDisk(chanID lnwire.ChannelID) bool {
3✔
2273
        // If this is a newly added channel, no need to reestablish.
3✔
2274
        _, added := p.addedChannels.Load(chanID)
3✔
2275
        if added {
6✔
2276
                return false
3✔
2277
        }
3✔
2278

2279
        // Return false if the channel is unknown.
2280
        channel, ok := p.activeChannels.Load(chanID)
3✔
2281
        if !ok {
3✔
2282
                return false
×
2283
        }
×
2284

2285
        // During startup, we will use a nil value to mark a pending channel
2286
        // that's loaded from disk.
2287
        return channel == nil
3✔
2288
}
2289

2290
// isActiveChannel returns true if the provided channel id is active, otherwise
2291
// returns false.
2292
func (p *Brontide) isActiveChannel(chanID lnwire.ChannelID) bool {
11✔
2293
        // The channel would be nil if,
11✔
2294
        // - the channel doesn't exist, or,
11✔
2295
        // - the channel exists, but is pending. In this case, we don't
11✔
2296
        //   consider this channel active.
11✔
2297
        channel, _ := p.activeChannels.Load(chanID)
11✔
2298

11✔
2299
        return channel != nil
11✔
2300
}
11✔
2301

2302
// isPendingChannel returns true if the provided channel ID is pending, and
2303
// returns false if the channel is active or unknown.
2304
func (p *Brontide) isPendingChannel(chanID lnwire.ChannelID) bool {
9✔
2305
        // Return false if the channel is unknown.
9✔
2306
        channel, ok := p.activeChannels.Load(chanID)
9✔
2307
        if !ok {
15✔
2308
                return false
6✔
2309
        }
6✔
2310

2311
        return channel == nil
6✔
2312
}
2313

2314
// hasChannel returns true if the peer has a pending/active channel specified
2315
// by the channel ID.
2316
func (p *Brontide) hasChannel(chanID lnwire.ChannelID) bool {
3✔
2317
        _, ok := p.activeChannels.Load(chanID)
3✔
2318
        return ok
3✔
2319
}
3✔
2320

2321
// storeError stores an error in our peer's buffer of recent errors with the
2322
// current timestamp. Errors are only stored if we have at least one active
2323
// channel with the peer to mitigate a dos vector where a peer costlessly
2324
// connects to us and spams us with errors.
2325
func (p *Brontide) storeError(err error) {
3✔
2326
        var haveChannels bool
3✔
2327

3✔
2328
        p.activeChannels.Range(func(_ lnwire.ChannelID,
3✔
2329
                channel *lnwallet.LightningChannel) bool {
6✔
2330

3✔
2331
                // Pending channels will be nil in the activeChannels map.
3✔
2332
                if channel == nil {
6✔
2333
                        // Return true to continue the iteration.
3✔
2334
                        return true
3✔
2335
                }
3✔
2336

2337
                haveChannels = true
3✔
2338

3✔
2339
                // Return false to break the iteration.
3✔
2340
                return false
3✔
2341
        })
2342

2343
        // If we do not have any active channels with the peer, we do not store
2344
        // errors as a dos mitigation.
2345
        if !haveChannels {
6✔
2346
                p.log.Trace("no channels with peer, not storing err")
3✔
2347
                return
3✔
2348
        }
3✔
2349

2350
        p.cfg.ErrorBuffer.Add(
3✔
2351
                &TimestampedError{Timestamp: time.Now(), Error: err},
3✔
2352
        )
3✔
2353
}
2354

2355
// handleWarningOrError processes a warning or error msg and returns true if
2356
// msg should be forwarded to the associated channel link. False is returned if
2357
// any necessary forwarding of msg was already handled by this method. If msg is
2358
// an error from a peer with an active channel, we'll store it in memory.
2359
//
2360
// NOTE: This method should only be called from within the readHandler.
2361
func (p *Brontide) handleWarningOrError(chanID lnwire.ChannelID,
2362
        msg lnwire.Message) bool {
3✔
2363

3✔
2364
        if errMsg, ok := msg.(*lnwire.Error); ok {
6✔
2365
                p.storeError(errMsg)
3✔
2366
        }
3✔
2367

2368
        switch {
3✔
2369
        // Connection wide messages should be forwarded to all channel links
2370
        // with this peer.
2371
        case chanID == lnwire.ConnectionWideID:
×
2372
                for _, chanStream := range p.activeMsgStreams {
×
2373
                        chanStream.AddMsg(msg)
×
2374
                }
×
2375

2376
                return false
×
2377

2378
        // If the channel ID for the message corresponds to a pending channel,
2379
        // then the funding manager will handle it.
2380
        case p.cfg.FundingManager.IsPendingChannel(chanID, p):
3✔
2381
                p.cfg.FundingManager.ProcessFundingMsg(msg, p)
3✔
2382
                return false
3✔
2383

2384
        // If not we hand the message to the channel link for this channel.
2385
        case p.isActiveChannel(chanID):
3✔
2386
                return true
3✔
2387

2388
        default:
3✔
2389
                return false
3✔
2390
        }
2391
}
2392

2393
// messageSummary returns a human-readable string that summarizes a
2394
// incoming/outgoing message. Not all messages will have a summary, only those
2395
// which have additional data that can be informative at a glance.
2396
func messageSummary(msg lnwire.Message) string {
3✔
2397
        switch msg := msg.(type) {
3✔
2398
        case *lnwire.Init:
3✔
2399
                // No summary.
3✔
2400
                return ""
3✔
2401

2402
        case *lnwire.OpenChannel:
3✔
2403
                return fmt.Sprintf("temp_chan_id=%x, chain=%v, csv=%v, amt=%v, "+
3✔
2404
                        "push_amt=%v, reserve=%v, flags=%v",
3✔
2405
                        msg.PendingChannelID[:], msg.ChainHash,
3✔
2406
                        msg.CsvDelay, msg.FundingAmount, msg.PushAmount,
3✔
2407
                        msg.ChannelReserve, msg.ChannelFlags)
3✔
2408

2409
        case *lnwire.AcceptChannel:
3✔
2410
                return fmt.Sprintf("temp_chan_id=%x, reserve=%v, csv=%v, num_confs=%v",
3✔
2411
                        msg.PendingChannelID[:], msg.ChannelReserve, msg.CsvDelay,
3✔
2412
                        msg.MinAcceptDepth)
3✔
2413

2414
        case *lnwire.FundingCreated:
3✔
2415
                return fmt.Sprintf("temp_chan_id=%x, chan_point=%v",
3✔
2416
                        msg.PendingChannelID[:], msg.FundingPoint)
3✔
2417

2418
        case *lnwire.FundingSigned:
3✔
2419
                return fmt.Sprintf("chan_id=%v", msg.ChanID)
3✔
2420

2421
        case *lnwire.ChannelReady:
3✔
2422
                return fmt.Sprintf("chan_id=%v, next_point=%x",
3✔
2423
                        msg.ChanID, msg.NextPerCommitmentPoint.SerializeCompressed())
3✔
2424

2425
        case *lnwire.Shutdown:
3✔
2426
                return fmt.Sprintf("chan_id=%v, script=%x", msg.ChannelID,
3✔
2427
                        msg.Address[:])
3✔
2428

2429
        case *lnwire.ClosingComplete:
3✔
2430
                return fmt.Sprintf("chan_id=%v, fee_sat=%v, locktime=%v",
3✔
2431
                        msg.ChannelID, msg.FeeSatoshis, msg.LockTime)
3✔
2432

2433
        case *lnwire.ClosingSig:
3✔
2434
                return fmt.Sprintf("chan_id=%v", msg.ChannelID)
3✔
2435

2436
        case *lnwire.ClosingSigned:
3✔
2437
                return fmt.Sprintf("chan_id=%v, fee_sat=%v", msg.ChannelID,
3✔
2438
                        msg.FeeSatoshis)
3✔
2439

2440
        case *lnwire.UpdateAddHTLC:
3✔
2441
                var blindingPoint []byte
3✔
2442
                msg.BlindingPoint.WhenSome(
3✔
2443
                        func(b tlv.RecordT[lnwire.BlindingPointTlvType,
3✔
2444
                                *btcec.PublicKey]) {
6✔
2445

3✔
2446
                                blindingPoint = b.Val.SerializeCompressed()
3✔
2447
                        },
3✔
2448
                )
2449

2450
                return fmt.Sprintf("chan_id=%v, id=%v, amt=%v, expiry=%v, "+
3✔
2451
                        "hash=%x, blinding_point=%x, custom_records=%v",
3✔
2452
                        msg.ChanID, msg.ID, msg.Amount, msg.Expiry,
3✔
2453
                        msg.PaymentHash[:], blindingPoint, msg.CustomRecords)
3✔
2454

2455
        case *lnwire.UpdateFailHTLC:
3✔
2456
                return fmt.Sprintf("chan_id=%v, id=%v, reason=%x", msg.ChanID,
3✔
2457
                        msg.ID, msg.Reason)
3✔
2458

2459
        case *lnwire.UpdateFulfillHTLC:
3✔
2460
                return fmt.Sprintf("chan_id=%v, id=%v, preimage=%x, "+
3✔
2461
                        "custom_records=%v", msg.ChanID, msg.ID,
3✔
2462
                        msg.PaymentPreimage[:], msg.CustomRecords)
3✔
2463

2464
        case *lnwire.CommitSig:
3✔
2465
                return fmt.Sprintf("chan_id=%v, num_htlcs=%v", msg.ChanID,
3✔
2466
                        len(msg.HtlcSigs))
3✔
2467

2468
        case *lnwire.RevokeAndAck:
3✔
2469
                return fmt.Sprintf("chan_id=%v, rev=%x, next_point=%x",
3✔
2470
                        msg.ChanID, msg.Revocation[:],
3✔
2471
                        msg.NextRevocationKey.SerializeCompressed())
3✔
2472

2473
        case *lnwire.UpdateFailMalformedHTLC:
3✔
2474
                return fmt.Sprintf("chan_id=%v, id=%v, fail_code=%v",
3✔
2475
                        msg.ChanID, msg.ID, msg.FailureCode)
3✔
2476

2477
        case *lnwire.Warning:
×
2478
                return fmt.Sprintf("%v", msg.Warning())
×
2479

2480
        case *lnwire.Error:
3✔
2481
                return fmt.Sprintf("%v", msg.Error())
3✔
2482

2483
        case *lnwire.AnnounceSignatures1:
3✔
2484
                return fmt.Sprintf("chan_id=%v, short_chan_id=%v", msg.ChannelID,
3✔
2485
                        msg.ShortChannelID.ToUint64())
3✔
2486

2487
        case *lnwire.ChannelAnnouncement1:
3✔
2488
                return fmt.Sprintf("chain_hash=%v, short_chan_id=%v",
3✔
2489
                        msg.ChainHash, msg.ShortChannelID.ToUint64())
3✔
2490

2491
        case *lnwire.ChannelUpdate1:
3✔
2492
                return fmt.Sprintf("chain_hash=%v, short_chan_id=%v, "+
3✔
2493
                        "mflags=%v, cflags=%v, update_time=%v", msg.ChainHash,
3✔
2494
                        msg.ShortChannelID.ToUint64(), msg.MessageFlags,
3✔
2495
                        msg.ChannelFlags, time.Unix(int64(msg.Timestamp), 0))
3✔
2496

2497
        case *lnwire.NodeAnnouncement1:
3✔
2498
                return fmt.Sprintf("node=%x, update_time=%v",
3✔
2499
                        msg.NodeID, time.Unix(int64(msg.Timestamp), 0))
3✔
2500

2501
        case *lnwire.Ping:
×
2502
                return fmt.Sprintf("ping_bytes=%x", msg.PaddingBytes[:])
×
2503

2504
        case *lnwire.Pong:
×
2505
                return fmt.Sprintf("len(pong_bytes)=%d", len(msg.PongBytes[:]))
×
2506

2507
        case *lnwire.UpdateFee:
×
2508
                return fmt.Sprintf("chan_id=%v, fee_update_sat=%v",
×
2509
                        msg.ChanID, int64(msg.FeePerKw))
×
2510

2511
        case *lnwire.ChannelReestablish:
3✔
2512
                return fmt.Sprintf("chan_id=%v, next_local_height=%v, "+
3✔
2513
                        "remote_tail_height=%v", msg.ChanID,
3✔
2514
                        msg.NextLocalCommitHeight, msg.RemoteCommitTailHeight)
3✔
2515

2516
        case *lnwire.ReplyShortChanIDsEnd:
3✔
2517
                return fmt.Sprintf("chain_hash=%v, complete=%v", msg.ChainHash,
3✔
2518
                        msg.Complete)
3✔
2519

2520
        case *lnwire.ReplyChannelRange:
3✔
2521
                return fmt.Sprintf("start_height=%v, end_height=%v, "+
3✔
2522
                        "num_chans=%v, encoding=%v", msg.FirstBlockHeight,
3✔
2523
                        msg.LastBlockHeight(), len(msg.ShortChanIDs),
3✔
2524
                        msg.EncodingType)
3✔
2525

2526
        case *lnwire.QueryShortChanIDs:
3✔
2527
                return fmt.Sprintf("chain_hash=%v, encoding=%v, num_chans=%v",
3✔
2528
                        msg.ChainHash, msg.EncodingType, len(msg.ShortChanIDs))
3✔
2529

2530
        case *lnwire.QueryChannelRange:
3✔
2531
                return fmt.Sprintf("chain_hash=%v, start_height=%v, "+
3✔
2532
                        "end_height=%v", msg.ChainHash, msg.FirstBlockHeight,
3✔
2533
                        msg.LastBlockHeight())
3✔
2534

2535
        case *lnwire.GossipTimestampRange:
3✔
2536
                return fmt.Sprintf("chain_hash=%v, first_stamp=%v, "+
3✔
2537
                        "stamp_range=%v", msg.ChainHash,
3✔
2538
                        time.Unix(int64(msg.FirstTimestamp), 0),
3✔
2539
                        msg.TimestampRange)
3✔
2540

2541
        case *lnwire.Stfu:
3✔
2542
                return fmt.Sprintf("chan_id=%v, initiator=%v", msg.ChanID,
3✔
2543
                        msg.Initiator)
3✔
2544

2545
        case *lnwire.Custom:
3✔
2546
                return fmt.Sprintf("type=%d", msg.Type)
3✔
2547
        }
2548

2549
        return fmt.Sprintf("unknown msg type=%T", msg)
×
2550
}
2551

2552
// logWireMessage logs the receipt or sending of particular wire message. This
2553
// function is used rather than just logging the message in order to produce
2554
// less spammy log messages in trace mode by setting the 'Curve" parameter to
2555
// nil. Doing this avoids printing out each of the field elements in the curve
2556
// parameters for secp256k1.
2557
func (p *Brontide) logWireMessage(msg lnwire.Message, read bool) {
20✔
2558
        summaryPrefix := "Received"
20✔
2559
        if !read {
36✔
2560
                summaryPrefix = "Sending"
16✔
2561
        }
16✔
2562

2563
        p.log.Debugf("%v", lnutils.NewLogClosure(func() string {
23✔
2564
                // Debug summary of message.
3✔
2565
                summary := messageSummary(msg)
3✔
2566
                if len(summary) > 0 {
6✔
2567
                        summary = "(" + summary + ")"
3✔
2568
                }
3✔
2569

2570
                preposition := "to"
3✔
2571
                if read {
6✔
2572
                        preposition = "from"
3✔
2573
                }
3✔
2574

2575
                var msgType string
3✔
2576
                if msg.MsgType() < lnwire.CustomTypeStart {
6✔
2577
                        msgType = msg.MsgType().String()
3✔
2578
                } else {
6✔
2579
                        msgType = "custom"
3✔
2580
                }
3✔
2581

2582
                return fmt.Sprintf("%v %v%s %v %s", summaryPrefix,
3✔
2583
                        msgType, summary, preposition, p)
3✔
2584
        }))
2585

2586
        prefix := "readMessage from peer"
20✔
2587
        if !read {
36✔
2588
                prefix = "writeMessage to peer"
16✔
2589
        }
16✔
2590

2591
        p.log.Tracef(prefix+": %v", lnutils.SpewLogClosure(msg))
20✔
2592
}
2593

2594
// writeMessage writes and flushes the target lnwire.Message to the remote peer.
2595
// If the passed message is nil, this method will only try to flush an existing
2596
// message buffered on the connection. It is safe to call this method again
2597
// with a nil message iff a timeout error is returned. This will continue to
2598
// flush the pending message to the wire.
2599
//
2600
// NOTE:
2601
// Besides its usage in Start, this function should not be used elsewhere
2602
// except in writeHandler. If multiple goroutines call writeMessage at the same
2603
// time, panics can occur because WriteMessage and Flush don't use any locking
2604
// internally.
2605
func (p *Brontide) writeMessage(msg lnwire.Message) error {
16✔
2606
        // Only log the message on the first attempt.
16✔
2607
        if msg != nil {
32✔
2608
                p.logWireMessage(msg, false)
16✔
2609
        }
16✔
2610

2611
        noiseConn := p.cfg.Conn
16✔
2612

16✔
2613
        flushMsg := func() error {
32✔
2614
                // Ensure the write deadline is set before we attempt to send
16✔
2615
                // the message.
16✔
2616
                writeDeadline := time.Now().Add(
16✔
2617
                        p.scaleTimeout(writeMessageTimeout),
16✔
2618
                )
16✔
2619
                err := noiseConn.SetWriteDeadline(writeDeadline)
16✔
2620
                if err != nil {
16✔
2621
                        return err
×
2622
                }
×
2623

2624
                // Flush the pending message to the wire. If an error is
2625
                // encountered, e.g. write timeout, the number of bytes written
2626
                // so far will be returned.
2627
                n, err := noiseConn.Flush()
16✔
2628

16✔
2629
                // Record the number of bytes written on the wire, if any.
16✔
2630
                if n > 0 {
19✔
2631
                        atomic.AddUint64(&p.bytesSent, uint64(n))
3✔
2632
                }
3✔
2633

2634
                return err
16✔
2635
        }
2636

2637
        // If the current message has already been serialized, encrypted, and
2638
        // buffered on the underlying connection we will skip straight to
2639
        // flushing it to the wire.
2640
        if msg == nil {
16✔
2641
                return flushMsg()
×
2642
        }
×
2643

2644
        // Otherwise, this is a new message. We'll acquire a write buffer to
2645
        // serialize the message and buffer the ciphertext on the connection.
2646
        err := p.cfg.WritePool.Submit(func(buf *bytes.Buffer) error {
32✔
2647
                // Using a buffer allocated by the write pool, encode the
16✔
2648
                // message directly into the buffer.
16✔
2649
                _, writeErr := lnwire.WriteMessage(buf, msg, 0)
16✔
2650
                if writeErr != nil {
16✔
2651
                        return writeErr
×
2652
                }
×
2653

2654
                // Finally, write the message itself in a single swoop. This
2655
                // will buffer the ciphertext on the underlying connection. We
2656
                // will defer flushing the message until the write pool has been
2657
                // released.
2658
                return noiseConn.WriteMessage(buf.Bytes())
16✔
2659
        })
2660
        if err != nil {
16✔
2661
                return err
×
2662
        }
×
2663

2664
        return flushMsg()
16✔
2665
}
2666

2667
// writeHandler is a goroutine dedicated to reading messages off of an incoming
2668
// queue, and writing them out to the wire. This goroutine coordinates with the
2669
// queueHandler in order to ensure the incoming message queue is quickly
2670
// drained.
2671
//
2672
// NOTE: This method MUST be run as a goroutine.
2673
func (p *Brontide) writeHandler() {
6✔
2674
        // We'll stop the timer after a new messages is sent, and also reset it
6✔
2675
        // after we process the next message.
6✔
2676
        idleTimer := time.AfterFunc(idleTimeout, func() {
6✔
2677
                err := fmt.Errorf("peer %s no write for %s -- disconnecting",
×
2678
                        p, idleTimeout)
×
2679
                p.Disconnect(err)
×
2680
        })
×
2681

2682
        var exitErr error
6✔
2683

6✔
2684
out:
6✔
2685
        for {
16✔
2686
                select {
10✔
2687
                case outMsg := <-p.sendQueue:
7✔
2688
                        // Record the time at which we first attempt to send the
7✔
2689
                        // message.
7✔
2690
                        startTime := time.Now()
7✔
2691

7✔
2692
                retry:
7✔
2693
                        // Write out the message to the socket. If a timeout
2694
                        // error is encountered, we will catch this and retry
2695
                        // after backing off in case the remote peer is just
2696
                        // slow to process messages from the wire.
2697
                        err := p.writeMessage(outMsg.msg)
7✔
2698
                        if nerr, ok := err.(net.Error); ok && nerr.Timeout() {
7✔
2699
                                p.log.Debugf("Write timeout detected for "+
×
2700
                                        "peer, first write for message "+
×
2701
                                        "attempted %v ago",
×
2702
                                        time.Since(startTime))
×
2703

×
2704
                                // If we received a timeout error, this implies
×
2705
                                // that the message was buffered on the
×
2706
                                // connection successfully and that a flush was
×
2707
                                // attempted. We'll set the message to nil so
×
2708
                                // that on a subsequent pass we only try to
×
2709
                                // flush the buffered message, and forgo
×
2710
                                // reserializing or reencrypting it.
×
2711
                                outMsg.msg = nil
×
2712

×
2713
                                goto retry
×
2714
                        }
2715

2716
                        // Message has either been successfully sent or an
2717
                        // unrecoverable error occurred.  Either way, we can
2718
                        // free the memory used to store the message.
2719
                        if bConn, ok := p.cfg.Conn.(*brontide.Conn); ok {
10✔
2720
                                bConn.ClearPendingSend()
3✔
2721
                        }
3✔
2722

2723
                        // The write succeeded, reset the idle timer to prevent
2724
                        // us from disconnecting the peer.
2725
                        if !idleTimer.Stop() {
7✔
2726
                                select {
×
2727
                                case <-idleTimer.C:
×
2728
                                default:
×
2729
                                }
2730
                        }
2731
                        idleTimer.Reset(idleTimeout)
7✔
2732

7✔
2733
                        // If the peer requested a synchronous write, respond
7✔
2734
                        // with the error.
7✔
2735
                        if outMsg.errChan != nil {
11✔
2736
                                outMsg.errChan <- err
4✔
2737
                        }
4✔
2738

2739
                        if err != nil {
7✔
2740
                                exitErr = fmt.Errorf("unable to write "+
×
2741
                                        "message: %v", err)
×
2742
                                break out
×
2743
                        }
2744

2745
                case <-p.cg.Done():
3✔
2746
                        exitErr = lnpeer.ErrPeerExiting
3✔
2747
                        break out
3✔
2748
                }
2749
        }
2750

2751
        // Avoid an exit deadlock by ensuring WaitGroups are decremented before
2752
        // disconnect.
2753
        p.cg.WgDone()
3✔
2754

3✔
2755
        p.Disconnect(exitErr)
3✔
2756

3✔
2757
        p.log.Trace("writeHandler for peer done")
3✔
2758
}
2759

2760
// queueHandler is responsible for accepting messages from outside subsystems
2761
// to be eventually sent out on the wire by the writeHandler.
2762
//
2763
// NOTE: This method MUST be run as a goroutine.
2764
func (p *Brontide) queueHandler() {
6✔
2765
        defer p.cg.WgDone()
6✔
2766

6✔
2767
        // priorityMsgs holds an in order list of messages deemed high-priority
6✔
2768
        // to be added to the sendQueue. This predominately includes messages
6✔
2769
        // from the funding manager and htlcswitch.
6✔
2770
        priorityMsgs := list.New()
6✔
2771

6✔
2772
        // lazyMsgs holds an in order list of messages deemed low-priority to be
6✔
2773
        // added to the sendQueue only after all high-priority messages have
6✔
2774
        // been queued. This predominately includes messages from the gossiper.
6✔
2775
        lazyMsgs := list.New()
6✔
2776

6✔
2777
        for {
20✔
2778
                // Examine the front of the priority queue, if it is empty check
14✔
2779
                // the low priority queue.
14✔
2780
                elem := priorityMsgs.Front()
14✔
2781
                if elem == nil {
25✔
2782
                        elem = lazyMsgs.Front()
11✔
2783
                }
11✔
2784

2785
                if elem != nil {
21✔
2786
                        front := elem.Value.(outgoingMsg)
7✔
2787

7✔
2788
                        // There's an element on the queue, try adding
7✔
2789
                        // it to the sendQueue. We also watch for
7✔
2790
                        // messages on the outgoingQueue, in case the
7✔
2791
                        // writeHandler cannot accept messages on the
7✔
2792
                        // sendQueue.
7✔
2793
                        select {
7✔
2794
                        case p.sendQueue <- front:
7✔
2795
                                if front.priority {
13✔
2796
                                        priorityMsgs.Remove(elem)
6✔
2797
                                } else {
10✔
2798
                                        lazyMsgs.Remove(elem)
4✔
2799
                                }
4✔
2800
                        case msg := <-p.outgoingQueue:
3✔
2801
                                if msg.priority {
6✔
2802
                                        priorityMsgs.PushBack(msg)
3✔
2803
                                } else {
6✔
2804
                                        lazyMsgs.PushBack(msg)
3✔
2805
                                }
3✔
2806
                        case <-p.cg.Done():
×
2807
                                return
×
2808
                        }
2809
                } else {
10✔
2810
                        // If there weren't any messages to send to the
10✔
2811
                        // writeHandler, then we'll accept a new message
10✔
2812
                        // into the queue from outside sub-systems.
10✔
2813
                        select {
10✔
2814
                        case msg := <-p.outgoingQueue:
7✔
2815
                                if msg.priority {
13✔
2816
                                        priorityMsgs.PushBack(msg)
6✔
2817
                                } else {
10✔
2818
                                        lazyMsgs.PushBack(msg)
4✔
2819
                                }
4✔
2820
                        case <-p.cg.Done():
3✔
2821
                                return
3✔
2822
                        }
2823
                }
2824
        }
2825
}
2826

2827
// PingTime returns the estimated ping time to the peer in microseconds.
2828
func (p *Brontide) PingTime() int64 {
3✔
2829
        return p.pingManager.GetPingTimeMicroSeconds()
3✔
2830
}
3✔
2831

2832
// queueMsg adds the lnwire.Message to the back of the high priority send queue.
2833
// If the errChan is non-nil, an error is sent back if the msg failed to queue
2834
// or failed to write, and nil otherwise.
2835
func (p *Brontide) queueMsg(msg lnwire.Message, errChan chan error) {
28✔
2836
        p.queue(true, msg, errChan)
28✔
2837
}
28✔
2838

2839
// queueMsgLazy adds the lnwire.Message to the back of the low priority send
2840
// queue. If the errChan is non-nil, an error is sent back if the msg failed to
2841
// queue or failed to write, and nil otherwise.
2842
func (p *Brontide) queueMsgLazy(msg lnwire.Message, errChan chan error) {
4✔
2843
        p.queue(false, msg, errChan)
4✔
2844
}
4✔
2845

2846
// queue sends a given message to the queueHandler using the passed priority. If
2847
// the errChan is non-nil, an error is sent back if the msg failed to queue or
2848
// failed to write, and nil otherwise.
2849
func (p *Brontide) queue(priority bool, msg lnwire.Message,
2850
        errChan chan error) {
29✔
2851

29✔
2852
        select {
29✔
2853
        case p.outgoingQueue <- outgoingMsg{priority, msg, errChan}:
28✔
2854
        case <-p.cg.Done():
×
2855
                p.log.Tracef("Peer shutting down, could not enqueue msg: %v.",
×
2856
                        lnutils.SpewLogClosure(msg))
×
2857
                if errChan != nil {
×
2858
                        errChan <- lnpeer.ErrPeerExiting
×
2859
                }
×
2860
        }
2861
}
2862

2863
// ChannelSnapshots returns a slice of channel snapshots detailing all
2864
// currently active channels maintained with the remote peer.
2865
func (p *Brontide) ChannelSnapshots() []*channeldb.ChannelSnapshot {
3✔
2866
        snapshots := make(
3✔
2867
                []*channeldb.ChannelSnapshot, 0, p.activeChannels.Len(),
3✔
2868
        )
3✔
2869

3✔
2870
        p.activeChannels.ForEach(func(_ lnwire.ChannelID,
3✔
2871
                activeChan *lnwallet.LightningChannel) error {
6✔
2872

3✔
2873
                // If the activeChan is nil, then we skip it as the channel is
3✔
2874
                // pending.
3✔
2875
                if activeChan == nil {
6✔
2876
                        return nil
3✔
2877
                }
3✔
2878

2879
                // We'll only return a snapshot for channels that are
2880
                // *immediately* available for routing payments over.
2881
                if activeChan.RemoteNextRevocation() == nil {
6✔
2882
                        return nil
3✔
2883
                }
3✔
2884

2885
                snapshot := activeChan.StateSnapshot()
3✔
2886
                snapshots = append(snapshots, snapshot)
3✔
2887

3✔
2888
                return nil
3✔
2889
        })
2890

2891
        return snapshots
3✔
2892
}
2893

2894
// genDeliveryScript returns a new script to be used to send our funds to in
2895
// the case of a cooperative channel close negotiation.
2896
func (p *Brontide) genDeliveryScript() ([]byte, error) {
9✔
2897
        // We'll send a normal p2wkh address unless we've negotiated the
9✔
2898
        // shutdown-any-segwit feature.
9✔
2899
        addrType := lnwallet.WitnessPubKey
9✔
2900
        if p.taprootShutdownAllowed() {
12✔
2901
                addrType = lnwallet.TaprootPubkey
3✔
2902
        }
3✔
2903

2904
        deliveryAddr, err := p.cfg.Wallet.NewAddress(
9✔
2905
                addrType, false, lnwallet.DefaultAccountName,
9✔
2906
        )
9✔
2907
        if err != nil {
9✔
2908
                return nil, err
×
2909
        }
×
2910
        p.log.Infof("Delivery addr for channel close: %v",
9✔
2911
                deliveryAddr)
9✔
2912

9✔
2913
        return txscript.PayToAddrScript(deliveryAddr)
9✔
2914
}
2915

2916
// channelManager is goroutine dedicated to handling all requests/signals
2917
// pertaining to the opening, cooperative closing, and force closing of all
2918
// channels maintained with the remote peer.
2919
//
2920
// NOTE: This method MUST be run as a goroutine.
2921
func (p *Brontide) channelManager() {
20✔
2922
        defer p.cg.WgDone()
20✔
2923

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

20✔
2929
out:
20✔
2930
        for {
61✔
2931
                select {
41✔
2932
                // A new pending channel has arrived which means we are about
2933
                // to complete a funding workflow and is waiting for the final
2934
                // `ChannelReady` messages to be exchanged. We will add this
2935
                // channel to the `activeChannels` with a nil value to indicate
2936
                // this is a pending channel.
2937
                case req := <-p.newPendingChannel:
4✔
2938
                        p.handleNewPendingChannel(req)
4✔
2939

2940
                // A new channel has arrived which means we've just completed a
2941
                // funding workflow. We'll initialize the necessary local
2942
                // state, and notify the htlc switch of a new link.
2943
                case req := <-p.newActiveChannel:
3✔
2944
                        p.handleNewActiveChannel(req)
3✔
2945

2946
                // The funding flow for a pending channel is failed, we will
2947
                // remove it from Brontide.
2948
                case req := <-p.removePendingChannel:
4✔
2949
                        p.handleRemovePendingChannel(req)
4✔
2950

2951
                // We've just received a local request to close an active
2952
                // channel. It will either kick of a cooperative channel
2953
                // closure negotiation, or be a notification of a breached
2954
                // contract that should be abandoned.
2955
                case req := <-p.localCloseChanReqs:
10✔
2956
                        p.handleLocalCloseReq(req)
10✔
2957

2958
                // We've received a link failure from a link that was added to
2959
                // the switch. This will initiate the teardown of the link, and
2960
                // initiate any on-chain closures if necessary.
2961
                case failure := <-p.linkFailures:
3✔
2962
                        p.handleLinkFailure(failure)
3✔
2963

2964
                // We've received a new cooperative channel closure related
2965
                // message from the remote peer, we'll use this message to
2966
                // advance the chan closer state machine.
2967
                case closeMsg := <-p.chanCloseMsgs:
16✔
2968
                        p.handleCloseMsg(closeMsg)
16✔
2969

2970
                // The channel reannounce delay has elapsed, broadcast the
2971
                // reenabled channel updates to the network. This should only
2972
                // fire once, so we set the reenableTimeout channel to nil to
2973
                // mark it for garbage collection. If the peer is torn down
2974
                // before firing, reenabling will not be attempted.
2975
                // TODO(conner): consolidate reenables timers inside chan status
2976
                // manager
2977
                case <-reenableTimeout:
3✔
2978
                        p.reenableActiveChannels()
3✔
2979

3✔
2980
                        // Since this channel will never fire again during the
3✔
2981
                        // lifecycle of the peer, we nil the channel to mark it
3✔
2982
                        // eligible for garbage collection, and make this
3✔
2983
                        // explicitly ineligible to receive in future calls to
3✔
2984
                        // select. This also shaves a few CPU cycles since the
3✔
2985
                        // select will ignore this case entirely.
3✔
2986
                        reenableTimeout = nil
3✔
2987

3✔
2988
                        // Once the reenabling is attempted, we also cancel the
3✔
2989
                        // channel event subscription to free up the overflow
3✔
2990
                        // queue used in channel notifier.
3✔
2991
                        //
3✔
2992
                        // NOTE: channelEventClient will be nil if the
3✔
2993
                        // reenableTimeout is greater than 1 minute.
3✔
2994
                        if p.channelEventClient != nil {
6✔
2995
                                p.channelEventClient.Cancel()
3✔
2996
                        }
3✔
2997

2998
                case <-p.cg.Done():
3✔
2999
                        // As, we've been signalled to exit, we'll reset all
3✔
3000
                        // our active channel back to their default state.
3✔
3001
                        p.activeChannels.ForEach(func(_ lnwire.ChannelID,
3✔
3002
                                lc *lnwallet.LightningChannel) error {
6✔
3003

3✔
3004
                                // Exit if the channel is nil as it's a pending
3✔
3005
                                // channel.
3✔
3006
                                if lc == nil {
6✔
3007
                                        return nil
3✔
3008
                                }
3✔
3009

3010
                                lc.ResetState()
3✔
3011

3✔
3012
                                return nil
3✔
3013
                        })
3014

3015
                        break out
3✔
3016
                }
3017
        }
3018
}
3019

3020
// reenableActiveChannels searches the index of channels maintained with this
3021
// peer, and reenables each public, non-pending channel. This is done at the
3022
// gossip level by broadcasting a new ChannelUpdate with the disabled bit unset.
3023
// No message will be sent if the channel is already enabled.
3024
func (p *Brontide) reenableActiveChannels() {
3✔
3025
        // First, filter all known channels with this peer for ones that are
3✔
3026
        // both public and not pending.
3✔
3027
        activePublicChans := p.filterChannelsToEnable()
3✔
3028

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

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

3✔
3038
                switch {
3✔
3039
                // No error occurred, continue to request the next channel.
3040
                case err == nil:
3✔
3041
                        continue
3✔
3042

3043
                // Cannot auto enable a manually disabled channel so we do
3044
                // nothing but proceed to the next channel.
3045
                case errors.Is(err, netann.ErrEnableManuallyDisabledChan):
3✔
3046
                        p.log.Debugf("Channel(%v) was manually disabled, "+
3✔
3047
                                "ignoring automatic enable request", chanPoint)
3✔
3048

3✔
3049
                        continue
3✔
3050

3051
                // If the channel is reported as inactive, we will give it
3052
                // another chance. When handling the request, ChanStatusManager
3053
                // will check whether the link is active or not. One of the
3054
                // conditions is whether the link has been marked as
3055
                // reestablished, which happens inside a goroutine(htlcManager)
3056
                // after the link is started. And we may get a false negative
3057
                // saying the link is not active because that goroutine hasn't
3058
                // reached the line to mark the reestablishment. Thus we give
3059
                // it a second chance to send the request.
3060
                case errors.Is(err, netann.ErrEnableInactiveChan):
×
3061
                        // If we don't have a client created, it means we
×
3062
                        // shouldn't retry enabling the channel.
×
3063
                        if p.channelEventClient == nil {
×
3064
                                p.log.Errorf("Channel(%v) request enabling "+
×
3065
                                        "failed due to inactive link",
×
3066
                                        chanPoint)
×
3067

×
3068
                                continue
×
3069
                        }
3070

3071
                        p.log.Warnf("Channel(%v) cannot be enabled as " +
×
3072
                                "ChanStatusManager reported inactive, retrying")
×
3073

×
3074
                        // Add the channel to the retry map.
×
3075
                        retryChans[chanPoint] = struct{}{}
×
3076
                }
3077
        }
3078

3079
        // Retry the channels if we have any.
3080
        if len(retryChans) != 0 {
3✔
3081
                p.retryRequestEnable(retryChans)
×
3082
        }
×
3083
}
3084

3085
// fetchActiveChanCloser attempts to fetch the active chan closer state machine
3086
// for the target channel ID. If the channel isn't active an error is returned.
3087
// Otherwise, either an existing state machine will be returned, or a new one
3088
// will be created.
3089
func (p *Brontide) fetchActiveChanCloser(chanID lnwire.ChannelID) (
3090
        *chanCloserFsm, error) {
16✔
3091

16✔
3092
        chanCloser, found := p.activeChanCloses.Load(chanID)
16✔
3093
        if found {
29✔
3094
                // An entry will only be found if the closer has already been
13✔
3095
                // created for a non-pending channel or for a channel that had
13✔
3096
                // previously started the shutdown process but the connection
13✔
3097
                // was restarted.
13✔
3098
                return &chanCloser, nil
13✔
3099
        }
13✔
3100

3101
        // First, we'll ensure that we actually know of the target channel. If
3102
        // not, we'll ignore this message.
3103
        channel, ok := p.activeChannels.Load(chanID)
6✔
3104

6✔
3105
        // If the channel isn't in the map or the channel is nil, return
6✔
3106
        // ErrChannelNotFound as the channel is pending.
6✔
3107
        if !ok || channel == nil {
9✔
3108
                return nil, ErrChannelNotFound
3✔
3109
        }
3✔
3110

3111
        // We'll create a valid closing state machine in order to respond to
3112
        // the initiated cooperative channel closure. First, we set the
3113
        // delivery script that our funds will be paid out to. If an upfront
3114
        // shutdown script was set, we will use it. Otherwise, we get a fresh
3115
        // delivery script.
3116
        //
3117
        // TODO: Expose option to allow upfront shutdown script from watch-only
3118
        // accounts.
3119
        deliveryScript := channel.LocalUpfrontShutdownScript()
6✔
3120
        if len(deliveryScript) == 0 {
12✔
3121
                var err error
6✔
3122
                deliveryScript, err = p.genDeliveryScript()
6✔
3123
                if err != nil {
6✔
3124
                        p.log.Errorf("unable to gen delivery script: %v",
×
3125
                                err)
×
3126
                        return nil, fmt.Errorf("close addr unavailable")
×
3127
                }
×
3128
        }
3129

3130
        // In order to begin fee negotiations, we'll first compute our target
3131
        // ideal fee-per-kw.
3132
        feePerKw, err := p.cfg.FeeEstimator.EstimateFeePerKW(
6✔
3133
                p.cfg.CoopCloseTargetConfs,
6✔
3134
        )
6✔
3135
        if err != nil {
6✔
3136
                p.log.Errorf("unable to query fee estimator: %v", err)
×
3137
                return nil, fmt.Errorf("unable to estimate fee")
×
3138
        }
×
3139

3140
        addr, err := p.addrWithInternalKey(deliveryScript)
6✔
3141
        if err != nil {
6✔
3142
                return nil, fmt.Errorf("unable to parse addr: %w", err)
×
3143
        }
×
3144
        negotiateChanCloser, err := p.createChanCloser(
6✔
3145
                channel, addr, feePerKw, nil, lntypes.Remote,
6✔
3146
        )
6✔
3147
        if err != nil {
6✔
3148
                p.log.Errorf("unable to create chan closer: %v", err)
×
3149
                return nil, fmt.Errorf("unable to create chan closer")
×
3150
        }
×
3151

3152
        chanCloser = makeNegotiateCloser(negotiateChanCloser)
6✔
3153

6✔
3154
        p.activeChanCloses.Store(chanID, chanCloser)
6✔
3155

6✔
3156
        return &chanCloser, nil
6✔
3157
}
3158

3159
// filterChannelsToEnable filters a list of channels to be enabled upon start.
3160
// The filtered channels are active channels that's neither private nor
3161
// pending.
3162
func (p *Brontide) filterChannelsToEnable() []wire.OutPoint {
3✔
3163
        var activePublicChans []wire.OutPoint
3✔
3164

3✔
3165
        p.activeChannels.Range(func(chanID lnwire.ChannelID,
3✔
3166
                lnChan *lnwallet.LightningChannel) bool {
6✔
3167

3✔
3168
                // If the lnChan is nil, continue as this is a pending channel.
3✔
3169
                if lnChan == nil {
5✔
3170
                        return true
2✔
3171
                }
2✔
3172

3173
                dbChan := lnChan.State()
3✔
3174
                isPublic := dbChan.ChannelFlags&lnwire.FFAnnounceChannel != 0
3✔
3175
                if !isPublic || dbChan.IsPending {
3✔
3176
                        return true
×
3177
                }
×
3178

3179
                // We'll also skip any channels added during this peer's
3180
                // lifecycle since they haven't waited out the timeout. Their
3181
                // first announcement will be enabled, and the chan status
3182
                // manager will begin monitoring them passively since they exist
3183
                // in the database.
3184
                if _, ok := p.addedChannels.Load(chanID); ok {
4✔
3185
                        return true
1✔
3186
                }
1✔
3187

3188
                activePublicChans = append(
3✔
3189
                        activePublicChans, dbChan.FundingOutpoint,
3✔
3190
                )
3✔
3191

3✔
3192
                return true
3✔
3193
        })
3194

3195
        return activePublicChans
3✔
3196
}
3197

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

×
3205
        // retryEnable is a helper closure that sends an enable request and
×
3206
        // removes the channel from the map if it's matched.
×
3207
        retryEnable := func(chanPoint wire.OutPoint) error {
×
3208
                // If this is an active channel event, check whether it's in
×
3209
                // our targeted channels map.
×
3210
                _, found := activeChans[chanPoint]
×
3211

×
3212
                // If this channel is irrelevant, return nil so the loop can
×
3213
                // jump to next iteration.
×
3214
                if !found {
×
3215
                        return nil
×
3216
                }
×
3217

3218
                // Otherwise we've just received an active signal for a channel
3219
                // that's previously failed to be enabled, we send the request
3220
                // again.
3221
                //
3222
                // We only give the channel one more shot, so we delete it from
3223
                // our map first to keep it from being attempted again.
3224
                delete(activeChans, chanPoint)
×
3225

×
3226
                // Send the request.
×
3227
                err := p.cfg.ChanStatusMgr.RequestEnable(chanPoint, false)
×
3228
                if err != nil {
×
3229
                        return fmt.Errorf("request enabling channel %v "+
×
3230
                                "failed: %w", chanPoint, err)
×
3231
                }
×
3232

3233
                return nil
×
3234
        }
3235

3236
        for {
×
3237
                // If activeChans is empty, we've done processing all the
×
3238
                // channels.
×
3239
                if len(activeChans) == 0 {
×
3240
                        p.log.Debug("Finished retry enabling channels")
×
3241
                        return
×
3242
                }
×
3243

3244
                select {
×
3245
                // A new event has been sent by the ChannelNotifier. We now
3246
                // check whether it's an active or inactive channel event.
3247
                case e := <-p.channelEventClient.Updates():
×
3248
                        // If this is an active channel event, try enable the
×
3249
                        // channel then jump to the next iteration.
×
3250
                        active, ok := e.(channelnotifier.ActiveChannelEvent)
×
3251
                        if ok {
×
3252
                                chanPoint := *active.ChannelPoint
×
3253

×
3254
                                // If we received an error for this particular
×
3255
                                // channel, we log an error and won't quit as
×
3256
                                // we still want to retry other channels.
×
3257
                                if err := retryEnable(chanPoint); err != nil {
×
3258
                                        p.log.Errorf("Retry failed: %v", err)
×
3259
                                }
×
3260

3261
                                continue
×
3262
                        }
3263

3264
                        // Otherwise check for inactive link event, and jump to
3265
                        // next iteration if it's not.
3266
                        inactive, ok := e.(channelnotifier.InactiveLinkEvent)
×
3267
                        if !ok {
×
3268
                                continue
×
3269
                        }
3270

3271
                        // Found an inactive link event, if this is our
3272
                        // targeted channel, remove it from our map.
3273
                        chanPoint := *inactive.ChannelPoint
×
3274
                        _, found := activeChans[chanPoint]
×
3275
                        if !found {
×
3276
                                continue
×
3277
                        }
3278

3279
                        delete(activeChans, chanPoint)
×
3280
                        p.log.Warnf("Re-enable channel %v failed, received "+
×
3281
                                "inactive link event", chanPoint)
×
3282

3283
                case <-p.cg.Done():
×
3284
                        p.log.Debugf("Peer shutdown during retry enabling")
×
3285
                        return
×
3286
                }
3287
        }
3288
}
3289

3290
// chooseDeliveryScript takes two optionally set shutdown scripts and returns
3291
// a suitable script to close out to. This may be nil if neither script is
3292
// set. If both scripts are set, this function will error if they do not match.
3293
func chooseDeliveryScript(upfront, requested lnwire.DeliveryAddress,
3294
        genDeliveryScript func() ([]byte, error),
3295
) (lnwire.DeliveryAddress, error) {
15✔
3296

15✔
3297
        switch {
15✔
3298
        // If no script was provided, then we'll generate a new delivery script.
3299
        case len(upfront) == 0 && len(requested) == 0:
7✔
3300
                return genDeliveryScript()
7✔
3301

3302
        // If no upfront shutdown script was provided, return the user
3303
        // requested address (which may be nil).
3304
        case len(upfront) == 0:
5✔
3305
                return requested, nil
5✔
3306

3307
        // If an upfront shutdown script was provided, and the user did not
3308
        // request a custom shutdown script, return the upfront address.
3309
        case len(requested) == 0:
5✔
3310
                return upfront, nil
5✔
3311

3312
        // If both an upfront shutdown script and a custom close script were
3313
        // provided, error if the user provided shutdown script does not match
3314
        // the upfront shutdown script (because closing out to a different
3315
        // script would violate upfront shutdown).
3316
        case !bytes.Equal(upfront, requested):
2✔
3317
                return nil, chancloser.ErrUpfrontShutdownScriptMismatch
2✔
3318

3319
        // The user requested script matches the upfront shutdown script, so we
3320
        // can return it without error.
3321
        default:
2✔
3322
                return upfront, nil
2✔
3323
        }
3324
}
3325

3326
// restartCoopClose checks whether we need to restart the cooperative close
3327
// process for a given channel.
3328
func (p *Brontide) restartCoopClose(lnChan *lnwallet.LightningChannel) (
3329
        *lnwire.Shutdown, error) {
3✔
3330

3✔
3331
        isTaprootChan := lnChan.ChanType().IsTaproot()
3✔
3332

3✔
3333
        // If this channel has status ChanStatusCoopBroadcasted and does not
3✔
3334
        // have a closing transaction, then the cooperative close process was
3✔
3335
        // started but never finished. We'll re-create the chanCloser state
3✔
3336
        // machine and resend Shutdown. BOLT#2 requires that we retransmit
3✔
3337
        // Shutdown exactly, but doing so would mean persisting the RPC
3✔
3338
        // provided close script. Instead use the LocalUpfrontShutdownScript
3✔
3339
        // or generate a script.
3✔
3340
        c := lnChan.State()
3✔
3341
        _, err := c.BroadcastedCooperative()
3✔
3342
        if err != nil && err != channeldb.ErrNoCloseTx {
3✔
3343
                // An error other than ErrNoCloseTx was encountered.
×
3344
                return nil, err
×
3345
        } else if err == nil && !p.rbfCoopCloseAllowed() {
3✔
3346
                // This is a channel that doesn't support RBF coop close, and it
×
3347
                // already had a coop close txn broadcast. As a result, we can
×
3348
                // just exit here as all we can do is wait for it to confirm.
×
3349
                return nil, nil
×
3350
        }
×
3351

3352
        chanID := lnwire.NewChanIDFromOutPoint(c.FundingOutpoint)
3✔
3353

3✔
3354
        var deliveryScript []byte
3✔
3355

3✔
3356
        shutdownInfo, err := c.ShutdownInfo()
3✔
3357
        switch {
3✔
3358
        // We have previously stored the delivery script that we need to use
3359
        // in the shutdown message. Re-use this script.
3360
        case err == nil:
3✔
3361
                shutdownInfo.WhenSome(func(info channeldb.ShutdownInfo) {
6✔
3362
                        deliveryScript = info.DeliveryScript.Val
3✔
3363
                })
3✔
3364

3365
        // An error other than ErrNoShutdownInfo was returned
3366
        case !errors.Is(err, channeldb.ErrNoShutdownInfo):
×
3367
                return nil, err
×
3368

3369
        case errors.Is(err, channeldb.ErrNoShutdownInfo):
×
3370
                deliveryScript = c.LocalShutdownScript
×
3371
                if len(deliveryScript) == 0 {
×
3372
                        var err error
×
3373
                        deliveryScript, err = p.genDeliveryScript()
×
3374
                        if err != nil {
×
3375
                                p.log.Errorf("unable to gen delivery script: "+
×
3376
                                        "%v", err)
×
3377

×
3378
                                return nil, fmt.Errorf("close addr unavailable")
×
3379
                        }
×
3380
                }
3381
        }
3382

3383
        // If the new RBF co-op close is negotiated, then we'll init and start
3384
        // that state machine, skipping the steps for the negotiate machine
3385
        // below. We don't support this close type for taproot channels though.
3386
        if p.rbfCoopCloseAllowed() && !isTaprootChan {
6✔
3387
                _, err := p.initRbfChanCloser(lnChan)
3✔
3388
                if err != nil {
3✔
3389
                        return nil, fmt.Errorf("unable to init rbf chan "+
×
3390
                                "closer during restart: %w", err)
×
3391
                }
×
3392

3393
                shutdownDesc := fn.MapOption(
3✔
3394
                        newRestartShutdownInit,
3✔
3395
                )(shutdownInfo)
3✔
3396

3✔
3397
                err = p.startRbfChanCloser(
3✔
3398
                        fn.FlattenOption(shutdownDesc), lnChan.ChannelPoint(),
3✔
3399
                )
3✔
3400

3✔
3401
                return nil, err
3✔
3402
        }
3403

3404
        // Compute an ideal fee.
3405
        feePerKw, err := p.cfg.FeeEstimator.EstimateFeePerKW(
×
3406
                p.cfg.CoopCloseTargetConfs,
×
3407
        )
×
3408
        if err != nil {
×
3409
                p.log.Errorf("unable to query fee estimator: %v", err)
×
3410
                return nil, fmt.Errorf("unable to estimate fee")
×
3411
        }
×
3412

3413
        // Determine whether we or the peer are the initiator of the coop
3414
        // close attempt by looking at the channel's status.
3415
        closingParty := lntypes.Remote
×
3416
        if c.HasChanStatus(channeldb.ChanStatusLocalCloseInitiator) {
×
3417
                closingParty = lntypes.Local
×
3418
        }
×
3419

3420
        addr, err := p.addrWithInternalKey(deliveryScript)
×
3421
        if err != nil {
×
3422
                return nil, fmt.Errorf("unable to parse addr: %w", err)
×
3423
        }
×
3424
        chanCloser, err := p.createChanCloser(
×
3425
                lnChan, addr, feePerKw, nil, closingParty,
×
3426
        )
×
3427
        if err != nil {
×
3428
                p.log.Errorf("unable to create chan closer: %v", err)
×
3429
                return nil, fmt.Errorf("unable to create chan closer")
×
3430
        }
×
3431

3432
        p.activeChanCloses.Store(chanID, makeNegotiateCloser(chanCloser))
×
3433

×
3434
        // Create the Shutdown message.
×
3435
        shutdownMsg, err := chanCloser.ShutdownChan()
×
3436
        if err != nil {
×
3437
                p.log.Errorf("unable to create shutdown message: %v", err)
×
3438
                p.activeChanCloses.Delete(chanID)
×
3439
                return nil, err
×
3440
        }
×
3441

3442
        return shutdownMsg, nil
×
3443
}
3444

3445
// createChanCloser constructs a ChanCloser from the passed parameters and is
3446
// used to de-duplicate code.
3447
func (p *Brontide) createChanCloser(channel *lnwallet.LightningChannel,
3448
        deliveryScript *chancloser.DeliveryAddrWithKey,
3449
        fee chainfee.SatPerKWeight, req *htlcswitch.ChanClose,
3450
        closer lntypes.ChannelParty) (*chancloser.ChanCloser, error) {
12✔
3451

12✔
3452
        _, startingHeight, err := p.cfg.ChainIO.GetBestBlock()
12✔
3453
        if err != nil {
12✔
3454
                p.log.Errorf("unable to obtain best block: %v", err)
×
3455
                return nil, fmt.Errorf("cannot obtain best block")
×
3456
        }
×
3457

3458
        // The req will only be set if we initiated the co-op closing flow.
3459
        var maxFee chainfee.SatPerKWeight
12✔
3460
        if req != nil {
21✔
3461
                maxFee = req.MaxFee
9✔
3462
        }
9✔
3463

3464
        chanCloser := chancloser.NewChanCloser(
12✔
3465
                chancloser.ChanCloseCfg{
12✔
3466
                        Channel:      channel,
12✔
3467
                        MusigSession: NewMusigChanCloser(channel),
12✔
3468
                        FeeEstimator: &chancloser.SimpleCoopFeeEstimator{},
12✔
3469
                        BroadcastTx:  p.cfg.Wallet.PublishTransaction,
12✔
3470
                        AuxCloser:    p.cfg.AuxChanCloser,
12✔
3471
                        DisableChannel: func(op wire.OutPoint) error {
24✔
3472
                                return p.cfg.ChanStatusMgr.RequestDisable(
12✔
3473
                                        op, false,
12✔
3474
                                )
12✔
3475
                        },
12✔
3476
                        MaxFee: maxFee,
3477
                        Disconnect: func() error {
×
3478
                                return p.cfg.DisconnectPeer(p.IdentityKey())
×
3479
                        },
×
3480
                        ChainParams: &p.cfg.Wallet.Cfg.NetParams,
3481
                },
3482
                *deliveryScript,
3483
                fee,
3484
                uint32(startingHeight),
3485
                req,
3486
                closer,
3487
        )
3488

3489
        return chanCloser, nil
12✔
3490
}
3491

3492
// initNegotiateChanCloser initializes the channel closer for a channel that is
3493
// using the original "negotiation" based protocol. This path is used when
3494
// we're the one initiating the channel close.
3495
//
3496
// TODO(roasbeef): can make a MsgEndpoint for existing handling logic to
3497
// further abstract.
3498
func (p *Brontide) initNegotiateChanCloser(req *htlcswitch.ChanClose,
3499
        channel *lnwallet.LightningChannel) error {
10✔
3500

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

10✔
3504
        // An upfront shutdown and user provided script are both optional, but
10✔
3505
        // must be equal if both set  (because we cannot serve a request to
10✔
3506
        // close out to a script which violates upfront shutdown). Get the
10✔
3507
        // appropriate address to close out to (which may be nil if neither are
10✔
3508
        // set) and error if they are both set and do not match.
10✔
3509
        deliveryScript, err := chooseDeliveryScript(
10✔
3510
                channel.LocalUpfrontShutdownScript(), req.DeliveryScript,
10✔
3511
                p.genDeliveryScript,
10✔
3512
        )
10✔
3513
        if err != nil {
11✔
3514
                return fmt.Errorf("cannot close channel %v: %w",
1✔
3515
                        req.ChanPoint, err)
1✔
3516
        }
1✔
3517

3518
        addr, err := p.addrWithInternalKey(deliveryScript)
9✔
3519
        if err != nil {
9✔
3520
                return fmt.Errorf("unable to parse addr for channel "+
×
3521
                        "%v: %w", req.ChanPoint, err)
×
3522
        }
×
3523

3524
        chanCloser, err := p.createChanCloser(
9✔
3525
                channel, addr, req.TargetFeePerKw, req, lntypes.Local,
9✔
3526
        )
9✔
3527
        if err != nil {
9✔
3528
                return fmt.Errorf("unable to make chan closer: %w", err)
×
3529
        }
×
3530

3531
        chanID := lnwire.NewChanIDFromOutPoint(channel.ChannelPoint())
9✔
3532
        p.activeChanCloses.Store(chanID, makeNegotiateCloser(chanCloser))
9✔
3533

9✔
3534
        // Finally, we'll initiate the channel shutdown within the
9✔
3535
        // chanCloser, and send the shutdown message to the remote
9✔
3536
        // party to kick things off.
9✔
3537
        shutdownMsg, err := chanCloser.ShutdownChan()
9✔
3538
        if err != nil {
9✔
3539
                // As we were unable to shutdown the channel, we'll return it
×
3540
                // back to its normal state.
×
3541
                defer channel.ResetState()
×
3542

×
3543
                p.activeChanCloses.Delete(chanID)
×
3544

×
3545
                return fmt.Errorf("unable to shutdown channel: %w", err)
×
3546
        }
×
3547

3548
        link := p.fetchLinkFromKeyAndCid(chanID)
9✔
3549
        if link == nil {
9✔
3550
                // If the link is nil then it means it was already removed from
×
3551
                // the switch or it never existed in the first place. The
×
3552
                // latter case is handled at the beginning of this function, so
×
3553
                // in the case where it has already been removed, we can skip
×
3554
                // adding the commit hook to queue a Shutdown message.
×
3555
                p.log.Warnf("link not found during attempted closure: "+
×
3556
                        "%v", chanID)
×
3557
                return nil
×
3558
        }
×
3559

3560
        if !link.DisableAdds(htlcswitch.Outgoing) {
9✔
3561
                p.log.Warnf("Outgoing link adds already "+
×
3562
                        "disabled: %v", link.ChanID())
×
3563
        }
×
3564

3565
        link.OnCommitOnce(htlcswitch.Outgoing, func() {
18✔
3566
                p.queueMsg(shutdownMsg, nil)
9✔
3567
        })
9✔
3568

3569
        return nil
9✔
3570
}
3571

3572
// chooseAddr returns the provided address if it is non-zero length, otherwise
3573
// None.
3574
func chooseAddr(addr lnwire.DeliveryAddress) fn.Option[lnwire.DeliveryAddress] {
3✔
3575
        if len(addr) == 0 {
6✔
3576
                return fn.None[lnwire.DeliveryAddress]()
3✔
3577
        }
3✔
3578

3579
        return fn.Some(addr)
×
3580
}
3581

3582
// observeRbfCloseUpdates observes the channel for any updates that may
3583
// indicate that a new txid has been broadcasted, or the channel fully closed
3584
// on chain.
3585
func (p *Brontide) observeRbfCloseUpdates(chanCloser *chancloser.RbfChanCloser,
3586
        closeReq *htlcswitch.ChanClose,
3587
        coopCloseStates chancloser.RbfStateSub) {
3✔
3588

3✔
3589
        newStateChan := coopCloseStates.NewItemCreated.ChanOut()
3✔
3590
        defer chanCloser.RemoveStateSub(coopCloseStates)
3✔
3591

3✔
3592
        var (
3✔
3593
                lastTxids    lntypes.Dual[chainhash.Hash]
3✔
3594
                lastFeeRates lntypes.Dual[chainfee.SatPerVByte]
3✔
3595
        )
3✔
3596

3✔
3597
        maybeNotifyTxBroadcast := func(state chancloser.AsymmetricPeerState,
3✔
3598
                party lntypes.ChannelParty) {
6✔
3599

3✔
3600
                // First, check to see if we have an error to report to the
3✔
3601
                // caller. If so, then we''ll return that error and exit, as the
3✔
3602
                // stream will exit as well.
3✔
3603
                if closeErr, ok := state.(*chancloser.CloseErr); ok {
6✔
3604
                        // We hit an error during the last state transition, so
3✔
3605
                        // we'll extract the error then send it to the
3✔
3606
                        // user.
3✔
3607
                        err := closeErr.Err()
3✔
3608

3✔
3609
                        peerLog.Warnf("ChannelPoint(%v): encountered close "+
3✔
3610
                                "err: %v", closeReq.ChanPoint, err)
3✔
3611

3✔
3612
                        select {
3✔
3613
                        case closeReq.Err <- err:
3✔
3614
                        case <-closeReq.Ctx.Done():
×
3615
                        case <-p.cg.Done():
×
3616
                        }
3617

3618
                        return
3✔
3619
                }
3620

3621
                closePending, ok := state.(*chancloser.ClosePending)
3✔
3622

3✔
3623
                // If this isn't the close pending state, we aren't at the
3✔
3624
                // terminal state yet.
3✔
3625
                if !ok {
6✔
3626
                        return
3✔
3627
                }
3✔
3628

3629
                // Only notify if the fee rate is greater.
3630
                newFeeRate := closePending.FeeRate
3✔
3631
                lastFeeRate := lastFeeRates.GetForParty(party)
3✔
3632
                if newFeeRate <= lastFeeRate {
6✔
3633
                        peerLog.Debugf("ChannelPoint(%v): remote party made "+
3✔
3634
                                "update for fee rate %v, but we already have "+
3✔
3635
                                "a higher fee rate of %v", closeReq.ChanPoint,
3✔
3636
                                newFeeRate, lastFeeRate)
3✔
3637

3✔
3638
                        return
3✔
3639
                }
3✔
3640

3641
                feeRate := closePending.FeeRate
3✔
3642
                lastFeeRates.SetForParty(party, feeRate)
3✔
3643

3✔
3644
                // At this point, we'll have a txid that we can use to notify
3✔
3645
                // the client, but only if it's different from the last one we
3✔
3646
                // sent. If the user attempted to bump, but was rejected due to
3✔
3647
                // RBF, then we'll send a redundant update.
3✔
3648
                closingTxid := closePending.CloseTx.TxHash()
3✔
3649
                lastTxid := lastTxids.GetForParty(party)
3✔
3650
                if closeReq != nil && closingTxid != lastTxid {
6✔
3651
                        select {
3✔
3652
                        case closeReq.Updates <- &PendingUpdate{
3653
                                Txid:        closingTxid[:],
3654
                                FeePerVbyte: fn.Some(closePending.FeeRate),
3655
                                IsLocalCloseTx: fn.Some(
3656
                                        party == lntypes.Local,
3657
                                ),
3658
                        }:
3✔
3659

3660
                        case <-closeReq.Ctx.Done():
×
3661
                                return
×
3662

3663
                        case <-p.cg.Done():
×
3664
                                return
×
3665
                        }
3666
                }
3667

3668
                lastTxids.SetForParty(party, closingTxid)
3✔
3669
        }
3670

3671
        peerLog.Infof("Observing RBF close updates for channel %v",
3✔
3672
                closeReq.ChanPoint)
3✔
3673

3✔
3674
        // We'll consume each new incoming state to send out the appropriate
3✔
3675
        // RPC update.
3✔
3676
        for {
6✔
3677
                select {
3✔
3678
                case newState := <-newStateChan:
3✔
3679

3✔
3680
                        switch closeState := newState.(type) {
3✔
3681
                        // Once we've reached the state of pending close, we
3682
                        // have a txid that we broadcasted.
3683
                        case *chancloser.ClosingNegotiation:
3✔
3684
                                peerState := closeState.PeerState
3✔
3685

3✔
3686
                                // Each side may have gained a new co-op close
3✔
3687
                                // tx, so we'll examine both to see if they've
3✔
3688
                                // changed.
3✔
3689
                                maybeNotifyTxBroadcast(
3✔
3690
                                        peerState.GetForParty(lntypes.Local),
3✔
3691
                                        lntypes.Local,
3✔
3692
                                )
3✔
3693
                                maybeNotifyTxBroadcast(
3✔
3694
                                        peerState.GetForParty(lntypes.Remote),
3✔
3695
                                        lntypes.Remote,
3✔
3696
                                )
3✔
3697

3698
                        // Otherwise, if we're transition to CloseFin, then we
3699
                        // know that we're done.
3700
                        case *chancloser.CloseFin:
3✔
3701
                                // To clean up, we'll remove the chan closer
3✔
3702
                                // from the active map, and send the final
3✔
3703
                                // update to the client.
3✔
3704
                                closingTxid := closeState.ConfirmedTx.TxHash()
3✔
3705
                                if closeReq != nil {
6✔
3706
                                        closeReq.Updates <- &ChannelCloseUpdate{
3✔
3707
                                                ClosingTxid: closingTxid[:],
3✔
3708
                                                Success:     true,
3✔
3709
                                        }
3✔
3710
                                }
3✔
3711
                                chanID := lnwire.NewChanIDFromOutPoint(
3✔
3712
                                        *closeReq.ChanPoint,
3✔
3713
                                )
3✔
3714
                                p.activeChanCloses.Delete(chanID)
3✔
3715

3✔
3716
                                return
3✔
3717
                        }
3718

3719
                case <-closeReq.Ctx.Done():
3✔
3720
                        return
3✔
3721

3722
                case <-p.cg.Done():
3✔
3723
                        return
3✔
3724
                }
3725
        }
3726
}
3727

3728
// chanErrorReporter is a simple implementation of the
3729
// chancloser.ErrorReporter. This is bound to a single channel by the channel
3730
// ID.
3731
type chanErrorReporter struct {
3732
        chanID lnwire.ChannelID
3733
        peer   *Brontide
3734
}
3735

3736
// newChanErrorReporter creates a new instance of the chanErrorReporter.
3737
func newChanErrorReporter(chanID lnwire.ChannelID,
3738
        peer *Brontide) *chanErrorReporter {
3✔
3739

3✔
3740
        return &chanErrorReporter{
3✔
3741
                chanID: chanID,
3✔
3742
                peer:   peer,
3✔
3743
        }
3✔
3744
}
3✔
3745

3746
// ReportError is a method that's used to report an error that occurred during
3747
// state machine execution. This is used by the RBF close state machine to
3748
// terminate the state machine and send an error to the remote peer.
3749
//
3750
// This is a part of the chancloser.ErrorReporter interface.
3751
func (c *chanErrorReporter) ReportError(chanErr error) {
×
3752
        c.peer.log.Errorf("coop close error for channel %v: %v",
×
3753
                c.chanID, chanErr)
×
3754

×
3755
        var errMsg []byte
×
3756
        if errors.Is(chanErr, chancloser.ErrInvalidStateTransition) {
×
3757
                errMsg = []byte("unexpected protocol message")
×
3758
        } else {
×
3759
                errMsg = []byte(chanErr.Error())
×
3760
        }
×
3761

3762
        err := c.peer.SendMessageLazy(false, &lnwire.Error{
×
3763
                ChanID: c.chanID,
×
3764
                Data:   errMsg,
×
3765
        })
×
3766
        if err != nil {
×
3767
                c.peer.log.Warnf("unable to send error message to peer: %v",
×
3768
                        err)
×
3769
        }
×
3770

3771
        // After we send the error message to the peer, we'll re-initialize the
3772
        // coop close state machine as they may send a shutdown message to
3773
        // retry the coop close.
3774
        lnChan, ok := c.peer.activeChannels.Load(c.chanID)
×
3775
        if !ok {
×
3776
                return
×
3777
        }
×
3778

3779
        if lnChan == nil {
×
3780
                c.peer.log.Debugf("channel %v is pending, not "+
×
3781
                        "re-initializing coop close state machine",
×
3782
                        c.chanID)
×
3783

×
3784
                return
×
3785
        }
×
3786

3787
        if _, err := c.peer.initRbfChanCloser(lnChan); err != nil {
×
3788
                c.peer.activeChanCloses.Delete(c.chanID)
×
3789

×
3790
                c.peer.log.Errorf("unable to init RBF chan closer after "+
×
3791
                        "error case: %v", err)
×
3792
        }
×
3793
}
3794

3795
// chanFlushEventSentinel is used to send the RBF coop close state machine the
3796
// channel flushed event. We'll wait until the state machine enters the
3797
// ChannelFlushing state, then request the link to send the event once flushed.
3798
//
3799
// NOTE: This MUST be run as a goroutine.
3800
func (p *Brontide) chanFlushEventSentinel(chanCloser *chancloser.RbfChanCloser,
3801
        link htlcswitch.ChannelUpdateHandler,
3802
        channel *lnwallet.LightningChannel) {
3✔
3803

3✔
3804
        defer p.cg.WgDone()
3✔
3805

3✔
3806
        // If there's no link, then the channel has already been flushed, so we
3✔
3807
        // don't need to continue.
3✔
3808
        if link == nil {
6✔
3809
                return
3✔
3810
        }
3✔
3811

3812
        coopCloseStates := chanCloser.RegisterStateEvents()
3✔
3813
        defer chanCloser.RemoveStateSub(coopCloseStates)
3✔
3814

3✔
3815
        newStateChan := coopCloseStates.NewItemCreated.ChanOut()
3✔
3816

3✔
3817
        sendChanFlushed := func() {
6✔
3818
                chanState := channel.StateSnapshot()
3✔
3819

3✔
3820
                peerLog.Infof("ChannelPoint(%v) has been flushed for co-op "+
3✔
3821
                        "close, sending event to chan closer",
3✔
3822
                        channel.ChannelPoint())
3✔
3823

3✔
3824
                chanBalances := chancloser.ShutdownBalances{
3✔
3825
                        LocalBalance:  chanState.LocalBalance,
3✔
3826
                        RemoteBalance: chanState.RemoteBalance,
3✔
3827
                }
3✔
3828
                ctx := context.Background()
3✔
3829
                chanCloser.SendEvent(ctx, &chancloser.ChannelFlushed{
3✔
3830
                        ShutdownBalances: chanBalances,
3✔
3831
                        FreshFlush:       true,
3✔
3832
                })
3✔
3833
        }
3✔
3834

3835
        // We'll wait until the channel enters the ChannelFlushing state. We
3836
        // exit after a success loop. As after the first RBF iteration, the
3837
        // channel will always be flushed.
3838
        for {
6✔
3839
                select {
3✔
3840
                case newState, ok := <-newStateChan:
3✔
3841
                        if !ok {
3✔
3842
                                return
×
3843
                        }
×
3844

3845
                        if _, ok := newState.(*chancloser.ChannelFlushing); ok {
6✔
3846
                                peerLog.Infof("ChannelPoint(%v): rbf coop "+
3✔
3847
                                        "close is awaiting a flushed state, "+
3✔
3848
                                        "registering with link..., ",
3✔
3849
                                        channel.ChannelPoint())
3✔
3850

3✔
3851
                                // Request the link to send the event once the
3✔
3852
                                // channel is flushed. We only need this event
3✔
3853
                                // sent once, so we can exit now.
3✔
3854
                                link.OnFlushedOnce(sendChanFlushed)
3✔
3855

3✔
3856
                                return
3✔
3857
                        }
3✔
3858

3859
                case <-p.cg.Done():
3✔
3860
                        return
3✔
3861
                }
3862
        }
3863
}
3864

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

3✔
3871
        chanID := lnwire.NewChanIDFromOutPoint(channel.ChannelPoint())
3✔
3872

3✔
3873
        link := p.fetchLinkFromKeyAndCid(chanID)
3✔
3874

3✔
3875
        _, startingHeight, err := p.cfg.ChainIO.GetBestBlock()
3✔
3876
        if err != nil {
3✔
3877
                return nil, fmt.Errorf("cannot obtain best block: %w", err)
×
3878
        }
×
3879

3880
        defaultFeePerKw, err := p.cfg.FeeEstimator.EstimateFeePerKW(
3✔
3881
                p.cfg.CoopCloseTargetConfs,
3✔
3882
        )
3✔
3883
        if err != nil {
3✔
3884
                return nil, fmt.Errorf("unable to estimate fee: %w", err)
×
3885
        }
×
3886

3887
        thawHeight, err := channel.AbsoluteThawHeight()
3✔
3888
        if err != nil {
3✔
3889
                return nil, fmt.Errorf("unable to get thaw height: %w", err)
×
3890
        }
×
3891

3892
        peerPub := *p.IdentityKey()
3✔
3893

3✔
3894
        msgMapper := chancloser.NewRbfMsgMapper(
3✔
3895
                uint32(startingHeight), chanID, peerPub,
3✔
3896
        )
3✔
3897

3✔
3898
        initialState := chancloser.ChannelActive{}
3✔
3899

3✔
3900
        scid := channel.ZeroConfRealScid().UnwrapOr(
3✔
3901
                channel.ShortChanID(),
3✔
3902
        )
3✔
3903

3✔
3904
        env := chancloser.Environment{
3✔
3905
                ChainParams:    p.cfg.Wallet.Cfg.NetParams,
3✔
3906
                ChanPeer:       peerPub,
3✔
3907
                ChanPoint:      channel.ChannelPoint(),
3✔
3908
                ChanID:         chanID,
3✔
3909
                Scid:           scid,
3✔
3910
                ChanType:       channel.ChanType(),
3✔
3911
                DefaultFeeRate: defaultFeePerKw.FeePerVByte(),
3✔
3912
                ThawHeight:     fn.Some(thawHeight),
3✔
3913
                RemoteUpfrontShutdown: chooseAddr(
3✔
3914
                        channel.RemoteUpfrontShutdownScript(),
3✔
3915
                ),
3✔
3916
                LocalUpfrontShutdown: chooseAddr(
3✔
3917
                        channel.LocalUpfrontShutdownScript(),
3✔
3918
                ),
3✔
3919
                NewDeliveryScript: func() (lnwire.DeliveryAddress, error) {
6✔
3920
                        return p.genDeliveryScript()
3✔
3921
                },
3✔
3922
                FeeEstimator: &chancloser.SimpleCoopFeeEstimator{},
3923
                CloseSigner:  channel,
3924
                ChanObserver: newChanObserver(
3925
                        channel, link, p.cfg.ChanStatusMgr,
3926
                ),
3927
        }
3928

3929
        spendEvent := protofsm.RegisterSpend[chancloser.ProtocolEvent]{
3✔
3930
                OutPoint:   channel.ChannelPoint(),
3✔
3931
                PkScript:   channel.FundingTxOut().PkScript,
3✔
3932
                HeightHint: channel.DeriveHeightHint(),
3✔
3933
                PostSpendEvent: fn.Some[chancloser.RbfSpendMapper](
3✔
3934
                        chancloser.SpendMapper,
3✔
3935
                ),
3✔
3936
        }
3✔
3937

3✔
3938
        daemonAdapters := NewLndDaemonAdapters(LndAdapterCfg{
3✔
3939
                MsgSender:     newPeerMsgSender(peerPub, p),
3✔
3940
                TxBroadcaster: p.cfg.Wallet,
3✔
3941
                ChainNotifier: p.cfg.ChainNotifier,
3✔
3942
        })
3✔
3943

3✔
3944
        protoCfg := chancloser.RbfChanCloserCfg{
3✔
3945
                Daemon:        daemonAdapters,
3✔
3946
                InitialState:  &initialState,
3✔
3947
                Env:           &env,
3✔
3948
                InitEvent:     fn.Some[protofsm.DaemonEvent](&spendEvent),
3✔
3949
                ErrorReporter: newChanErrorReporter(chanID, p),
3✔
3950
                MsgMapper: fn.Some[protofsm.MsgMapper[chancloser.ProtocolEvent]]( //nolint:ll
3✔
3951
                        msgMapper,
3✔
3952
                ),
3✔
3953
        }
3✔
3954

3✔
3955
        ctx := context.Background()
3✔
3956
        chanCloser := protofsm.NewStateMachine(protoCfg)
3✔
3957
        chanCloser.Start(ctx)
3✔
3958

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

3✔
3964
                return r.RegisterEndpoint(&chanCloser)
3✔
3965
        })
3✔
3966
        if err != nil {
3✔
3967
                chanCloser.Stop()
×
3968

×
3969
                return nil, fmt.Errorf("unable to register endpoint for co-op "+
×
3970
                        "close: %w", err)
×
3971
        }
×
3972

3973
        p.activeChanCloses.Store(chanID, makeRbfCloser(&chanCloser))
3✔
3974

3✔
3975
        // Now that we've created the rbf closer state machine, we'll launch a
3✔
3976
        // new goroutine to eventually send in the ChannelFlushed event once
3✔
3977
        // needed.
3✔
3978
        p.cg.WgAdd(1)
3✔
3979
        go p.chanFlushEventSentinel(&chanCloser, link, channel)
3✔
3980

3✔
3981
        return &chanCloser, nil
3✔
3982
}
3983

3984
// shutdownInit describes the two ways we can initiate a new shutdown. Either we
3985
// got an RPC request to do so (left), or we sent a shutdown message to the
3986
// party (for w/e reason), but crashed before the close was complete.
3987
//
3988
//nolint:ll
3989
type shutdownInit = fn.Option[fn.Either[*htlcswitch.ChanClose, channeldb.ShutdownInfo]]
3990

3991
// shutdownStartFeeRate returns the fee rate that should be used for the
3992
// shutdown.  This returns a doubly wrapped option as the shutdown info might
3993
// be none, and the fee rate is only defined for the user initiated shutdown.
3994
func shutdownStartFeeRate(s shutdownInit) fn.Option[chainfee.SatPerKWeight] {
3✔
3995
        feeRateOpt := fn.MapOption(func(init fn.Either[*htlcswitch.ChanClose,
3✔
3996
                channeldb.ShutdownInfo]) fn.Option[chainfee.SatPerKWeight] {
6✔
3997

3✔
3998
                var feeRate fn.Option[chainfee.SatPerKWeight]
3✔
3999
                init.WhenLeft(func(req *htlcswitch.ChanClose) {
6✔
4000
                        feeRate = fn.Some(req.TargetFeePerKw)
3✔
4001
                })
3✔
4002

4003
                return feeRate
3✔
4004
        })(s)
4005

4006
        return fn.FlattenOption(feeRateOpt)
3✔
4007
}
4008

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

3✔
4016
                var addr fn.Option[lnwire.DeliveryAddress]
3✔
4017
                init.WhenLeft(func(req *htlcswitch.ChanClose) {
6✔
4018
                        if len(req.DeliveryScript) != 0 {
6✔
4019
                                addr = fn.Some(req.DeliveryScript)
3✔
4020
                        }
3✔
4021
                })
4022
                init.WhenRight(func(info channeldb.ShutdownInfo) {
6✔
4023
                        addr = fn.Some(info.DeliveryScript.Val)
3✔
4024
                })
3✔
4025

4026
                return addr
3✔
4027
        })(s)
4028

4029
        return fn.FlattenOption(addrOpt)
3✔
4030
}
4031

4032
// whenRPCShutdown registers a callback to be executed when the shutdown init
4033
// type is and RPC request.
4034
func whenRPCShutdown(s shutdownInit, f func(r *htlcswitch.ChanClose)) {
3✔
4035
        s.WhenSome(func(init fn.Either[*htlcswitch.ChanClose,
3✔
4036
                channeldb.ShutdownInfo]) {
6✔
4037

3✔
4038
                init.WhenLeft(f)
3✔
4039
        })
3✔
4040
}
4041

4042
// newRestartShutdownInit creates a new shutdownInit for the case where we need
4043
// to restart the shutdown flow after a restart.
4044
func newRestartShutdownInit(info channeldb.ShutdownInfo) shutdownInit {
3✔
4045
        return fn.Some(fn.NewRight[*htlcswitch.ChanClose](info))
3✔
4046
}
3✔
4047

4048
// newRPCShutdownInit creates a new shutdownInit for the case where we
4049
// initiated the shutdown via an RPC client.
4050
func newRPCShutdownInit(req *htlcswitch.ChanClose) shutdownInit {
3✔
4051
        return fn.Some(
3✔
4052
                fn.NewLeft[*htlcswitch.ChanClose, channeldb.ShutdownInfo](req),
3✔
4053
        )
3✔
4054
}
3✔
4055

4056
// waitUntilRbfCoastClear waits until the RBF co-op close state machine has
4057
// advanced to a terminal state before attempting another fee bump.
4058
func waitUntilRbfCoastClear(ctx context.Context,
4059
        rbfCloser *chancloser.RbfChanCloser) error {
3✔
4060

3✔
4061
        coopCloseStates := rbfCloser.RegisterStateEvents()
3✔
4062
        newStateChan := coopCloseStates.NewItemCreated.ChanOut()
3✔
4063
        defer rbfCloser.RemoveStateSub(coopCloseStates)
3✔
4064

3✔
4065
        isTerminalState := func(newState chancloser.RbfState) bool {
6✔
4066
                // If we're not in the negotiation sub-state, then we aren't at
3✔
4067
                // the terminal state yet.
3✔
4068
                state, ok := newState.(*chancloser.ClosingNegotiation)
3✔
4069
                if !ok {
3✔
4070
                        return false
×
4071
                }
×
4072

4073
                localState := state.PeerState.GetForParty(lntypes.Local)
3✔
4074

3✔
4075
                // If this isn't the close pending state, we aren't at the
3✔
4076
                // terminal state yet.
3✔
4077
                _, ok = localState.(*chancloser.ClosePending)
3✔
4078

3✔
4079
                return ok
3✔
4080
        }
4081

4082
        // Before we enter the subscription loop below, check to see if we're
4083
        // already in the terminal state.
4084
        rbfState, err := rbfCloser.CurrentState()
3✔
4085
        if err != nil {
3✔
4086
                return err
×
4087
        }
×
4088
        if isTerminalState(rbfState) {
6✔
4089
                return nil
3✔
4090
        }
3✔
4091

4092
        peerLog.Debugf("Waiting for RBF iteration to complete...")
×
4093

×
4094
        for {
×
4095
                select {
×
4096
                case newState := <-newStateChan:
×
4097
                        if isTerminalState(newState) {
×
4098
                                return nil
×
4099
                        }
×
4100

4101
                case <-ctx.Done():
×
4102
                        return fmt.Errorf("context canceled")
×
4103
                }
4104
        }
4105
}
4106

4107
// startRbfChanCloser kicks off the co-op close process using the new RBF based
4108
// co-op close protocol. This is called when we're the one that's initiating
4109
// the cooperative channel close.
4110
//
4111
// TODO(roasbeef): just accept the two shutdown pointer params instead??
4112
func (p *Brontide) startRbfChanCloser(shutdown shutdownInit,
4113
        chanPoint wire.OutPoint) error {
3✔
4114

3✔
4115
        // Unlike the old negotiate chan closer, we'll always create the RBF
3✔
4116
        // chan closer on startup, so we can skip init here.
3✔
4117
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
3✔
4118
        chanCloser, found := p.activeChanCloses.Load(chanID)
3✔
4119
        if !found {
3✔
4120
                return fmt.Errorf("rbf chan closer not found for channel %v",
×
4121
                        chanPoint)
×
4122
        }
×
4123

4124
        defaultFeePerKw, err := shutdownStartFeeRate(
3✔
4125
                shutdown,
3✔
4126
        ).UnwrapOrFuncErr(func() (chainfee.SatPerKWeight, error) {
6✔
4127
                return p.cfg.FeeEstimator.EstimateFeePerKW(
3✔
4128
                        p.cfg.CoopCloseTargetConfs,
3✔
4129
                )
3✔
4130
        })
3✔
4131
        if err != nil {
3✔
4132
                return fmt.Errorf("unable to estimate fee: %w", err)
×
4133
        }
×
4134

4135
        chanCloser.WhenRight(func(rbfCloser *chancloser.RbfChanCloser) {
6✔
4136
                peerLog.Infof("ChannelPoint(%v): rbf-coop close requested, "+
3✔
4137
                        "sending shutdown", chanPoint)
3✔
4138

3✔
4139
                rbfState, err := rbfCloser.CurrentState()
3✔
4140
                if err != nil {
3✔
4141
                        peerLog.Warnf("ChannelPoint(%v): unable to get "+
×
4142
                                "current state for rbf-coop close: %v",
×
4143
                                chanPoint, err)
×
4144

×
4145
                        return
×
4146
                }
×
4147

4148
                coopCloseStates := rbfCloser.RegisterStateEvents()
3✔
4149

3✔
4150
                // Before we send our event below, we'll launch a goroutine to
3✔
4151
                // watch for the final terminal state to send updates to the RPC
3✔
4152
                // client. We only need to do this if there's an RPC caller.
3✔
4153
                var rpcShutdown bool
3✔
4154
                whenRPCShutdown(shutdown, func(req *htlcswitch.ChanClose) {
6✔
4155
                        rpcShutdown = true
3✔
4156

3✔
4157
                        p.cg.WgAdd(1)
3✔
4158
                        go func() {
6✔
4159
                                defer p.cg.WgDone()
3✔
4160

3✔
4161
                                p.observeRbfCloseUpdates(
3✔
4162
                                        rbfCloser, req, coopCloseStates,
3✔
4163
                                )
3✔
4164
                        }()
3✔
4165
                })
4166

4167
                if !rpcShutdown {
6✔
4168
                        defer rbfCloser.RemoveStateSub(coopCloseStates)
3✔
4169
                }
3✔
4170

4171
                ctx, _ := p.cg.Create(context.Background())
3✔
4172
                feeRate := defaultFeePerKw.FeePerVByte()
3✔
4173

3✔
4174
                // Depending on the state of the state machine, we'll either
3✔
4175
                // kick things off by sending shutdown, or attempt to send a new
3✔
4176
                // offer to the remote party.
3✔
4177
                switch rbfState.(type) {
3✔
4178
                // The channel is still active, so we'll now kick off the co-op
4179
                // close process by instructing it to send a shutdown message to
4180
                // the remote party.
4181
                case *chancloser.ChannelActive:
3✔
4182
                        rbfCloser.SendEvent(
3✔
4183
                                context.Background(),
3✔
4184
                                &chancloser.SendShutdown{
3✔
4185
                                        IdealFeeRate: feeRate,
3✔
4186
                                        DeliveryAddr: shutdownStartAddr(
3✔
4187
                                                shutdown,
3✔
4188
                                        ),
3✔
4189
                                },
3✔
4190
                        )
3✔
4191

4192
                // If we haven't yet sent an offer (didn't have enough funds at
4193
                // the prior fee rate), or we've sent an offer, then we'll
4194
                // trigger a new offer event.
4195
                case *chancloser.ClosingNegotiation:
3✔
4196
                        // Before we send the event below, we'll wait until
3✔
4197
                        // we're in a semi-terminal state.
3✔
4198
                        err := waitUntilRbfCoastClear(ctx, rbfCloser)
3✔
4199
                        if err != nil {
3✔
4200
                                peerLog.Warnf("ChannelPoint(%v): unable to "+
×
4201
                                        "wait for coast to clear: %v",
×
4202
                                        chanPoint, err)
×
4203

×
4204
                                return
×
4205
                        }
×
4206

4207
                        event := chancloser.ProtocolEvent(
3✔
4208
                                &chancloser.SendOfferEvent{
3✔
4209
                                        TargetFeeRate: feeRate,
3✔
4210
                                },
3✔
4211
                        )
3✔
4212
                        rbfCloser.SendEvent(ctx, event)
3✔
4213

4214
                default:
×
4215
                        peerLog.Warnf("ChannelPoint(%v): unexpected state "+
×
4216
                                "for rbf-coop close: %T", chanPoint, rbfState)
×
4217
                }
4218
        })
4219

4220
        return nil
3✔
4221
}
4222

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

10✔
4228
        channel, ok := p.activeChannels.Load(chanID)
10✔
4229

10✔
4230
        // Though this function can't be called for pending channels, we still
10✔
4231
        // check whether channel is nil for safety.
10✔
4232
        if !ok || channel == nil {
10✔
4233
                err := fmt.Errorf("unable to close channel, ChannelID(%v) is "+
×
4234
                        "unknown", chanID)
×
4235
                p.log.Errorf(err.Error())
×
4236
                req.Err <- err
×
4237
                return
×
4238
        }
×
4239

4240
        isTaprootChan := channel.ChanType().IsTaproot()
10✔
4241

10✔
4242
        switch req.CloseType {
10✔
4243
        // A type of CloseRegular indicates that the user has opted to close
4244
        // out this channel on-chain, so we execute the cooperative channel
4245
        // closure workflow.
4246
        case contractcourt.CloseRegular:
10✔
4247
                var err error
10✔
4248
                switch {
10✔
4249
                // If this is the RBF coop state machine, then we'll instruct
4250
                // it to send the shutdown message. This also might be an RBF
4251
                // iteration, in which case we'll be obtaining a new
4252
                // transaction w/ a higher fee rate.
4253
                //
4254
                // We don't support this close type for taproot channels yet
4255
                // however.
4256
                case !isTaprootChan && p.rbfCoopCloseAllowed():
3✔
4257
                        err = p.startRbfChanCloser(
3✔
4258
                                newRPCShutdownInit(req), channel.ChannelPoint(),
3✔
4259
                        )
3✔
4260
                default:
10✔
4261
                        err = p.initNegotiateChanCloser(req, channel)
10✔
4262
                }
4263

4264
                if err != nil {
11✔
4265
                        p.log.Errorf(err.Error())
1✔
4266
                        req.Err <- err
1✔
4267
                }
1✔
4268

4269
        // A type of CloseBreach indicates that the counterparty has breached
4270
        // the channel therefore we need to clean up our local state.
4271
        case contractcourt.CloseBreach:
1✔
4272
                // TODO(roasbeef): no longer need with newer beach logic?
1✔
4273
                p.log.Infof("ChannelPoint(%v) has been breached, wiping "+
1✔
4274
                        "channel", req.ChanPoint)
1✔
4275
                p.WipeChannel(req.ChanPoint)
1✔
4276
        }
4277
}
4278

4279
// linkFailureReport is sent to the channelManager whenever a link reports a
4280
// link failure, and is forced to exit. The report houses the necessary
4281
// information to clean up the channel state, send back the error message, and
4282
// force close if necessary.
4283
type linkFailureReport struct {
4284
        chanPoint   wire.OutPoint
4285
        chanID      lnwire.ChannelID
4286
        shortChanID lnwire.ShortChannelID
4287
        linkErr     htlcswitch.LinkFailureError
4288
}
4289

4290
// handleLinkFailure processes a link failure report when a link in the switch
4291
// fails. It facilitates the removal of all channel state within the peer,
4292
// force closing the channel depending on severity, and sending the error
4293
// message back to the remote party.
4294
func (p *Brontide) handleLinkFailure(failure linkFailureReport) {
3✔
4295
        // Retrieve the channel from the map of active channels. We do this to
3✔
4296
        // have access to it even after WipeChannel remove it from the map.
3✔
4297
        chanID := lnwire.NewChanIDFromOutPoint(failure.chanPoint)
3✔
4298
        lnChan, _ := p.activeChannels.Load(chanID)
3✔
4299

3✔
4300
        // We begin by wiping the link, which will remove it from the switch,
3✔
4301
        // such that it won't be attempted used for any more updates.
3✔
4302
        //
3✔
4303
        // TODO(halseth): should introduce a way to atomically stop/pause the
3✔
4304
        // link and cancel back any adds in its mailboxes such that we can
3✔
4305
        // safely force close without the link being added again and updates
3✔
4306
        // being applied.
3✔
4307
        p.WipeChannel(&failure.chanPoint)
3✔
4308

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

3✔
4314
                closeTx, err := p.cfg.ChainArb.ForceCloseContract(
3✔
4315
                        failure.chanPoint,
3✔
4316
                )
3✔
4317
                if err != nil {
6✔
4318
                        p.log.Errorf("unable to force close "+
3✔
4319
                                "link(%v): %v", failure.shortChanID, err)
3✔
4320
                } else {
6✔
4321
                        p.log.Infof("channel(%v) force "+
3✔
4322
                                "closed with txid %v",
3✔
4323
                                failure.shortChanID, closeTx.TxHash())
3✔
4324
                }
3✔
4325
        }
4326

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

×
4332
                if err := lnChan.State().MarkBorked(); err != nil {
×
4333
                        p.log.Errorf("Unable to mark channel %v borked: %v",
×
4334
                                failure.shortChanID, err)
×
4335
                }
×
4336
        }
4337

4338
        // Send an error to the peer, why we failed the channel.
4339
        if failure.linkErr.ShouldSendToPeer() {
6✔
4340
                // If SendData is set, send it to the peer. If not, we'll use
3✔
4341
                // the standard error messages in the payload. We only include
3✔
4342
                // sendData in the cases where the error data does not contain
3✔
4343
                // sensitive information.
3✔
4344
                data := []byte(failure.linkErr.Error())
3✔
4345
                if failure.linkErr.SendData != nil {
3✔
4346
                        data = failure.linkErr.SendData
×
4347
                }
×
4348

4349
                var networkMsg lnwire.Message
3✔
4350
                if failure.linkErr.Warning {
3✔
4351
                        networkMsg = &lnwire.Warning{
×
4352
                                ChanID: failure.chanID,
×
4353
                                Data:   data,
×
4354
                        }
×
4355
                } else {
3✔
4356
                        networkMsg = &lnwire.Error{
3✔
4357
                                ChanID: failure.chanID,
3✔
4358
                                Data:   data,
3✔
4359
                        }
3✔
4360
                }
3✔
4361

4362
                err := p.SendMessage(true, networkMsg)
3✔
4363
                if err != nil {
3✔
4364
                        p.log.Errorf("unable to send msg to "+
×
4365
                                "remote peer: %v", err)
×
4366
                }
×
4367
        }
4368

4369
        // If the failure action is disconnect, then we'll execute that now. If
4370
        // we had to send an error above, it was a sync call, so we expect the
4371
        // message to be flushed on the wire by now.
4372
        if failure.linkErr.FailureAction == htlcswitch.LinkFailureDisconnect {
3✔
4373
                p.Disconnect(fmt.Errorf("link requested disconnect"))
×
4374
        }
×
4375
}
4376

4377
// fetchLinkFromKeyAndCid fetches a link from the switch via the remote's
4378
// public key and the channel id.
4379
func (p *Brontide) fetchLinkFromKeyAndCid(
4380
        cid lnwire.ChannelID) htlcswitch.ChannelUpdateHandler {
22✔
4381

22✔
4382
        var chanLink htlcswitch.ChannelUpdateHandler
22✔
4383

22✔
4384
        // We don't need to check the error here, and can instead just loop
22✔
4385
        // over the slice and return nil.
22✔
4386
        links, _ := p.cfg.Switch.GetLinksByInterface(p.cfg.PubKeyBytes)
22✔
4387
        for _, link := range links {
43✔
4388
                if link.ChanID() == cid {
42✔
4389
                        chanLink = link
21✔
4390
                        break
21✔
4391
                }
4392
        }
4393

4394
        return chanLink
22✔
4395
}
4396

4397
// finalizeChanClosure performs the final clean up steps once the cooperative
4398
// closure transaction has been fully broadcast. The finalized closing state
4399
// machine should be passed in. Once the transaction has been sufficiently
4400
// confirmed, the channel will be marked as fully closed within the database,
4401
// and any clients will be notified of updates to the closing state.
4402
func (p *Brontide) finalizeChanClosure(chanCloser *chancloser.ChanCloser) {
7✔
4403
        closeReq := chanCloser.CloseRequest()
7✔
4404

7✔
4405
        // First, we'll clear all indexes related to the channel in question.
7✔
4406
        chanPoint := chanCloser.Channel().ChannelPoint()
7✔
4407
        p.WipeChannel(&chanPoint)
7✔
4408

7✔
4409
        // Also clear the activeChanCloses map of this channel.
7✔
4410
        cid := lnwire.NewChanIDFromOutPoint(chanPoint)
7✔
4411
        p.activeChanCloses.Delete(cid) // TODO(roasbeef): existing race
7✔
4412

7✔
4413
        // Next, we'll launch a goroutine which will request to be notified by
7✔
4414
        // the ChainNotifier once the closure transaction obtains a single
7✔
4415
        // confirmation.
7✔
4416
        notifier := p.cfg.ChainNotifier
7✔
4417

7✔
4418
        // If any error happens during waitForChanToClose, forward it to
7✔
4419
        // closeReq. If this channel closure is not locally initiated, closeReq
7✔
4420
        // will be nil, so just ignore the error.
7✔
4421
        errChan := make(chan error, 1)
7✔
4422
        if closeReq != nil {
12✔
4423
                errChan = closeReq.Err
5✔
4424
        }
5✔
4425

4426
        closingTx, err := chanCloser.ClosingTx()
7✔
4427
        if err != nil {
7✔
4428
                if closeReq != nil {
×
4429
                        p.log.Error(err)
×
4430
                        closeReq.Err <- err
×
4431
                }
×
4432
        }
4433

4434
        closingTxid := closingTx.TxHash()
7✔
4435

7✔
4436
        // If this is a locally requested shutdown, update the caller with a
7✔
4437
        // new event detailing the current pending state of this request.
7✔
4438
        if closeReq != nil {
12✔
4439
                closeReq.Updates <- &PendingUpdate{
5✔
4440
                        Txid: closingTxid[:],
5✔
4441
                }
5✔
4442
        }
5✔
4443

4444
        localOut := chanCloser.LocalCloseOutput()
7✔
4445
        remoteOut := chanCloser.RemoteCloseOutput()
7✔
4446
        auxOut := chanCloser.AuxOutputs()
7✔
4447
        go WaitForChanToClose(
7✔
4448
                chanCloser.NegotiationHeight(), notifier, errChan,
7✔
4449
                &chanPoint, &closingTxid, closingTx.TxOut[0].PkScript, func() {
14✔
4450
                        // Respond to the local subsystem which requested the
7✔
4451
                        // channel closure.
7✔
4452
                        if closeReq != nil {
12✔
4453
                                closeReq.Updates <- &ChannelCloseUpdate{
5✔
4454
                                        ClosingTxid:       closingTxid[:],
5✔
4455
                                        Success:           true,
5✔
4456
                                        LocalCloseOutput:  localOut,
5✔
4457
                                        RemoteCloseOutput: remoteOut,
5✔
4458
                                        AuxOutputs:        auxOut,
5✔
4459
                                }
5✔
4460
                        }
5✔
4461
                },
4462
        )
4463
}
4464

4465
// WaitForChanToClose uses the passed notifier to wait until the channel has
4466
// been detected as closed on chain and then concludes by executing the
4467
// following actions: the channel point will be sent over the settleChan, and
4468
// finally the callback will be executed. If any error is encountered within
4469
// the function, then it will be sent over the errChan.
4470
func WaitForChanToClose(bestHeight uint32, notifier chainntnfs.ChainNotifier,
4471
        errChan chan error, chanPoint *wire.OutPoint,
4472
        closingTxID *chainhash.Hash, closeScript []byte, cb func()) {
7✔
4473

7✔
4474
        peerLog.Infof("Waiting for confirmation of close of ChannelPoint(%v) "+
7✔
4475
                "with txid: %v", chanPoint, closingTxID)
7✔
4476

7✔
4477
        // TODO(roasbeef): add param for num needed confs
7✔
4478
        confNtfn, err := notifier.RegisterConfirmationsNtfn(
7✔
4479
                closingTxID, closeScript, 1, bestHeight,
7✔
4480
        )
7✔
4481
        if err != nil {
7✔
4482
                if errChan != nil {
×
4483
                        errChan <- err
×
4484
                }
×
4485
                return
×
4486
        }
4487

4488
        // In the case that the ChainNotifier is shutting down, all subscriber
4489
        // notification channels will be closed, generating a nil receive.
4490
        height, ok := <-confNtfn.Confirmed
7✔
4491
        if !ok {
10✔
4492
                return
3✔
4493
        }
3✔
4494

4495
        // The channel has been closed, remove it from any active indexes, and
4496
        // the database state.
4497
        peerLog.Infof("ChannelPoint(%v) is now closed at "+
7✔
4498
                "height %v", chanPoint, height.BlockHeight)
7✔
4499

7✔
4500
        // Finally, execute the closure call back to mark the confirmation of
7✔
4501
        // the transaction closing the contract.
7✔
4502
        cb()
7✔
4503
}
4504

4505
// WipeChannel removes the passed channel point from all indexes associated with
4506
// the peer and the switch.
4507
func (p *Brontide) WipeChannel(chanPoint *wire.OutPoint) {
7✔
4508
        chanID := lnwire.NewChanIDFromOutPoint(*chanPoint)
7✔
4509

7✔
4510
        p.activeChannels.Delete(chanID)
7✔
4511

7✔
4512
        // Instruct the HtlcSwitch to close this link as the channel is no
7✔
4513
        // longer active.
7✔
4514
        p.cfg.Switch.RemoveLink(chanID)
7✔
4515
}
7✔
4516

4517
// handleInitMsg handles the incoming init message which contains global and
4518
// local feature vectors. If feature vectors are incompatible then disconnect.
4519
func (p *Brontide) handleInitMsg(msg *lnwire.Init) error {
6✔
4520
        // First, merge any features from the legacy global features field into
6✔
4521
        // those presented in the local features fields.
6✔
4522
        err := msg.Features.Merge(msg.GlobalFeatures)
6✔
4523
        if err != nil {
6✔
4524
                return fmt.Errorf("unable to merge legacy global features: %w",
×
4525
                        err)
×
4526
        }
×
4527

4528
        // Then, finalize the remote feature vector providing the flattened
4529
        // feature bit namespace.
4530
        p.remoteFeatures = lnwire.NewFeatureVector(
6✔
4531
                msg.Features, lnwire.Features,
6✔
4532
        )
6✔
4533

6✔
4534
        // Now that we have their features loaded, we'll ensure that they
6✔
4535
        // didn't set any required bits that we don't know of.
6✔
4536
        err = feature.ValidateRequired(p.remoteFeatures)
6✔
4537
        if err != nil {
6✔
4538
                return fmt.Errorf("invalid remote features: %w", err)
×
4539
        }
×
4540

4541
        // Ensure the remote party's feature vector contains all transitive
4542
        // dependencies. We know ours are correct since they are validated
4543
        // during the feature manager's instantiation.
4544
        err = feature.ValidateDeps(p.remoteFeatures)
6✔
4545
        if err != nil {
6✔
4546
                return fmt.Errorf("invalid remote features: %w", err)
×
4547
        }
×
4548

4549
        // Now that we know we understand their requirements, we'll check to
4550
        // see if they don't support anything that we deem to be mandatory.
4551
        if !p.remoteFeatures.HasFeature(lnwire.DataLossProtectRequired) {
6✔
4552
                return fmt.Errorf("data loss protection required")
×
4553
        }
×
4554

4555
        // If we have an AuxChannelNegotiator and the peer sent aux features,
4556
        // process them.
4557
        p.cfg.AuxChannelNegotiator.WhenSome(
6✔
4558
                func(acn lnwallet.AuxChannelNegotiator) {
6✔
4559
                        err = acn.ProcessInitRecords(
×
4560
                                p.cfg.PubKeyBytes, msg.CustomRecords.Copy(),
×
4561
                        )
×
4562
                },
×
4563
        )
4564
        if err != nil {
6✔
4565
                return fmt.Errorf("could not process init records: %w", err)
×
4566
        }
×
4567

4568
        return nil
6✔
4569
}
4570

4571
// LocalFeatures returns the set of global features that has been advertised by
4572
// the local node. This allows sub-systems that use this interface to gate their
4573
// behavior off the set of negotiated feature bits.
4574
//
4575
// NOTE: Part of the lnpeer.Peer interface.
4576
func (p *Brontide) LocalFeatures() *lnwire.FeatureVector {
3✔
4577
        return p.cfg.Features
3✔
4578
}
3✔
4579

4580
// RemoteFeatures returns the set of global features that has been advertised by
4581
// the remote node. This allows sub-systems that use this interface to gate
4582
// their behavior off the set of negotiated feature bits.
4583
//
4584
// NOTE: Part of the lnpeer.Peer interface.
4585
func (p *Brontide) RemoteFeatures() *lnwire.FeatureVector {
23✔
4586
        return p.remoteFeatures
23✔
4587
}
23✔
4588

4589
// hasNegotiatedScidAlias returns true if we've negotiated the
4590
// option-scid-alias feature bit with the peer.
4591
func (p *Brontide) hasNegotiatedScidAlias() bool {
6✔
4592
        peerHas := p.remoteFeatures.HasFeature(lnwire.ScidAliasOptional)
6✔
4593
        localHas := p.cfg.Features.HasFeature(lnwire.ScidAliasOptional)
6✔
4594
        return peerHas && localHas
6✔
4595
}
6✔
4596

4597
// sendInitMsg sends the Init message to the remote peer. This message contains
4598
// our currently supported local and global features.
4599
func (p *Brontide) sendInitMsg(legacyChan bool) error {
10✔
4600
        features := p.cfg.Features.Clone()
10✔
4601
        legacyFeatures := p.cfg.LegacyFeatures.Clone()
10✔
4602

10✔
4603
        // If we have a legacy channel open with a peer, we downgrade static
10✔
4604
        // remote required to optional in case the peer does not understand the
10✔
4605
        // required feature bit. If we do not do this, the peer will reject our
10✔
4606
        // connection because it does not understand a required feature bit, and
10✔
4607
        // our channel will be unusable.
10✔
4608
        if legacyChan && features.RequiresFeature(lnwire.StaticRemoteKeyRequired) {
11✔
4609
                p.log.Infof("Legacy channel open with peer, " +
1✔
4610
                        "downgrading static remote required feature bit to " +
1✔
4611
                        "optional")
1✔
4612

1✔
4613
                // Unset and set in both the local and global features to
1✔
4614
                // ensure both sets are consistent and merge able by old and
1✔
4615
                // new nodes.
1✔
4616
                features.Unset(lnwire.StaticRemoteKeyRequired)
1✔
4617
                legacyFeatures.Unset(lnwire.StaticRemoteKeyRequired)
1✔
4618

1✔
4619
                features.Set(lnwire.StaticRemoteKeyOptional)
1✔
4620
                legacyFeatures.Set(lnwire.StaticRemoteKeyOptional)
1✔
4621
        }
1✔
4622

4623
        msg := lnwire.NewInitMessage(
10✔
4624
                legacyFeatures.RawFeatureVector,
10✔
4625
                features.RawFeatureVector,
10✔
4626
        )
10✔
4627

10✔
4628
        var err error
10✔
4629

10✔
4630
        // If we have an AuxChannelNegotiator, get custom feature bits to
10✔
4631
        // include in the init message.
10✔
4632
        p.cfg.AuxChannelNegotiator.WhenSome(
10✔
4633
                func(negotiator lnwallet.AuxChannelNegotiator) {
10✔
4634
                        var auxRecords lnwire.CustomRecords
×
4635
                        auxRecords, err = negotiator.GetInitRecords(
×
4636
                                p.cfg.PubKeyBytes,
×
4637
                        )
×
4638
                        if err != nil {
×
4639
                                p.log.Warnf("Failed to get aux init features: "+
×
4640
                                        "%v", err)
×
4641
                                return
×
4642
                        }
×
4643

4644
                        mergedRecs := msg.CustomRecords.MergedCopy(auxRecords)
×
4645
                        msg.CustomRecords = mergedRecs
×
4646
                },
4647
        )
4648
        if err != nil {
10✔
4649
                return err
×
4650
        }
×
4651

4652
        return p.writeMessage(msg)
10✔
4653
}
4654

4655
// resendChanSyncMsg will attempt to find a channel sync message for the closed
4656
// channel and resend it to our peer.
4657
func (p *Brontide) resendChanSyncMsg(cid lnwire.ChannelID) error {
3✔
4658
        // If we already re-sent the mssage for this channel, we won't do it
3✔
4659
        // again.
3✔
4660
        if _, ok := p.resentChanSyncMsg[cid]; ok {
4✔
4661
                return nil
1✔
4662
        }
1✔
4663

4664
        // Check if we have any channel sync messages stored for this channel.
4665
        c, err := p.cfg.ChannelDB.FetchClosedChannelForID(cid)
3✔
4666
        if err != nil {
6✔
4667
                return fmt.Errorf("unable to fetch channel sync messages for "+
3✔
4668
                        "peer %v: %v", p, err)
3✔
4669
        }
3✔
4670

4671
        if c.LastChanSyncMsg == nil {
3✔
4672
                return fmt.Errorf("no chan sync message stored for channel %v",
×
4673
                        cid)
×
4674
        }
×
4675

4676
        if !c.RemotePub.IsEqual(p.IdentityKey()) {
3✔
4677
                return fmt.Errorf("ignoring channel reestablish from "+
×
4678
                        "peer=%x", p.IdentityKey().SerializeCompressed())
×
4679
        }
×
4680

4681
        p.log.Debugf("Re-sending channel sync message for channel %v to "+
3✔
4682
                "peer", cid)
3✔
4683

3✔
4684
        if err := p.SendMessage(true, c.LastChanSyncMsg); err != nil {
3✔
4685
                return fmt.Errorf("failed resending channel sync "+
×
4686
                        "message to peer %v: %v", p, err)
×
4687
        }
×
4688

4689
        p.log.Debugf("Re-sent channel sync message for channel %v to peer ",
3✔
4690
                cid)
3✔
4691

3✔
4692
        // Note down that we sent the message, so we won't resend it again for
3✔
4693
        // this connection.
3✔
4694
        p.resentChanSyncMsg[cid] = struct{}{}
3✔
4695

3✔
4696
        return nil
3✔
4697
}
4698

4699
// SendMessage sends a variadic number of high-priority messages to the remote
4700
// peer. The first argument denotes if the method should block until the
4701
// messages have been sent to the remote peer or an error is returned,
4702
// otherwise it returns immediately after queuing.
4703
//
4704
// NOTE: Part of the lnpeer.Peer interface.
4705
func (p *Brontide) SendMessage(sync bool, msgs ...lnwire.Message) error {
6✔
4706
        return p.sendMessage(sync, true, msgs...)
6✔
4707
}
6✔
4708

4709
// SendMessageLazy sends a variadic number of low-priority messages to the
4710
// remote peer. The first argument denotes if the method should block until
4711
// the messages have been sent to the remote peer or an error is returned,
4712
// otherwise it returns immediately after queueing.
4713
//
4714
// NOTE: Part of the lnpeer.Peer interface.
4715
func (p *Brontide) SendMessageLazy(sync bool, msgs ...lnwire.Message) error {
4✔
4716
        return p.sendMessage(sync, false, msgs...)
4✔
4717
}
4✔
4718

4719
// sendMessage queues a variadic number of messages using the passed priority
4720
// to the remote peer. If sync is true, this method will block until the
4721
// messages have been sent to the remote peer or an error is returned, otherwise
4722
// it returns immediately after queueing.
4723
func (p *Brontide) sendMessage(sync, priority bool, msgs ...lnwire.Message) error {
7✔
4724
        // Add all incoming messages to the outgoing queue. A list of error
7✔
4725
        // chans is populated for each message if the caller requested a sync
7✔
4726
        // send.
7✔
4727
        var errChans []chan error
7✔
4728
        if sync {
11✔
4729
                errChans = make([]chan error, 0, len(msgs))
4✔
4730
        }
4✔
4731
        for _, msg := range msgs {
14✔
4732
                // If a sync send was requested, create an error chan to listen
7✔
4733
                // for an ack from the writeHandler.
7✔
4734
                var errChan chan error
7✔
4735
                if sync {
11✔
4736
                        errChan = make(chan error, 1)
4✔
4737
                        errChans = append(errChans, errChan)
4✔
4738
                }
4✔
4739

4740
                if priority {
13✔
4741
                        p.queueMsg(msg, errChan)
6✔
4742
                } else {
10✔
4743
                        p.queueMsgLazy(msg, errChan)
4✔
4744
                }
4✔
4745
        }
4746

4747
        // Wait for all replies from the writeHandler. For async sends, this
4748
        // will be a NOP as the list of error chans is nil.
4749
        for _, errChan := range errChans {
11✔
4750
                select {
4✔
4751
                case err := <-errChan:
4✔
4752
                        return err
4✔
4753
                case <-p.cg.Done():
×
4754
                        return lnpeer.ErrPeerExiting
×
4755
                case <-p.cfg.Quit:
×
4756
                        return lnpeer.ErrPeerExiting
×
4757
                }
4758
        }
4759

4760
        return nil
6✔
4761
}
4762

4763
// PubKey returns the pubkey of the peer in compressed serialized format.
4764
//
4765
// NOTE: Part of the lnpeer.Peer interface.
4766
func (p *Brontide) PubKey() [33]byte {
5✔
4767
        return p.cfg.PubKeyBytes
5✔
4768
}
5✔
4769

4770
// IdentityKey returns the public key of the remote peer.
4771
//
4772
// NOTE: Part of the lnpeer.Peer interface.
4773
func (p *Brontide) IdentityKey() *btcec.PublicKey {
18✔
4774
        return p.cfg.Addr.IdentityKey
18✔
4775
}
18✔
4776

4777
// Address returns the network address of the remote peer.
4778
//
4779
// NOTE: Part of the lnpeer.Peer interface.
4780
func (p *Brontide) Address() net.Addr {
3✔
4781
        return p.cfg.Addr.Address
3✔
4782
}
3✔
4783

4784
// AddNewChannel adds a new channel to the peer. The channel should fail to be
4785
// added if the cancel channel is closed.
4786
//
4787
// NOTE: Part of the lnpeer.Peer interface.
4788
func (p *Brontide) AddNewChannel(newChan *lnpeer.NewChannel,
4789
        cancel <-chan struct{}) error {
3✔
4790

3✔
4791
        errChan := make(chan error, 1)
3✔
4792
        newChanMsg := &newChannelMsg{
3✔
4793
                channel: newChan,
3✔
4794
                err:     errChan,
3✔
4795
        }
3✔
4796

3✔
4797
        select {
3✔
4798
        case p.newActiveChannel <- newChanMsg:
3✔
4799
        case <-cancel:
×
4800
                return errors.New("canceled adding new channel")
×
4801
        case <-p.cg.Done():
×
4802
                return lnpeer.ErrPeerExiting
×
4803
        }
4804

4805
        // We pause here to wait for the peer to recognize the new channel
4806
        // before we close the channel barrier corresponding to the channel.
4807
        select {
3✔
4808
        case err := <-errChan:
3✔
4809
                return err
3✔
4810
        case <-p.cg.Done():
×
4811
                return lnpeer.ErrPeerExiting
×
4812
        }
4813
}
4814

4815
// AddPendingChannel adds a pending open channel to the peer. The channel
4816
// should fail to be added if the cancel channel is closed.
4817
//
4818
// NOTE: Part of the lnpeer.Peer interface.
4819
func (p *Brontide) AddPendingChannel(cid lnwire.ChannelID,
4820
        cancel <-chan struct{}) error {
3✔
4821

3✔
4822
        errChan := make(chan error, 1)
3✔
4823
        newChanMsg := &newChannelMsg{
3✔
4824
                channelID: cid,
3✔
4825
                err:       errChan,
3✔
4826
        }
3✔
4827

3✔
4828
        select {
3✔
4829
        case p.newPendingChannel <- newChanMsg:
3✔
4830

4831
        case <-cancel:
×
4832
                return errors.New("canceled adding pending channel")
×
4833

4834
        case <-p.cg.Done():
×
4835
                return lnpeer.ErrPeerExiting
×
4836
        }
4837

4838
        // We pause here to wait for the peer to recognize the new pending
4839
        // channel before we close the channel barrier corresponding to the
4840
        // channel.
4841
        select {
3✔
4842
        case err := <-errChan:
3✔
4843
                return err
3✔
4844

4845
        case <-cancel:
×
4846
                return errors.New("canceled adding pending channel")
×
4847

4848
        case <-p.cg.Done():
×
4849
                return lnpeer.ErrPeerExiting
×
4850
        }
4851
}
4852

4853
// RemovePendingChannel removes a pending open channel from the peer.
4854
//
4855
// NOTE: Part of the lnpeer.Peer interface.
4856
func (p *Brontide) RemovePendingChannel(cid lnwire.ChannelID) error {
3✔
4857
        errChan := make(chan error, 1)
3✔
4858
        newChanMsg := &newChannelMsg{
3✔
4859
                channelID: cid,
3✔
4860
                err:       errChan,
3✔
4861
        }
3✔
4862

3✔
4863
        select {
3✔
4864
        case p.removePendingChannel <- newChanMsg:
3✔
4865
        case <-p.cg.Done():
×
4866
                return lnpeer.ErrPeerExiting
×
4867
        }
4868

4869
        // We pause here to wait for the peer to respond to the cancellation of
4870
        // the pending channel before we close the channel barrier
4871
        // corresponding to the channel.
4872
        select {
3✔
4873
        case err := <-errChan:
3✔
4874
                return err
3✔
4875

4876
        case <-p.cg.Done():
×
4877
                return lnpeer.ErrPeerExiting
×
4878
        }
4879
}
4880

4881
// StartTime returns the time at which the connection was established if the
4882
// peer started successfully, and zero otherwise.
4883
func (p *Brontide) StartTime() time.Time {
3✔
4884
        return p.startTime
3✔
4885
}
3✔
4886

4887
// handleCloseMsg is called when a new cooperative channel closure related
4888
// message is received from the remote peer. We'll use this message to advance
4889
// the chan closer state machine.
4890
func (p *Brontide) handleCloseMsg(msg *closeMsg) {
16✔
4891
        link := p.fetchLinkFromKeyAndCid(msg.cid)
16✔
4892

16✔
4893
        // We'll now fetch the matching closing state machine in order to
16✔
4894
        // continue, or finalize the channel closure process.
16✔
4895
        chanCloserE, err := p.fetchActiveChanCloser(msg.cid)
16✔
4896
        if err != nil {
19✔
4897
                // If the channel is not known to us, we'll simply ignore this
3✔
4898
                // message.
3✔
4899
                if err == ErrChannelNotFound {
6✔
4900
                        return
3✔
4901
                }
3✔
4902

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

×
4905
                errMsg := &lnwire.Error{
×
4906
                        ChanID: msg.cid,
×
4907
                        Data:   lnwire.ErrorData(err.Error()),
×
4908
                }
×
4909
                p.queueMsg(errMsg, nil)
×
4910
                return
×
4911
        }
4912

4913
        if chanCloserE.IsRight() {
16✔
4914
                // TODO(roasbeef): assert?
×
4915
                return
×
4916
        }
×
4917

4918
        // At this point, we'll only enter this call path if a negotiate chan
4919
        // closer was used. So we'll extract that from the either now.
4920
        //
4921
        // TODO(roabeef): need extra helper func for either to make cleaner
4922
        var chanCloser *chancloser.ChanCloser
16✔
4923
        chanCloserE.WhenLeft(func(c *chancloser.ChanCloser) {
32✔
4924
                chanCloser = c
16✔
4925
        })
16✔
4926

4927
        handleErr := func(err error) {
17✔
4928
                err = fmt.Errorf("unable to process close msg: %w", err)
1✔
4929
                p.log.Error(err)
1✔
4930

1✔
4931
                // As the negotiations failed, we'll reset the channel state
1✔
4932
                // machine to ensure we act to on-chain events as normal.
1✔
4933
                chanCloser.Channel().ResetState()
1✔
4934
                if chanCloser.CloseRequest() != nil {
1✔
4935
                        chanCloser.CloseRequest().Err <- err
×
4936
                }
×
4937

4938
                p.activeChanCloses.Delete(msg.cid)
1✔
4939

1✔
4940
                p.Disconnect(err)
1✔
4941
        }
4942

4943
        // Next, we'll process the next message using the target state machine.
4944
        // We'll either continue negotiation, or halt.
4945
        switch typed := msg.msg.(type) {
16✔
4946
        case *lnwire.Shutdown:
8✔
4947
                // Disable incoming adds immediately.
8✔
4948
                if link != nil && !link.DisableAdds(htlcswitch.Incoming) {
8✔
4949
                        p.log.Warnf("Incoming link adds already disabled: %v",
×
4950
                                link.ChanID())
×
4951
                }
×
4952

4953
                oShutdown, err := chanCloser.ReceiveShutdown(*typed)
8✔
4954
                if err != nil {
8✔
4955
                        handleErr(err)
×
4956
                        return
×
4957
                }
×
4958

4959
                oShutdown.WhenSome(func(msg lnwire.Shutdown) {
14✔
4960
                        // If the link is nil it means we can immediately queue
6✔
4961
                        // the Shutdown message since we don't have to wait for
6✔
4962
                        // commitment transaction synchronization.
6✔
4963
                        if link == nil {
7✔
4964
                                p.queueMsg(&msg, nil)
1✔
4965
                                return
1✔
4966
                        }
1✔
4967

4968
                        // Immediately disallow any new HTLC's from being added
4969
                        // in the outgoing direction.
4970
                        if !link.DisableAdds(htlcswitch.Outgoing) {
5✔
4971
                                p.log.Warnf("Outgoing link adds already "+
×
4972
                                        "disabled: %v", link.ChanID())
×
4973
                        }
×
4974

4975
                        // When we have a Shutdown to send, we defer it till the
4976
                        // next time we send a CommitSig to remain spec
4977
                        // compliant.
4978
                        link.OnCommitOnce(htlcswitch.Outgoing, func() {
10✔
4979
                                p.queueMsg(&msg, nil)
5✔
4980
                        })
5✔
4981
                })
4982

4983
                beginNegotiation := func() {
16✔
4984
                        oClosingSigned, err := chanCloser.BeginNegotiation()
8✔
4985
                        if err != nil {
8✔
4986
                                handleErr(err)
×
4987
                                return
×
4988
                        }
×
4989

4990
                        oClosingSigned.WhenSome(func(msg lnwire.ClosingSigned) {
16✔
4991
                                p.queueMsg(&msg, nil)
8✔
4992
                        })
8✔
4993
                }
4994

4995
                if link == nil {
9✔
4996
                        beginNegotiation()
1✔
4997
                } else {
8✔
4998
                        // Now we register a flush hook to advance the
7✔
4999
                        // ChanCloser and possibly send out a ClosingSigned
7✔
5000
                        // when the link finishes draining.
7✔
5001
                        link.OnFlushedOnce(func() {
14✔
5002
                                // Remove link in goroutine to prevent deadlock.
7✔
5003
                                go p.cfg.Switch.RemoveLink(msg.cid)
7✔
5004
                                beginNegotiation()
7✔
5005
                        })
7✔
5006
                }
5007

5008
        case *lnwire.ClosingSigned:
11✔
5009
                oClosingSigned, err := chanCloser.ReceiveClosingSigned(*typed)
11✔
5010
                if err != nil {
12✔
5011
                        handleErr(err)
1✔
5012
                        return
1✔
5013
                }
1✔
5014

5015
                oClosingSigned.WhenSome(func(msg lnwire.ClosingSigned) {
22✔
5016
                        p.queueMsg(&msg, nil)
11✔
5017
                })
11✔
5018

5019
        default:
×
5020
                panic("impossible closeMsg type")
×
5021
        }
5022

5023
        // If we haven't finished close negotiations, then we'll continue as we
5024
        // can't yet finalize the closure.
5025
        if _, err := chanCloser.ClosingTx(); err != nil {
26✔
5026
                return
11✔
5027
        }
11✔
5028

5029
        // Otherwise, we've agreed on a closing fee! In this case, we'll wrap up
5030
        // the channel closure by notifying relevant sub-systems and launching a
5031
        // goroutine to wait for close tx conf.
5032
        p.finalizeChanClosure(chanCloser)
7✔
5033
}
5034

5035
// HandleLocalCloseChanReqs accepts a *htlcswitch.ChanClose and passes it onto
5036
// the channelManager goroutine, which will shut down the link and possibly
5037
// close the channel.
5038
func (p *Brontide) HandleLocalCloseChanReqs(req *htlcswitch.ChanClose) {
3✔
5039
        select {
3✔
5040
        case p.localCloseChanReqs <- req:
3✔
5041
                p.log.Info("Local close channel request is going to be " +
3✔
5042
                        "delivered to the peer")
3✔
5043
        case <-p.cg.Done():
×
5044
                p.log.Info("Unable to deliver local close channel request " +
×
5045
                        "to peer")
×
5046
        }
5047
}
5048

5049
// NetAddress returns the network of the remote peer as an lnwire.NetAddress.
5050
func (p *Brontide) NetAddress() *lnwire.NetAddress {
3✔
5051
        return p.cfg.Addr
3✔
5052
}
3✔
5053

5054
// Inbound is a getter for the Brontide's Inbound boolean in cfg.
5055
func (p *Brontide) Inbound() bool {
3✔
5056
        return p.cfg.Inbound
3✔
5057
}
3✔
5058

5059
// ConnReq is a getter for the Brontide's connReq in cfg.
5060
func (p *Brontide) ConnReq() *connmgr.ConnReq {
3✔
5061
        return p.cfg.ConnReq
3✔
5062
}
3✔
5063

5064
// ErrorBuffer is a getter for the Brontide's errorBuffer in cfg.
5065
func (p *Brontide) ErrorBuffer() *queue.CircularBuffer {
3✔
5066
        return p.cfg.ErrorBuffer
3✔
5067
}
3✔
5068

5069
// SetAddress sets the remote peer's address given an address.
5070
func (p *Brontide) SetAddress(address net.Addr) {
×
5071
        p.cfg.Addr.Address = address
×
5072
}
×
5073

5074
// ActiveSignal returns the peer's active signal.
5075
func (p *Brontide) ActiveSignal() chan struct{} {
3✔
5076
        return p.activeSignal
3✔
5077
}
3✔
5078

5079
// Conn returns a pointer to the peer's connection struct.
5080
func (p *Brontide) Conn() net.Conn {
3✔
5081
        return p.cfg.Conn
3✔
5082
}
3✔
5083

5084
// BytesReceived returns the number of bytes received from the peer.
5085
func (p *Brontide) BytesReceived() uint64 {
3✔
5086
        return atomic.LoadUint64(&p.bytesReceived)
3✔
5087
}
3✔
5088

5089
// BytesSent returns the number of bytes sent to the peer.
5090
func (p *Brontide) BytesSent() uint64 {
3✔
5091
        return atomic.LoadUint64(&p.bytesSent)
3✔
5092
}
3✔
5093

5094
// LastRemotePingPayload returns the last payload the remote party sent as part
5095
// of their ping.
5096
func (p *Brontide) LastRemotePingPayload() []byte {
3✔
5097
        pingPayload := p.lastPingPayload.Load()
3✔
5098
        if pingPayload == nil {
6✔
5099
                return []byte{}
3✔
5100
        }
3✔
5101

5102
        pingBytes, ok := pingPayload.(lnwire.PingPayload)
×
5103
        if !ok {
×
5104
                return nil
×
5105
        }
×
5106

5107
        return pingBytes
×
5108
}
5109

5110
// attachChannelEventSubscription creates a channel event subscription and
5111
// attaches to client to Brontide if the reenableTimeout is no greater than 1
5112
// minute.
5113
func (p *Brontide) attachChannelEventSubscription() error {
6✔
5114
        // If the timeout is greater than 1 minute, it's unlikely that the link
6✔
5115
        // hasn't yet finished its reestablishment. Return a nil without
6✔
5116
        // creating the client to specify that we don't want to retry.
6✔
5117
        if p.cfg.ChanActiveTimeout > 1*time.Minute {
9✔
5118
                return nil
3✔
5119
        }
3✔
5120

5121
        // When the reenable timeout is less than 1 minute, it's likely the
5122
        // channel link hasn't finished its reestablishment yet. In that case,
5123
        // we'll give it a second chance by subscribing to the channel update
5124
        // events. Upon receiving the `ActiveLinkEvent`, we'll then request
5125
        // enabling the channel again.
5126
        sub, err := p.cfg.ChannelNotifier.SubscribeChannelEvents()
6✔
5127
        if err != nil {
6✔
5128
                return fmt.Errorf("SubscribeChannelEvents failed: %w", err)
×
5129
        }
×
5130

5131
        p.channelEventClient = sub
6✔
5132

6✔
5133
        return nil
6✔
5134
}
5135

5136
// updateNextRevocation updates the existing channel's next revocation if it's
5137
// nil.
5138
func (p *Brontide) updateNextRevocation(c *channeldb.OpenChannel) error {
6✔
5139
        chanPoint := c.FundingOutpoint
6✔
5140
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
6✔
5141

6✔
5142
        // Read the current channel.
6✔
5143
        currentChan, loaded := p.activeChannels.Load(chanID)
6✔
5144

6✔
5145
        // currentChan should exist, but we perform a check anyway to avoid nil
6✔
5146
        // pointer dereference.
6✔
5147
        if !loaded {
7✔
5148
                return fmt.Errorf("missing active channel with chanID=%v",
1✔
5149
                        chanID)
1✔
5150
        }
1✔
5151

5152
        // currentChan should not be nil, but we perform a check anyway to
5153
        // avoid nil pointer dereference.
5154
        if currentChan == nil {
6✔
5155
                return fmt.Errorf("found nil active channel with chanID=%v",
1✔
5156
                        chanID)
1✔
5157
        }
1✔
5158

5159
        // If we're being sent a new channel, and our existing channel doesn't
5160
        // have the next revocation, then we need to update the current
5161
        // existing channel.
5162
        if currentChan.RemoteNextRevocation() != nil {
4✔
5163
                return nil
×
5164
        }
×
5165

5166
        p.log.Infof("Processing retransmitted ChannelReady for "+
4✔
5167
                "ChannelPoint(%v)", chanPoint)
4✔
5168

4✔
5169
        nextRevoke := c.RemoteNextRevocation
4✔
5170

4✔
5171
        err := currentChan.InitNextRevocation(nextRevoke)
4✔
5172
        if err != nil {
4✔
5173
                return fmt.Errorf("unable to init next revocation: %w", err)
×
5174
        }
×
5175

5176
        return nil
4✔
5177
}
5178

5179
// addActiveChannel adds a new active channel to the `activeChannels` map. It
5180
// takes a `channeldb.OpenChannel`, creates a `lnwallet.LightningChannel` from
5181
// it and assembles it with a channel link.
5182
func (p *Brontide) addActiveChannel(c *lnpeer.NewChannel) error {
3✔
5183
        chanPoint := c.FundingOutpoint
3✔
5184
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
3✔
5185

3✔
5186
        // If we've reached this point, there are two possible scenarios.  If
3✔
5187
        // the channel was in the active channels map as nil, then it was
3✔
5188
        // loaded from disk and we need to send reestablish. Else, it was not
3✔
5189
        // loaded from disk and we don't need to send reestablish as this is a
3✔
5190
        // fresh channel.
3✔
5191
        shouldReestablish := p.isLoadedFromDisk(chanID)
3✔
5192

3✔
5193
        chanOpts := c.ChanOpts
3✔
5194
        if shouldReestablish {
6✔
5195
                // If we have to do the reestablish dance for this channel,
3✔
5196
                // ensure that we don't try to call InitRemoteMusigNonces twice
3✔
5197
                // by calling SkipNonceInit.
3✔
5198
                chanOpts = append(chanOpts, lnwallet.WithSkipNonceInit())
3✔
5199
        }
3✔
5200

5201
        p.cfg.AuxLeafStore.WhenSome(func(s lnwallet.AuxLeafStore) {
3✔
5202
                chanOpts = append(chanOpts, lnwallet.WithLeafStore(s))
×
5203
        })
×
5204
        p.cfg.AuxSigner.WhenSome(func(s lnwallet.AuxSigner) {
3✔
5205
                chanOpts = append(chanOpts, lnwallet.WithAuxSigner(s))
×
5206
        })
×
5207
        p.cfg.AuxResolver.WhenSome(func(s lnwallet.AuxContractResolver) {
3✔
5208
                chanOpts = append(chanOpts, lnwallet.WithAuxResolver(s))
×
5209
        })
×
5210

5211
        // If not already active, we'll add this channel to the set of active
5212
        // channels, so we can look it up later easily according to its channel
5213
        // ID.
5214
        lnChan, err := lnwallet.NewLightningChannel(
3✔
5215
                p.cfg.Signer, c.OpenChannel, p.cfg.SigPool, chanOpts...,
3✔
5216
        )
3✔
5217
        if err != nil {
3✔
5218
                return fmt.Errorf("unable to create LightningChannel: %w", err)
×
5219
        }
×
5220

5221
        // Store the channel in the activeChannels map.
5222
        p.activeChannels.Store(chanID, lnChan)
3✔
5223

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

3✔
5226
        // Next, we'll assemble a ChannelLink along with the necessary items it
3✔
5227
        // needs to function.
3✔
5228
        chainEvents, err := p.cfg.ChainArb.SubscribeChannelEvents(chanPoint)
3✔
5229
        if err != nil {
3✔
5230
                return fmt.Errorf("unable to subscribe to chain events: %w",
×
5231
                        err)
×
5232
        }
×
5233

5234
        // We'll query the channel DB for the new channel's initial forwarding
5235
        // policies to determine the policy we start out with.
5236
        initialPolicy, err := p.cfg.ChannelDB.GetInitialForwardingPolicy(chanID)
3✔
5237
        if err != nil {
3✔
5238
                return fmt.Errorf("unable to query for initial forwarding "+
×
5239
                        "policy: %v", err)
×
5240
        }
×
5241

5242
        // Create the link and add it to the switch.
5243
        err = p.addLink(
3✔
5244
                &chanPoint, lnChan, initialPolicy, chainEvents,
3✔
5245
                shouldReestablish, fn.None[lnwire.Shutdown](),
3✔
5246
        )
3✔
5247
        if err != nil {
3✔
5248
                return fmt.Errorf("can't register new channel link(%v) with "+
×
5249
                        "peer", chanPoint)
×
5250
        }
×
5251

5252
        isTaprootChan := c.ChanType.IsTaproot()
3✔
5253

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

5261
        // Now that the link has been added above, we'll also init an RBF chan
5262
        // closer for this channel, but only if the new close feature is
5263
        // negotiated.
5264
        //
5265
        // Creating this here ensures that any shutdown messages sent will be
5266
        // automatically routed by the msg router.
5267
        if _, err := p.initRbfChanCloser(lnChan); err != nil {
3✔
5268
                p.activeChanCloses.Delete(chanID)
×
5269

×
5270
                return fmt.Errorf("unable to init RBF chan closer for new "+
×
5271
                        "chan: %w", err)
×
5272
        }
×
5273

5274
        return nil
3✔
5275
}
5276

5277
// handleNewActiveChannel handles a `newChannelMsg` request. Depending on we
5278
// know this channel ID or not, we'll either add it to the `activeChannels` map
5279
// or init the next revocation for it.
5280
func (p *Brontide) handleNewActiveChannel(req *newChannelMsg) {
3✔
5281
        newChan := req.channel
3✔
5282
        chanPoint := newChan.FundingOutpoint
3✔
5283
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
3✔
5284

3✔
5285
        // Only update RemoteNextRevocation if the channel is in the
3✔
5286
        // activeChannels map and if we added the link to the switch. Only
3✔
5287
        // active channels will be added to the switch.
3✔
5288
        if p.isActiveChannel(chanID) {
6✔
5289
                p.log.Infof("Already have ChannelPoint(%v), ignoring",
3✔
5290
                        chanPoint)
3✔
5291

3✔
5292
                // Handle it and close the err chan on the request.
3✔
5293
                close(req.err)
3✔
5294

3✔
5295
                // Update the next revocation point.
3✔
5296
                err := p.updateNextRevocation(newChan.OpenChannel)
3✔
5297
                if err != nil {
3✔
5298
                        p.log.Errorf(err.Error())
×
5299
                }
×
5300

5301
                return
3✔
5302
        }
5303

5304
        // This is a new channel, we now add it to the map.
5305
        if err := p.addActiveChannel(req.channel); err != nil {
3✔
5306
                // Log and send back the error to the request.
×
5307
                p.log.Errorf(err.Error())
×
5308
                req.err <- err
×
5309

×
5310
                return
×
5311
        }
×
5312

5313
        // Close the err chan if everything went fine.
5314
        close(req.err)
3✔
5315
}
5316

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

7✔
5324
        chanID := req.channelID
7✔
5325

7✔
5326
        // If we already have this channel, something is wrong with the funding
7✔
5327
        // flow as it will only be marked as active after `ChannelReady` is
7✔
5328
        // handled. In this case, we will do nothing but log an error, just in
7✔
5329
        // case this is a legit channel.
7✔
5330
        if p.isActiveChannel(chanID) {
8✔
5331
                p.log.Errorf("Channel(%v) is already active, ignoring "+
1✔
5332
                        "pending channel request", chanID)
1✔
5333

1✔
5334
                return
1✔
5335
        }
1✔
5336

5337
        // The channel has already been added, we will do nothing and return.
5338
        if p.isPendingChannel(chanID) {
7✔
5339
                p.log.Infof("Channel(%v) is already added, ignoring "+
1✔
5340
                        "pending channel request", chanID)
1✔
5341

1✔
5342
                return
1✔
5343
        }
1✔
5344

5345
        // This is a new channel, we now add it to the map `activeChannels`
5346
        // with nil value and mark it as a newly added channel in
5347
        // `addedChannels`.
5348
        p.activeChannels.Store(chanID, nil)
5✔
5349
        p.addedChannels.Store(chanID, struct{}{})
5✔
5350
}
5351

5352
// handleRemovePendingChannel takes a `newChannelMsg` request and removes it
5353
// from `activeChannels` map. The request will be ignored if the channel is
5354
// considered active by Brontide. Noop if the channel ID cannot be found.
5355
func (p *Brontide) handleRemovePendingChannel(req *newChannelMsg) {
7✔
5356
        defer close(req.err)
7✔
5357

7✔
5358
        chanID := req.channelID
7✔
5359

7✔
5360
        // If we already have this channel, something is wrong with the funding
7✔
5361
        // flow as it will only be marked as active after `ChannelReady` is
7✔
5362
        // handled. In this case, we will log an error and exit.
7✔
5363
        if p.isActiveChannel(chanID) {
8✔
5364
                p.log.Errorf("Channel(%v) is active, ignoring remove request",
1✔
5365
                        chanID)
1✔
5366
                return
1✔
5367
        }
1✔
5368

5369
        // The channel has not been added yet, we will log a warning as there
5370
        // is an unexpected call from funding manager.
5371
        if !p.isPendingChannel(chanID) {
10✔
5372
                p.log.Warnf("Channel(%v) not found, removing it anyway", chanID)
4✔
5373
        }
4✔
5374

5375
        // Remove the record of this pending channel.
5376
        p.activeChannels.Delete(chanID)
6✔
5377
        p.addedChannels.Delete(chanID)
6✔
5378
}
5379

5380
// sendLinkUpdateMsg sends a message that updates the channel to the
5381
// channel's message stream.
5382
func (p *Brontide) sendLinkUpdateMsg(cid lnwire.ChannelID, msg lnwire.Message) {
3✔
5383
        p.log.Tracef("Sending link update msg=%v", msg.MsgType())
3✔
5384

3✔
5385
        chanStream, ok := p.activeMsgStreams[cid]
3✔
5386
        if !ok {
6✔
5387
                // If a stream hasn't yet been created, then we'll do so, add
3✔
5388
                // it to the map, and finally start it.
3✔
5389
                chanStream = newChanMsgStream(p, cid)
3✔
5390
                p.activeMsgStreams[cid] = chanStream
3✔
5391
                chanStream.Start()
3✔
5392

3✔
5393
                // Stop the stream when quit.
3✔
5394
                go func() {
6✔
5395
                        <-p.cg.Done()
3✔
5396
                        chanStream.Stop()
3✔
5397
                }()
3✔
5398
        }
5399

5400
        // With the stream obtained, add the message to the stream so we can
5401
        // continue processing message.
5402
        chanStream.AddMsg(msg)
3✔
5403
}
5404

5405
// scaleTimeout multiplies the argument duration by a constant factor depending
5406
// on variious heuristics. Currently this is only used to check whether our peer
5407
// appears to be connected over Tor and relaxes the timout deadline. However,
5408
// this is subject to change and should be treated as opaque.
5409
func (p *Brontide) scaleTimeout(timeout time.Duration) time.Duration {
70✔
5410
        if p.isTorConnection {
73✔
5411
                return timeout * time.Duration(torTimeoutMultiplier)
3✔
5412
        }
3✔
5413

5414
        return timeout
67✔
5415
}
5416

5417
// CoopCloseUpdates is a struct used to communicate updates for an active close
5418
// to the caller.
5419
type CoopCloseUpdates struct {
5420
        UpdateChan chan interface{}
5421

5422
        ErrChan chan error
5423
}
5424

5425
// ChanHasRbfCoopCloser returns true if the channel as identifier by the channel
5426
// point has an active RBF chan closer.
5427
func (p *Brontide) ChanHasRbfCoopCloser(chanPoint wire.OutPoint) bool {
3✔
5428
        chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
3✔
5429
        chanCloser, found := p.activeChanCloses.Load(chanID)
3✔
5430
        if !found {
6✔
5431
                return false
3✔
5432
        }
3✔
5433

5434
        return chanCloser.IsRight()
3✔
5435
}
5436

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

3✔
5445
        // If RBF coop close isn't permitted, then we'll an error.
3✔
5446
        if !p.rbfCoopCloseAllowed() {
3✔
5447
                return nil, fmt.Errorf("rbf coop close not enabled for " +
×
5448
                        "channel")
×
5449
        }
×
5450

5451
        closeUpdates := &CoopCloseUpdates{
3✔
5452
                UpdateChan: make(chan interface{}, 1),
3✔
5453
                ErrChan:    make(chan error, 1),
3✔
5454
        }
3✔
5455

3✔
5456
        // We'll re-use the existing switch struct here, even though we're
3✔
5457
        // bypassing the switch entirely.
3✔
5458
        closeReq := htlcswitch.ChanClose{
3✔
5459
                CloseType:      contractcourt.CloseRegular,
3✔
5460
                ChanPoint:      &chanPoint,
3✔
5461
                TargetFeePerKw: feeRate,
3✔
5462
                DeliveryScript: deliveryScript,
3✔
5463
                Updates:        closeUpdates.UpdateChan,
3✔
5464
                Err:            closeUpdates.ErrChan,
3✔
5465
                Ctx:            ctx,
3✔
5466
        }
3✔
5467

3✔
5468
        err := p.startRbfChanCloser(newRPCShutdownInit(&closeReq), chanPoint)
3✔
5469
        if err != nil {
3✔
5470
                return nil, err
×
5471
        }
×
5472

5473
        return closeUpdates, nil
3✔
5474
}
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