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

lightningnetwork / lnd / 16177692143

09 Jul 2025 06:49PM UTC coverage: 55.317% (-2.3%) from 57.611%
16177692143

Pull #10060

github

web-flow
Merge 4aec413e3 into 0e830da9d
Pull Request #10060: sweep: fix expected spending events being missed

9 of 25 new or added lines in 1 file covered. (36.0%)

23713 existing lines in 281 files now uncovered.

108499 of 196142 relevant lines covered (55.32%)

22331.52 hits per line

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

65.15
/routing/localchans/manager.go
1
package localchans
2

3
import (
4
        "bytes"
5
        "context"
6
        "errors"
7
        "fmt"
8
        "sync"
9
        "time"
10

11
        "github.com/btcsuite/btcd/btcec/v2"
12
        "github.com/btcsuite/btcd/wire"
13
        "github.com/lightningnetwork/lnd/channeldb"
14
        "github.com/lightningnetwork/lnd/discovery"
15
        "github.com/lightningnetwork/lnd/fn/v2"
16
        "github.com/lightningnetwork/lnd/graph/db/models"
17
        "github.com/lightningnetwork/lnd/lnrpc"
18
        "github.com/lightningnetwork/lnd/lnwire"
19
        "github.com/lightningnetwork/lnd/routing"
20
)
21

22
// Manager manages the node's local channels. The only operation that is
23
// currently implemented is updating forwarding policies.
24
type Manager struct {
25
        // SelfPub contains the public key of the local node.
26
        SelfPub *btcec.PublicKey
27

28
        // DefaultRoutingPolicy is the default routing policy.
29
        DefaultRoutingPolicy models.ForwardingPolicy
30

31
        // UpdateForwardingPolicies is used by the manager to update active
32
        // links with a new policy.
33
        UpdateForwardingPolicies func(
34
                chanPolicies map[wire.OutPoint]models.ForwardingPolicy)
35

36
        // PropagateChanPolicyUpdate is called to persist a new policy to disk
37
        // and broadcast it to the network.
38
        PropagateChanPolicyUpdate func(
39
                edgesToUpdate []discovery.EdgeWithInfo) error
40

41
        // ForAllOutgoingChannels is required to iterate over all our local
42
        // channels. The ChannelEdgePolicy parameter may be nil.
43
        ForAllOutgoingChannels func(cb func(*models.ChannelEdgeInfo,
44
                *models.ChannelEdgePolicy) error) error
45

46
        // FetchChannel is used to query local channel parameters. Optionally an
47
        // existing db tx can be supplied.
48
        FetchChannel func(chanPoint wire.OutPoint) (*channeldb.OpenChannel,
49
                error)
50

51
        // AddEdge is used to add edge/channel to the topology of the router.
52
        AddEdge func(ctx context.Context, edge *models.ChannelEdgeInfo) error
53

54
        // policyUpdateLock ensures that the database and the link do not fall
55
        // out of sync if there are concurrent fee update calls. Without it,
56
        // there is a chance that policy A updates the database, then policy B
57
        // updates the database, then policy B updates the link, then policy A
58
        // updates the link.
59
        policyUpdateLock sync.Mutex
60
}
61

62
// UpdatePolicy updates the policy for the specified channels on disk and in
63
// the active links.
64
func (r *Manager) UpdatePolicy(ctx context.Context,
65
        newSchema routing.ChannelPolicy,
66
        createMissingEdge bool, chanPoints ...wire.OutPoint) (
67
        []*lnrpc.FailedUpdate, error) {
6✔
68

6✔
69
        r.policyUpdateLock.Lock()
6✔
70
        defer r.policyUpdateLock.Unlock()
6✔
71

6✔
72
        // First, we'll construct a set of all the channels that we are
6✔
73
        // trying to update.
6✔
74
        unprocessedChans := make(map[wire.OutPoint]struct{})
6✔
75
        for _, chanPoint := range chanPoints {
11✔
76
                unprocessedChans[chanPoint] = struct{}{}
5✔
77
        }
5✔
78

79
        haveChanFilter := len(unprocessedChans) != 0
6✔
80

6✔
81
        var failedUpdates []*lnrpc.FailedUpdate
6✔
82
        var edgesToUpdate []discovery.EdgeWithInfo
6✔
83
        policiesToUpdate := make(map[wire.OutPoint]models.ForwardingPolicy)
6✔
84

6✔
85
        // NOTE: edge may be nil when this function is called.
6✔
86
        processChan := func(info *models.ChannelEdgeInfo,
6✔
87
                edge *models.ChannelEdgePolicy) error {
11✔
88

5✔
89
                // If we have a channel filter, and this channel isn't a part
5✔
90
                // of it, then we'll skip it.
5✔
91
                _, ok := unprocessedChans[info.ChannelPoint]
5✔
92
                if !ok && haveChanFilter {
6✔
93
                        return nil
1✔
94
                }
1✔
95

96
                // Mark this channel as found by removing it. unprocessedChans
97
                // will be used to report invalid channels later on.
98
                delete(unprocessedChans, info.ChannelPoint)
4✔
99

4✔
100
                if edge == nil {
4✔
101
                        log.Errorf("Got nil channel edge policy when updating "+
×
102
                                "a channel. Channel point: %v",
×
103
                                info.ChannelPoint.String())
×
104

×
105
                        failedUpdates = append(failedUpdates, makeFailureItem(
×
106
                                info.ChannelPoint,
×
107
                                lnrpc.UpdateFailure_UPDATE_FAILURE_NOT_FOUND,
×
108
                                "edge policy not found",
×
109
                        ))
×
110

×
111
                        return nil
×
112
                }
×
113

114
                // Apply the new policy to the edge.
115
                err := r.updateEdge(info.ChannelPoint, edge, newSchema)
4✔
116
                if err != nil {
4✔
117
                        failedUpdates = append(failedUpdates,
×
118
                                makeFailureItem(info.ChannelPoint,
×
119
                                        lnrpc.UpdateFailure_UPDATE_FAILURE_INVALID_PARAMETER,
×
120
                                        err.Error(),
×
121
                                ))
×
122

×
123
                        return nil
×
124
                }
×
125

126
                // Add updated edge to list of edges to send to gossiper.
127
                edgesToUpdate = append(edgesToUpdate, discovery.EdgeWithInfo{
4✔
128
                        Info: info,
4✔
129
                        Edge: edge,
4✔
130
                })
4✔
131

4✔
132
                var inboundWireFee lnwire.Fee
4✔
133
                edge.InboundFee.WhenSome(func(fee lnwire.Fee) {
4✔
UNCOV
134
                        inboundWireFee = fee
×
UNCOV
135
                })
×
136
                inboundFee := models.NewInboundFeeFromWire(inboundWireFee)
4✔
137

4✔
138
                // Add updated policy to list of policies to send to switch.
4✔
139
                policiesToUpdate[info.ChannelPoint] = models.ForwardingPolicy{
4✔
140
                        BaseFee:       edge.FeeBaseMSat,
4✔
141
                        FeeRate:       edge.FeeProportionalMillionths,
4✔
142
                        TimeLockDelta: uint32(edge.TimeLockDelta),
4✔
143
                        MinHTLCOut:    edge.MinHTLC,
4✔
144
                        MaxHTLC:       edge.MaxHTLC,
4✔
145
                        InboundFee:    inboundFee,
4✔
146
                }
4✔
147

4✔
148
                return nil
4✔
149
        }
150

151
        // Next, we'll loop over all the outgoing channels the router knows of.
152
        // If we have a filter then we'll only collect those channels, otherwise
153
        // we'll collect them all.
154
        err := r.ForAllOutgoingChannels(processChan)
6✔
155
        if err != nil {
6✔
156
                return nil, err
×
157
        }
×
158

159
        // Construct a list of failed policy updates.
160
        for chanPoint := range unprocessedChans {
9✔
161
                channel, err := r.FetchChannel(chanPoint)
3✔
162
                switch {
3✔
163
                case errors.Is(err, channeldb.ErrChannelNotFound):
1✔
164
                        failedUpdates = append(failedUpdates,
1✔
165
                                makeFailureItem(chanPoint,
1✔
166
                                        lnrpc.UpdateFailure_UPDATE_FAILURE_NOT_FOUND,
1✔
167
                                        "not found",
1✔
168
                                ))
1✔
169

170
                case err != nil:
×
171
                        failedUpdates = append(failedUpdates,
×
172
                                makeFailureItem(chanPoint,
×
173
                                        lnrpc.UpdateFailure_UPDATE_FAILURE_INTERNAL_ERR,
×
174
                                        err.Error(),
×
175
                                ))
×
176

177
                case channel.IsPending:
×
178
                        failedUpdates = append(failedUpdates,
×
179
                                makeFailureItem(chanPoint,
×
180
                                        lnrpc.UpdateFailure_UPDATE_FAILURE_PENDING,
×
181
                                        "not yet confirmed",
×
182
                                ))
×
183

184
                // If the edge was not found, but the channel is found, that
185
                // means the edge is missing in the graph database and should be
186
                // recreated. The edge and policy are created in-memory. The
187
                // edge is inserted in createEdge below and the policy will be
188
                // added to the graph in the PropagateChanPolicyUpdate call
189
                // below.
190
                case createMissingEdge:
1✔
191
                        log.Warnf("Missing edge for active channel (%s) "+
1✔
192
                                "during policy update. Recreating edge with "+
1✔
193
                                "default policy.",
1✔
194
                                channel.FundingOutpoint.String())
1✔
195

1✔
196
                        info, edge, failedUpdate := r.createMissingEdge(
1✔
197
                                ctx, channel, newSchema,
1✔
198
                        )
1✔
199
                        if failedUpdate == nil {
2✔
200
                                err = processChan(info, edge)
1✔
201
                                if err != nil {
1✔
202
                                        return nil, err
×
203
                                }
×
204
                        } else {
×
205
                                failedUpdates = append(
×
206
                                        failedUpdates, failedUpdate,
×
207
                                )
×
208
                        }
×
209

210
                default:
1✔
211
                        log.Warnf("Missing edge for active channel (%s) "+
1✔
212
                                "during policy update. Could not update "+
1✔
213
                                "policy.", channel.FundingOutpoint.String())
1✔
214

1✔
215
                        failedUpdates = append(failedUpdates,
1✔
216
                                makeFailureItem(chanPoint,
1✔
217
                                        lnrpc.UpdateFailure_UPDATE_FAILURE_UNKNOWN,
1✔
218
                                        "could not update policies",
1✔
219
                                ))
1✔
220
                }
221
        }
222

223
        // Commit the policy updates to disk and broadcast to the network. We
224
        // validated the new policy above, so we expect no validation errors. If
225
        // this would happen because of a bug, the link policy will be
226
        // desynchronized. It is currently not possible to atomically commit
227
        // multiple edge updates.
228
        err = r.PropagateChanPolicyUpdate(edgesToUpdate)
6✔
229
        if err != nil {
6✔
230
                return nil, err
×
231
        }
×
232

233
        // Update active links.
234
        r.UpdateForwardingPolicies(policiesToUpdate)
6✔
235

6✔
236
        return failedUpdates, nil
6✔
237
}
238

239
func (r *Manager) createMissingEdge(ctx context.Context,
240
        channel *channeldb.OpenChannel,
241
        newSchema routing.ChannelPolicy) (*models.ChannelEdgeInfo,
242
        *models.ChannelEdgePolicy, *lnrpc.FailedUpdate) {
1✔
243

1✔
244
        info, edge, err := r.createEdge(channel, time.Now())
1✔
245
        if err != nil {
1✔
246
                log.Errorf("Failed to recreate missing edge "+
×
247
                        "for channel (%s): %v",
×
248
                        channel.FundingOutpoint.String(), err)
×
249

×
250
                return nil, nil, makeFailureItem(
×
251
                        channel.FundingOutpoint,
×
252
                        lnrpc.UpdateFailure_UPDATE_FAILURE_UNKNOWN,
×
253
                        "could not update policies",
×
254
                )
×
255
        }
×
256

257
        // Validate the newly created edge policy with the user defined new
258
        // schema before adding the edge to the database.
259
        err = r.updateEdge(channel.FundingOutpoint, edge, newSchema)
1✔
260
        if err != nil {
1✔
261
                return nil, nil, makeFailureItem(
×
262
                        info.ChannelPoint,
×
263
                        lnrpc.UpdateFailure_UPDATE_FAILURE_INVALID_PARAMETER,
×
264
                        err.Error(),
×
265
                )
×
266
        }
×
267

268
        // Insert the edge into the database to avoid `edge not
269
        // found` errors during policy update propagation.
270
        err = r.AddEdge(ctx, info)
1✔
271
        if err != nil {
1✔
272
                log.Errorf("Attempt to add missing edge for "+
×
273
                        "channel (%s) errored with: %v",
×
274
                        channel.FundingOutpoint.String(), err)
×
275

×
276
                return nil, nil, makeFailureItem(
×
277
                        channel.FundingOutpoint,
×
278
                        lnrpc.UpdateFailure_UPDATE_FAILURE_UNKNOWN,
×
279
                        "could not add edge",
×
280
                )
×
281
        }
×
282

283
        return info, edge, nil
1✔
284
}
285

286
// createEdge recreates an edge and policy from an open channel in-memory.
287
func (r *Manager) createEdge(channel *channeldb.OpenChannel,
288
        timestamp time.Time) (*models.ChannelEdgeInfo,
289
        *models.ChannelEdgePolicy, error) {
3✔
290

3✔
291
        nodeKey1Bytes := r.SelfPub.SerializeCompressed()
3✔
292
        nodeKey2Bytes := channel.IdentityPub.SerializeCompressed()
3✔
293
        bitcoinKey1Bytes := channel.LocalChanCfg.MultiSigKey.PubKey.
3✔
294
                SerializeCompressed()
3✔
295
        bitcoinKey2Bytes := channel.RemoteChanCfg.MultiSigKey.PubKey.
3✔
296
                SerializeCompressed()
3✔
297
        channelFlags := lnwire.ChanUpdateChanFlags(0)
3✔
298

3✔
299
        // Make it such that node_id_1 is the lexicographically-lesser of the
3✔
300
        // two compressed keys sorted in ascending lexicographic order.
3✔
301
        if bytes.Compare(nodeKey2Bytes, nodeKey1Bytes) < 0 {
4✔
302
                nodeKey1Bytes, nodeKey2Bytes = nodeKey2Bytes, nodeKey1Bytes
1✔
303
                bitcoinKey1Bytes, bitcoinKey2Bytes = bitcoinKey2Bytes,
1✔
304
                        bitcoinKey1Bytes
1✔
305
                channelFlags = 1
1✔
306
        }
1✔
307

308
        // We need to make sure we use the real scid for public confirmed
309
        // zero-conf channels.
310
        shortChanID := channel.ShortChanID()
3✔
311
        isPublic := channel.ChannelFlags&lnwire.FFAnnounceChannel != 0
3✔
312
        if isPublic && channel.IsZeroConf() && channel.ZeroConfConfirmed() {
3✔
313
                shortChanID = channel.ZeroConfRealScid()
×
314
        }
×
315

316
        info := &models.ChannelEdgeInfo{
3✔
317
                ChannelID:    shortChanID.ToUint64(),
3✔
318
                ChainHash:    channel.ChainHash,
3✔
319
                Features:     lnwire.EmptyFeatureVector(),
3✔
320
                Capacity:     channel.Capacity,
3✔
321
                ChannelPoint: channel.FundingOutpoint,
3✔
322
        }
3✔
323

3✔
324
        copy(info.NodeKey1Bytes[:], nodeKey1Bytes)
3✔
325
        copy(info.NodeKey2Bytes[:], nodeKey2Bytes)
3✔
326
        copy(info.BitcoinKey1Bytes[:], bitcoinKey1Bytes)
3✔
327
        copy(info.BitcoinKey2Bytes[:], bitcoinKey2Bytes)
3✔
328

3✔
329
        // Construct a dummy channel edge policy with default values that will
3✔
330
        // be updated with the new values in the call to processChan below.
3✔
331
        timeLockDelta := uint16(r.DefaultRoutingPolicy.TimeLockDelta)
3✔
332
        edge := &models.ChannelEdgePolicy{
3✔
333
                ChannelID:                 shortChanID.ToUint64(),
3✔
334
                LastUpdate:                timestamp,
3✔
335
                TimeLockDelta:             timeLockDelta,
3✔
336
                ChannelFlags:              channelFlags,
3✔
337
                MessageFlags:              lnwire.ChanUpdateRequiredMaxHtlc,
3✔
338
                FeeBaseMSat:               r.DefaultRoutingPolicy.BaseFee,
3✔
339
                FeeProportionalMillionths: r.DefaultRoutingPolicy.FeeRate,
3✔
340
                MinHTLC:                   r.DefaultRoutingPolicy.MinHTLCOut,
3✔
341
                MaxHTLC:                   r.DefaultRoutingPolicy.MaxHTLC,
3✔
342
        }
3✔
343

3✔
344
        copy(edge.ToNode[:], channel.IdentityPub.SerializeCompressed())
3✔
345

3✔
346
        return info, edge, nil
3✔
347
}
348

349
// updateEdge updates the given edge with the new schema.
350
func (r *Manager) updateEdge(chanPoint wire.OutPoint,
351
        edge *models.ChannelEdgePolicy,
352
        newSchema routing.ChannelPolicy) error {
5✔
353

5✔
354
        channel, err := r.FetchChannel(chanPoint)
5✔
355
        if err != nil {
5✔
356
                return err
×
357
        }
×
358

359
        // Update forwarding fee scheme and required time lock delta.
360
        edge.FeeBaseMSat = newSchema.BaseFee
5✔
361
        edge.FeeProportionalMillionths = lnwire.MilliSatoshi(
5✔
362
                newSchema.FeeRate,
5✔
363
        )
5✔
364

5✔
365
        // If inbound fees are set, we update the edge with them.
5✔
366
        err = fn.MapOptionZ(newSchema.InboundFee,
5✔
367
                func(f models.InboundFee) error {
5✔
UNCOV
368
                        inboundWireFee := f.ToWire()
×
UNCOV
369
                        edge.InboundFee = fn.Some(inboundWireFee)
×
UNCOV
370

×
UNCOV
371
                        return edge.ExtraOpaqueData.PackRecords(
×
UNCOV
372
                                &inboundWireFee,
×
UNCOV
373
                        )
×
UNCOV
374
                })
×
375
        if err != nil {
5✔
376
                return err
×
377
        }
×
378

379
        edge.TimeLockDelta = uint16(newSchema.TimeLockDelta)
5✔
380

5✔
381
        // Retrieve negotiated channel htlc amt limits.
5✔
382
        amtMin, amtMax, err := r.getHtlcAmtLimits(channel)
5✔
383
        if err != nil {
5✔
384
                return err
×
385
        }
×
386

387
        // We now update the edge max htlc value.
388
        switch {
5✔
389
        // If a non-zero max htlc was specified, use it to update the edge.
390
        // Otherwise keep the value unchanged.
391
        case newSchema.MaxHTLC != 0:
4✔
392
                edge.MaxHTLC = newSchema.MaxHTLC
4✔
393

394
        // If this edge still doesn't have a max htlc set, set it to the max.
395
        // This is an on-the-fly migration.
396
        case !edge.MessageFlags.HasMaxHtlc():
×
397
                edge.MaxHTLC = amtMax
×
398

399
        // If this edge has a max htlc that exceeds what the channel can
400
        // actually carry, correct it now. This can happen, because we
401
        // previously set the max htlc to the channel capacity.
402
        case edge.MaxHTLC > amtMax:
×
403
                edge.MaxHTLC = amtMax
×
404
        }
405

406
        // If a new min htlc is specified, update the edge.
407
        if newSchema.MinHTLC != nil {
5✔
408
                edge.MinHTLC = *newSchema.MinHTLC
×
409
        }
×
410

411
        // If the MaxHtlc flag wasn't already set, we can set it now.
412
        edge.MessageFlags |= lnwire.ChanUpdateRequiredMaxHtlc
5✔
413

5✔
414
        // Validate htlc amount constraints.
5✔
415
        switch {
5✔
416
        case edge.MinHTLC < amtMin:
×
417
                return fmt.Errorf(
×
418
                        "min htlc amount of %v is below min htlc parameter of %v",
×
419
                        edge.MinHTLC, amtMin,
×
420
                )
×
421

422
        case edge.MaxHTLC > amtMax:
×
423
                return fmt.Errorf(
×
424
                        "max htlc size of %v is above max pending amount of %v",
×
425
                        edge.MaxHTLC, amtMax,
×
426
                )
×
427

428
        case edge.MinHTLC > edge.MaxHTLC:
×
429
                return fmt.Errorf(
×
430
                        "min_htlc %v greater than max_htlc %v",
×
431
                        edge.MinHTLC, edge.MaxHTLC,
×
432
                )
×
433
        }
434

435
        // Clear signature to help prevent usage of the previous signature.
436
        edge.SetSigBytes(nil)
5✔
437

5✔
438
        return nil
5✔
439
}
440

441
// getHtlcAmtLimits retrieves the negotiated channel min and max htlc amount
442
// constraints.
443
func (r *Manager) getHtlcAmtLimits(ch *channeldb.OpenChannel) (
444
        lnwire.MilliSatoshi, lnwire.MilliSatoshi, error) {
5✔
445

5✔
446
        // The max htlc policy field must be less than or equal to the channel
5✔
447
        // capacity AND less than or equal to the max in-flight HTLC value.
5✔
448
        // Since the latter is always less than or equal to the former, just
5✔
449
        // return the max in-flight value.
5✔
450
        maxAmt := ch.LocalChanCfg.ChannelStateBounds.MaxPendingAmount
5✔
451

5✔
452
        return ch.LocalChanCfg.MinHTLC, maxAmt, nil
5✔
453
}
5✔
454

455
// makeFailureItem creates a lnrpc.FailedUpdate object.
456
func makeFailureItem(outPoint wire.OutPoint, updateFailure lnrpc.UpdateFailure,
457
        errStr string) *lnrpc.FailedUpdate {
2✔
458

2✔
459
        outpoint := lnrpc.MarshalOutPoint(&outPoint)
2✔
460

2✔
461
        return &lnrpc.FailedUpdate{
2✔
462
                Outpoint:    outpoint,
2✔
463
                Reason:      updateFailure,
2✔
464
                UpdateError: errStr,
2✔
465
        }
2✔
466
}
2✔
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