forked from lightningnetwork/lnd
-
Notifications
You must be signed in to change notification settings - Fork 0
/
pilot.go
292 lines (254 loc) · 8.1 KB
/
pilot.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
package main
import (
"fmt"
"net"
"github.com/btcsuite/btcd/btcec"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil"
"github.com/davecgh/go-spew/spew"
"github.com/lightningnetwork/lnd/autopilot"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/tor"
)
// chanController is an implementation of the autopilot.ChannelController
// interface that's backed by a running lnd instance.
type chanController struct {
server *server
}
// OpenChannel opens a channel to a target peer, with a capacity of the
// specified amount. This function should un-block immediately after the
// funding transaction that marks the channel open has been broadcast.
func (c *chanController) OpenChannel(target *btcec.PublicKey,
amt btcutil.Amount, addrs []net.Addr) error {
// We can't establish a channel if no addresses were provided for the
// peer.
if len(addrs) == 0 {
return fmt.Errorf("Unable to create channel w/o an active " +
"address")
}
// First, we'll check if we're already connected to the target peer. If
// not, then we'll need to establish a connection.
if _, err := c.server.FindPeer(target); err != nil {
// TODO(roasbeef): try teach addr
atplLog.Tracef("Connecting to %x to auto-create channel: ",
target.SerializeCompressed())
lnAddr := &lnwire.NetAddress{
IdentityKey: target,
ChainNet: activeNetParams.Net,
}
// We'll attempt to successively connect to each of the
// advertised IP addresses until we've either exhausted the
// advertised IP addresses, or have made a connection.
var connected bool
for _, addr := range addrs {
switch addr.(type) {
case *net.TCPAddr, *tor.OnionAddr:
lnAddr.Address = addr
default:
return fmt.Errorf("unknown address type %T", addr)
}
// TODO(roasbeef): make perm connection in server after
// chan open?
err := c.server.ConnectToPeer(lnAddr, false)
if err != nil {
// If we weren't able to connect to the peer,
// then we'll move onto the next.
continue
}
connected = true
break
}
// If we weren't able to establish a connection at all, then
// we'll error out.
if !connected {
return fmt.Errorf("Unable to connect to %x",
target.SerializeCompressed())
}
}
// With the connection established, we'll now establish our connection
// to the target peer, waiting for the first update before we exit.
feePerVSize, err := c.server.cc.feeEstimator.EstimateFeePerVSize(3)
if err != nil {
return err
}
// TODO(halseth): make configurable?
minHtlc := lnwire.NewMSatFromSatoshis(1)
updateStream, errChan := c.server.OpenChannel(target, amt, 0,
minHtlc, feePerVSize, false, 0)
select {
case err := <-errChan:
// If we were not able to actually open a channel to the peer
// for whatever reason, then we'll disconnect from the peer to
// ensure we don't accumulate a bunch of unnecessary
// connections.
if err != nil {
dcErr := c.server.DisconnectPeer(target)
if dcErr != nil {
atplLog.Errorf("Unable to disconnect from peer %v",
target.SerializeCompressed())
}
}
return err
case <-updateStream:
return nil
case <-c.server.quit:
return nil
}
}
func (c *chanController) CloseChannel(chanPoint *wire.OutPoint) error {
return nil
}
func (c *chanController) SpliceIn(chanPoint *wire.OutPoint,
amt btcutil.Amount) (*autopilot.Channel, error) {
return nil, nil
}
func (c *chanController) SpliceOut(chanPoint *wire.OutPoint,
amt btcutil.Amount) (*autopilot.Channel, error) {
return nil, nil
}
// A compile time assertion to ensure chanController meets the
// autopilot.ChannelController interface.
var _ autopilot.ChannelController = (*chanController)(nil)
// initAutoPilot initializes a new autopilot.Agent instance based on the passed
// configuration struct. All interfaces needed to drive the pilot will be
// registered and launched.
func initAutoPilot(svr *server, cfg *autoPilotConfig) (*autopilot.Agent, error) {
atplLog.Infof("Instantiating autopilot with cfg: %v", spew.Sdump(cfg))
// First, we'll create the preferential attachment heuristic,
// initialized with the passed auto pilot configuration parameters.
prefAttachment := autopilot.NewConstrainedPrefAttachment(
btcutil.Amount(cfg.MinChannelSize),
btcutil.Amount(cfg.MaxChannelSize),
uint16(cfg.MaxChannels), cfg.Allocation,
)
// With the heuristic itself created, we can now populate the remainder
// of the items that the autopilot agent needs to perform its duties.
self := svr.identityPriv.PubKey()
pilotCfg := autopilot.Config{
Self: self,
Heuristic: prefAttachment,
ChanController: &chanController{svr},
WalletBalance: func() (btcutil.Amount, error) {
return svr.cc.wallet.ConfirmedBalance(1)
},
Graph: autopilot.ChannelGraphFromDatabase(svr.chanDB.ChannelGraph()),
MaxPendingOpens: 10,
}
// Next, we'll fetch the current state of open channels from the
// database to use as initial state for the auto-pilot agent.
activeChannels, err := svr.chanDB.FetchAllChannels()
if err != nil {
return nil, err
}
initialChanState := make([]autopilot.Channel, len(activeChannels))
for i, channel := range activeChannels {
initialChanState[i] = autopilot.Channel{
ChanID: channel.ShortChanID(),
Capacity: channel.Capacity,
Node: autopilot.NewNodeID(channel.IdentityPub),
}
}
// Now that we have all the initial dependencies, we can create the
// auto-pilot instance itself.
pilot, err := autopilot.New(pilotCfg, initialChanState)
if err != nil {
return nil, err
}
// Finally, we'll need to subscribe to two things: incoming
// transactions that modify the wallet's balance, and also any graph
// topology updates.
txnSubscription, err := svr.cc.wallet.SubscribeTransactions()
if err != nil {
return nil, err
}
graphSubscription, err := svr.chanRouter.SubscribeTopology()
if err != nil {
return nil, err
}
// We'll launch a goroutine to provide the agent with notifications
// whenever the balance of the wallet changes.
svr.wg.Add(2)
go func() {
defer txnSubscription.Cancel()
defer svr.wg.Done()
for {
select {
case txnUpdate := <-txnSubscription.ConfirmedTransactions():
pilot.OnBalanceChange(txnUpdate.Value)
case <-svr.quit:
return
}
}
}()
go func() {
defer svr.wg.Done()
for {
select {
// We won't act upon new unconfirmed transaction, as
// we'll only use confirmed outputs when funding.
// However, we will still drain this request in order
// to avoid goroutine leaks, and ensure we promptly
// read from the channel if available.
case <-txnSubscription.UnconfirmedTransactions():
case <-svr.quit:
return
}
}
}()
// We'll also launch a goroutine to provide the agent with
// notifications for when the graph topology controlled by the node
// changes.
svr.wg.Add(1)
go func() {
defer graphSubscription.Cancel()
defer svr.wg.Done()
for {
select {
case topChange, ok := <-graphSubscription.TopologyChanges:
// If the router is shutting down, then we will
// as well.
if !ok {
return
}
for _, edgeUpdate := range topChange.ChannelEdgeUpdates {
// If this isn't an advertisement by
// the backing lnd node, then we'll
// continue as we only want to add
// channels that we've created
// ourselves.
if !edgeUpdate.AdvertisingNode.IsEqual(self) {
continue
}
// If this is indeed a channel we
// opened, then we'll convert it to the
// autopilot.Channel format, and notify
// the pilot of the new channel.
chanNode := autopilot.NewNodeID(
edgeUpdate.ConnectingNode,
)
chanID := lnwire.NewShortChanIDFromInt(
edgeUpdate.ChanID,
)
edge := autopilot.Channel{
ChanID: chanID,
Capacity: edgeUpdate.Capacity,
Node: chanNode,
}
pilot.OnChannelOpen(edge)
}
// For each closed channel, we'll obtain
// the chanID of the closed channel and send it
// to the pilot.
for _, chanClose := range topChange.ClosedChannels {
chanID := lnwire.NewShortChanIDFromInt(
chanClose.ChanID,
)
pilot.OnChannelClose(chanID)
}
case <-svr.quit:
return
}
}
}()
return pilot, nil
}