routing: remove a context.TODO()
This commit is contained in:
4 files changed
+20
-11
No files matched your search
@@ -2,6 +2,7 @@ package localchans
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -48,7 +49,7 @@ type Manager struct {
|
|||||||
error)
|
error)
|
||||||
|
|
||||||
// AddEdge is used to add edge/channel to the topology of the router.
|
// AddEdge is used to add edge/channel to the topology of the router.
|
||||||
AddEdge func(edge *models.ChannelEdgeInfo) error
|
AddEdge func(ctx context.Context, edge *models.ChannelEdgeInfo) error
|
||||||
|
|
||||||
// policyUpdateLock ensures that the database and the link do not fall
|
// policyUpdateLock ensures that the database and the link do not fall
|
||||||
// out of sync if there are concurrent fee update calls. Without it,
|
// out of sync if there are concurrent fee update calls. Without it,
|
||||||
@@ -60,7 +61,8 @@ type Manager struct {
|
|||||||
|
|
||||||
// UpdatePolicy updates the policy for the specified channels on disk and in
|
// UpdatePolicy updates the policy for the specified channels on disk and in
|
||||||
// the active links.
|
// the active links.
|
||||||
func (r *Manager) UpdatePolicy(newSchema routing.ChannelPolicy,
|
func (r *Manager) UpdatePolicy(ctx context.Context,
|
||||||
|
newSchema routing.ChannelPolicy,
|
||||||
createMissingEdge bool, chanPoints ...wire.OutPoint) (
|
createMissingEdge bool, chanPoints ...wire.OutPoint) (
|
||||||
[]*lnrpc.FailedUpdate, error) {
|
[]*lnrpc.FailedUpdate, error) {
|
||||||
|
|
||||||
@@ -192,7 +194,7 @@ func (r *Manager) UpdatePolicy(newSchema routing.ChannelPolicy,
|
|||||||
channel.FundingOutpoint.String())
|
channel.FundingOutpoint.String())
|
||||||
|
|
||||||
info, edge, failedUpdate := r.createMissingEdge(
|
info, edge, failedUpdate := r.createMissingEdge(
|
||||||
channel, newSchema,
|
ctx, channel, newSchema,
|
||||||
)
|
)
|
||||||
if failedUpdate == nil {
|
if failedUpdate == nil {
|
||||||
err = processChan(info, edge)
|
err = processChan(info, edge)
|
||||||
@@ -234,7 +236,8 @@ func (r *Manager) UpdatePolicy(newSchema routing.ChannelPolicy,
|
|||||||
return failedUpdates, nil
|
return failedUpdates, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *Manager) createMissingEdge(channel *channeldb.OpenChannel,
|
func (r *Manager) createMissingEdge(ctx context.Context,
|
||||||
|
channel *channeldb.OpenChannel,
|
||||||
newSchema routing.ChannelPolicy) (*models.ChannelEdgeInfo,
|
newSchema routing.ChannelPolicy) (*models.ChannelEdgeInfo,
|
||||||
*models.ChannelEdgePolicy, *lnrpc.FailedUpdate) {
|
*models.ChannelEdgePolicy, *lnrpc.FailedUpdate) {
|
||||||
|
|
||||||
@@ -264,7 +267,7 @@ func (r *Manager) createMissingEdge(channel *channeldb.OpenChannel,
|
|||||||
|
|
||||||
// Insert the edge into the database to avoid `edge not
|
// Insert the edge into the database to avoid `edge not
|
||||||
// found` errors during policy update propagation.
|
// found` errors during policy update propagation.
|
||||||
err = r.AddEdge(info)
|
err = r.AddEdge(ctx, info)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("Attempt to add missing edge for "+
|
log.Errorf("Attempt to add missing edge for "+
|
||||||
"channel (%s) errored with: %v",
|
"channel (%s) errored with: %v",
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package localchans
|
package localchans
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -163,7 +164,7 @@ func TestManager(t *testing.T) {
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
addEdge := func(edge *models.ChannelEdgeInfo) error {
|
addEdge := func(_ context.Context, _ *models.ChannelEdgeInfo) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -314,7 +315,9 @@ func TestManager(t *testing.T) {
|
|||||||
channelSet = test.channelSet
|
channelSet = test.channelSet
|
||||||
expectedNumUpdates = test.expectedNumUpdates
|
expectedNumUpdates = test.expectedNumUpdates
|
||||||
|
|
||||||
failedUpdates, err := manager.UpdatePolicy(test.newPolicy,
|
failedUpdates, err := manager.UpdatePolicy(
|
||||||
|
context.Background(),
|
||||||
|
test.newPolicy,
|
||||||
test.createMissingEdge,
|
test.createMissingEdge,
|
||||||
test.specifiedChanPoints...)
|
test.specifiedChanPoints...)
|
||||||
|
|
||||||
|
|||||||
+3
-2
@@ -7936,8 +7936,9 @@ func (r *rpcServer) UpdateChannelPolicy(ctx context.Context,
|
|||||||
|
|
||||||
// With the scope resolved, we'll now send this to the local channel
|
// With the scope resolved, we'll now send this to the local channel
|
||||||
// manager so it can propagate the new policy for our target channel(s).
|
// manager so it can propagate the new policy for our target channel(s).
|
||||||
failedUpdates, err := r.server.localChanMgr.UpdatePolicy(chanPolicy,
|
failedUpdates, err := r.server.localChanMgr.UpdatePolicy(
|
||||||
req.CreateMissingEdge, targetChans...)
|
ctx, chanPolicy, req.CreateMissingEdge, targetChans...,
|
||||||
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1224,8 +1224,10 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
|
|||||||
PropagateChanPolicyUpdate: s.authGossiper.PropagateChanPolicyUpdate,
|
PropagateChanPolicyUpdate: s.authGossiper.PropagateChanPolicyUpdate,
|
||||||
UpdateForwardingPolicies: s.htlcSwitch.UpdateForwardingPolicies,
|
UpdateForwardingPolicies: s.htlcSwitch.UpdateForwardingPolicies,
|
||||||
FetchChannel: s.chanStateDB.FetchChannel,
|
FetchChannel: s.chanStateDB.FetchChannel,
|
||||||
AddEdge: func(edge *models.ChannelEdgeInfo) error {
|
AddEdge: func(ctx context.Context,
|
||||||
return s.graphBuilder.AddEdge(context.TODO(), edge)
|
edge *models.ChannelEdgeInfo) error {
|
||||||
|
|
||||||
|
return s.graphBuilder.AddEdge(ctx, edge)
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user