From e1a376d9f83d04eea0153a0f8565cec5142e6ccc Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Tue, 28 Aug 2018 19:17:14 -0700 Subject: [PATCH 1/3] autopilot: use updateBalance rather than tracking balance explicitly In this commit, we modify the balanceUpdate autopilot signal to update the balance according to what's returned to the WalletBalance callback rather than explicitly tracking the balance. This gives the agent a better sense of what the wallet's balance actually is. --- autopilot/agent.go | 25 ++++++++++--------------- autopilot/agent_test.go | 29 +++++++++++++++++++++-------- pilot.go | 4 ++-- 3 files changed, 33 insertions(+), 25 deletions(-) diff --git a/autopilot/agent.go b/autopilot/agent.go index d7c14e23c..6f137583a 100644 --- a/autopilot/agent.go +++ b/autopilot/agent.go @@ -154,13 +154,8 @@ func (a *Agent) Start() error { log.Infof("Autopilot Agent starting") - startingBalance, err := a.cfg.WalletBalance() - if err != nil { - return err - } - a.wg.Add(1) - go a.controller(startingBalance) + go a.controller() return nil } @@ -183,7 +178,6 @@ func (a *Agent) Stop() error { // balanceUpdate is a type of external state update that reflects an // increase/decrease in the funds currently available to the wallet. type balanceUpdate struct { - balanceDelta btcutil.Amount } // nodeUpdates is a type of external state update that reflects an addition or @@ -214,13 +208,13 @@ type chanCloseUpdate struct { // OnBalanceChange is a callback that should be executed each time the balance of // the backing wallet changes. -func (a *Agent) OnBalanceChange(delta btcutil.Amount) { +func (a *Agent) OnBalanceChange() { a.wg.Add(1) go func() { defer a.wg.Done() select { - case a.stateUpdates <- &balanceUpdate{balanceDelta: delta}: + case a.stateUpdates <- &balanceUpdate{}: case <-a.quit: } }() @@ -342,12 +336,12 @@ func mergeChanState(pendingChans map[NodeID]Channel, // and external state changes as a result of decisions it makes w.r.t channel // allocation, or attributes affecting its control loop being updated by the // backing Lightning Node. -func (a *Agent) controller(startingBalance btcutil.Amount) { +func (a *Agent) controller() { defer a.wg.Done() // We'll start off by assigning our starting balance, and injecting // that amount as an initial wake up to the main controller goroutine. - a.OnBalanceChange(startingBalance) + a.OnBalanceChange() // TODO(roasbeef): do we in fact need to maintain order? // * use sync.Cond if so @@ -388,15 +382,16 @@ func (a *Agent) controller(startingBalance btcutil.Amount) { // up an additional channel, or splice in funds to an // existing one. case *balanceUpdate: - log.Debugf("Applying external balance state "+ - "update of: %v", update.balanceDelta) + log.Debug("Applying external balance state " + + "update") - a.totalBalance += update.balanceDelta + updateBalance() // The channel we tried to open previously failed for // whatever reason. case *chanOpenFailureUpdate: - log.Debug("Retrying after previous channel open failure.") + log.Debug("Retrying after previous channel " + + "open failure.") updateBalance() diff --git a/autopilot/agent_test.go b/autopilot/agent_test.go index ff47c3219..113aad42e 100644 --- a/autopilot/agent_test.go +++ b/autopilot/agent_test.go @@ -527,7 +527,8 @@ func TestAgentBalanceUpdate(t *testing.T) { memGraph, _, _ := newMemChanGraph() // The wallet will start with 2 BTC available. - const walletBalance = btcutil.SatoshiPerBitcoin * 2 + var walletBalanceMtx sync.Mutex + walletBalance := btcutil.Amount(btcutil.SatoshiPerBitcoin * 2) // With the dependencies we created, we can now create the initial // agent itself. @@ -536,6 +537,8 @@ func TestAgentBalanceUpdate(t *testing.T) { Heuristic: heuristic, ChanController: chanController, WalletBalance: func() (btcutil.Amount, error) { + walletBalanceMtx.Lock() + defer walletBalanceMtx.Unlock() return walletBalance, nil }, ConnectToPeer: func(*btcec.PublicKey, []net.Addr) (bool, error) { @@ -583,8 +586,11 @@ func TestAgentBalanceUpdate(t *testing.T) { // Next we'll send a new balance update signal to the agent, adding 5 // BTC to the amount of available funds. - const balanceDelta = btcutil.SatoshiPerBitcoin * 5 - agent.OnBalanceChange(balanceDelta) + walletBalanceMtx.Lock() + walletBalance += btcutil.SatoshiPerBitcoin * 5 + walletBalanceMtx.Unlock() + + agent.OnBalanceChange() wg = sync.WaitGroup{} @@ -597,11 +603,10 @@ func TestAgentBalanceUpdate(t *testing.T) { // At this point, the local state of the agent should // have also been updated to reflect that the LN node // now has an additional 5BTC available. - const expectedAmt = walletBalance + balanceDelta - if agent.totalBalance != expectedAmt { + if agent.totalBalance != walletBalance { t.Fatalf("expected %v wallet balance "+ "instead have %v", agent.totalBalance, - expectedAmt) + walletBalance) } // With all of our assertions passed, we'll signal the @@ -932,7 +937,8 @@ func TestAgentPendingChannelState(t *testing.T) { memGraph, _, _ := newMemChanGraph() // The wallet will start with 6 BTC available. - const walletBalance = btcutil.SatoshiPerBitcoin * 6 + var walletBalanceMtx sync.Mutex + walletBalance := btcutil.Amount(btcutil.SatoshiPerBitcoin * 6) // With the dependencies we created, we can now create the initial // agent itself. @@ -941,6 +947,9 @@ func TestAgentPendingChannelState(t *testing.T) { Heuristic: heuristic, ChanController: chanController, WalletBalance: func() (btcutil.Amount, error) { + walletBalanceMtx.Lock() + defer walletBalanceMtx.Unlock() + return walletBalance, nil }, ConnectToPeer: func(*btcec.PublicKey, []net.Addr) (bool, error) { @@ -1039,7 +1048,11 @@ func TestAgentPendingChannelState(t *testing.T) { // Now, in order to test that the pending state was properly updated, // we'll trigger a balance update in order to trigger a query to the // heuristic. - agent.OnBalanceChange(0.4 * btcutil.SatoshiPerBitcoin) + walletBalanceMtx.Lock() + walletBalance += 0.4 * btcutil.SatoshiPerBitcoin + walletBalanceMtx.Unlock() + + agent.OnBalanceChange() // The heuristic should be queried, and the argument for the set of // channels passed in should include the pending channels that diff --git a/pilot.go b/pilot.go index 219702a7f..e0f444344 100644 --- a/pilot.go +++ b/pilot.go @@ -224,8 +224,8 @@ func initAutoPilot(svr *server, cfg *autoPilotConfig) (*autopilot.Agent, error) for { select { - case txnUpdate := <-txnSubscription.ConfirmedTransactions(): - pilot.OnBalanceChange(txnUpdate.Value) + case <-txnSubscription.ConfirmedTransactions(): + pilot.OnBalanceChange() case <-svr.quit: return } From e1a4657427d9ab09ad99f02a1c500f59143b526e Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Wed, 29 Aug 2018 02:03:02 -0700 Subject: [PATCH 2/3] pilot: remove disconnection logic from chanController In this commit, we remove the disconnection logic within the chanController when failing to open a channel with a peer. We do this as it's already done within the autopilot agent, where it should be, and because it's possible that we were already connected to this node and we happened to disconnect them anyway. --- pilot.go | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/pilot.go b/pilot.go index e0f444344..f218787b8 100644 --- a/pilot.go +++ b/pilot.go @@ -55,18 +55,6 @@ func (c *chanController) OpenChannel(target *btcec.PublicKey, updateStream, errChan := c.server.OpenChannel(req) 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 From 0a2986355e21c28c0cccf2c4c618eef7f0e05367 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Wed, 29 Aug 2018 02:04:54 -0700 Subject: [PATCH 3/3] pilot: improve error when unable to reach any of a peer's addresses --- pilot.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pilot.go b/pilot.go index f218787b8..c745082c2 100644 --- a/pilot.go +++ b/pilot.go @@ -160,8 +160,8 @@ func initAutoPilot(svr *server, cfg *autoPilotConfig) (*autopilot.Agent, error) // If we weren't able to establish a connection at all, // then we'll error out. if !connected { - return false, fmt.Errorf("unable to connect "+ - "to %x", target.SerializeCompressed()) + return false, errors.New("exhausted all " + + "advertised addresses") } return false, nil