From fa05add3b755ffbfd85e083e9d3ea1d575de116e Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 14 Aug 2020 09:33:41 +0200 Subject: [PATCH] account+rpcserver: handle consecutive batch confirmations This commit allows the account manager to handle the situation where an account participates in multiple consecutive batches that might still be in the mempool. Once the first batch confirms, the database state of the account might differ from the batch key used in that earlier batch. We need to cancel old spend and conf watches and only watch the latest batch for confirmation. While an account is in this pending batch state, no user modification like deposits or withdrawals are allowed so it's safe to just jump to the latest batch and wait for confirmations on that one. --- account/manager.go | 65 +++++++++++++++++--- account/manager_test.go | 79 ++++++++++++++++++++++++ account/watcher/watcher.go | 122 +++++++++++++++++++++++++++++++++---- rpcserver.go | 23 ++++++- 4 files changed, 267 insertions(+), 22 deletions(-) diff --git a/account/manager.go b/account/manager.go index 2c411e3..eca7897 100644 --- a/account/manager.go +++ b/account/manager.go @@ -334,6 +334,47 @@ func (m *Manager) InitAccount(ctx context.Context, value btcutil.Amount, return account, nil } +// WatchMatchedAccounts resumes accounts that were just matched in a batch and +// are expecting the batch transaction to confirm as their next account output. +// This will cancel all previous spend and conf watchers of all accounts +// involved in the batch. +func (m *Manager) WatchMatchedAccounts(ctx context.Context, + matchedAccounts []*btcec.PublicKey) error { + + for _, matchedAccount := range matchedAccounts { + acct, err := m.cfg.Store.Account(matchedAccount) + if err != nil { + return fmt.Errorf("error reading account %x: %v", + matchedAccount.SerializeCompressed(), err) + } + + // The account was just involved in a batch. That means our + // account output was spent by a batch transaction. Since we + // know that a batch transaction cannot simply be rolled back or + // replaced without us being involved, we know that the batch TX + // will eventually confirm. To handle the case where an account + // is involved in multiple consecutive batches that are all + // unconfirmed, we make sure we only track the latest state by + // canceling all previous spend and confirmation watchers. We + // then only watch the latest batch and once it confirms, create + // a new spend watcher on that. + m.watcher.CancelAccountSpend(matchedAccount) + m.watcher.CancelAccountConf(matchedAccount) + + // After taking part in a batch, the account is either pending + // closed because it was used up or pending batch update because + // it was recreated. Either way, let's resume it now by creating + // the appropriate watchers again. + err = m.resumeAccount(ctx, acct, false, false, 0) + if err != nil { + return fmt.Errorf("error resuming account %x: %v", + matchedAccount.SerializeCompressed(), err) + } + } + + return nil +} + // resumeAccount performs different operations based on the account's state. // This method serves as a way to consolidate the logic of resuming accounts on // startup and during normal operation. @@ -489,15 +530,20 @@ func (m *Manager) resumeAccount(ctx context.Context, account *Account, // nolint "%v", err) } - // In StatePendingUpdate, we've processed an account update due to - // either a matched order or trader modification, so we'll need to wait - // for its confirmation. Once it confirms, handleAccountConf will take - // care of the rest of the flow. + // In StatePendingUpdate or StatePendingBatch, we've processed an + // account update due to either a matched order or trader modification, + // so we'll need to wait for its confirmation. Once it confirms, + // handleAccountConf will take care of the rest of the flow. // // TODO(wilmer): Handle restart case where the client shuts down after // the modification has been reflected on-disk, but the auctioneer's // signature hasn't been received. - case StatePendingUpdate: + // + // TODO(guggero): Handle the case of a malicious auctioneer that + // replaces batch A with a batch A' that contains none of our accounts + // and would therefore not be noticed by us. The account would stay + // pending forever in that case. + case StatePendingUpdate, StatePendingBatch: numConfs := numConfsForValue(account.Value) log.Infof("Waiting for %v confirmation(s) of account %x", numConfs, account.TraderKey.PubKey.SerializeCompressed()) @@ -686,10 +732,13 @@ func (m *Manager) handleAccountConf(traderKey *btcec.PublicKey, // handleAccountSpend handles the different spend paths of an account. If an // account is spent by the expiration path, it'll always be marked as closed -// thereafter. If it spent by the cooperative path with the auctioneer, then the -// account will only remain open if the spending transaction recreates the +// thereafter. If it is spent by the cooperative path with the auctioneer, then +// the account will only remain open if the spending transaction recreates the // account with the expected next account script. Otherwise, it is also marked -// as closed. +// as closed. In case of multiple consecutive batches with the same account, we +// only track the spend of the latest batch, after it confirmed. So the account +// output in the spend transaction should always match our database state if +// it was a cooperative spend. func (m *Manager) handleAccountSpend(traderKey *btcec.PublicKey, spendDetails *chainntnfs.SpendDetail) error { diff --git a/account/manager_test.go b/account/manager_test.go index 130bce2..e1800be 100644 --- a/account/manager_test.go +++ b/account/manager_test.go @@ -20,6 +20,7 @@ import ( "github.com/lightningnetwork/lnd/lntest/wait" "github.com/lightningnetwork/lnd/lnwallet" "github.com/lightningnetwork/lnd/lnwallet/chainfee" + "github.com/stretchr/testify/require" ) const ( @@ -836,3 +837,81 @@ func TestAccountDeposit(t *testing.T) { // after the withdrawal. _ = h.closeAccount(account, nil, bestHeight) } + +// TestAccountConsecutiveBatches ensures that we can process an account update +// through multiple consecutive batches that only confirm after we've already +// updated our database state. +func TestAccountConsecutiveBatches(t *testing.T) { + t.Parallel() + + const bestHeight = 100 + + h := newTestHarness(t) + h.start() + defer h.stop() + + account := h.openAccount( + maxAccountValue, bestHeight+maxAccountExpiry, bestHeight, + ) + + // Then, we'll simulate the maximum number of unconfirmed batches to + // happen that'll all confirm in the same block. + const newValue = maxAccountValue / 2 + const numBatches = 10 + batchTxs := make([]*wire.MsgTx, numBatches) + for i := 0; i < numBatches; i++ { + newPkScript, err := account.NextOutputScript() + require.NoError(t, err) + + // Create an account spend which we'll notify later. This spend + // should take the multi-sig path to trigger the logic to lookup + // previous outpoints. + batchTx := &wire.MsgTx{ + Version: 2, + TxIn: []*wire.TxIn{{ + PreviousOutPoint: account.OutPoint, + Witness: wire.TxWitness{ + {0x01}, // Use multi-sig path. + {}, + {}, + }, + }}, + TxOut: []*wire.TxOut{{ + Value: int64(newValue), + PkScript: newPkScript, + }}, + } + batchTxs[i] = batchTx + + mods := []Modifier{ + ValueModifier(newValue - btcutil.Amount(i)), + StateModifier(StatePendingBatch), + OutPointModifier(wire.OutPoint{ + Hash: batchTx.TxHash(), + Index: 0, + }), + IncrementBatchKey(), + } + err = h.store.updateAccount(account, mods...) + require.NoError(t, err) + + // The RPC server will notify the manager each time a batch is + // finalized, we do the same here. + err = h.manager.WatchMatchedAccounts( + context.Background(), + []*btcec.PublicKey{account.TraderKey.PubKey}, + ) + require.NoError(t, err) + } + + // Notify the confirmation, causing the account to transition back to + // StateOpen. + confHeight := bestHeight + 6 + h.notifier.confChan <- &chainntnfs.TxConfirmation{ + Tx: batchTxs[len(batchTxs)-1], + BlockHeight: uint32(confHeight), + } + StateModifier(StateOpen)(account) + HeightHintModifier(uint32(confHeight))(account) + h.assertAccountExists(account) +} diff --git a/account/watcher/watcher.go b/account/watcher/watcher.go index 4458bdb..6a0dd7b 100644 --- a/account/watcher/watcher.go +++ b/account/watcher/watcher.go @@ -58,14 +58,22 @@ type Watcher struct { wg sync.WaitGroup quit chan struct{} ctxCancels []func() + + spendCancelMtx sync.Mutex + spendCancels map[[33]byte]func() + + confCancelMtx sync.Mutex + confCancels map[[33]byte]func() } // New instantiates a new chain watcher backed by the given config. func New(cfg *Config) *Watcher { return &Watcher{ - cfg: *cfg, - expiryReqs: make(chan *expiryReq), - quit: make(chan struct{}), + cfg: *cfg, + expiryReqs: make(chan *expiryReq), + quit: make(chan struct{}), + spendCancels: make(map[[33]byte]func()), + confCancels: make(map[[33]byte]func()), } } @@ -104,6 +112,18 @@ func (w *Watcher) Stop() { for _, cancel := range w.ctxCancels { cancel() } + + w.spendCancelMtx.Lock() + for _, cancel := range w.spendCancels { + cancel() + } + w.spendCancelMtx.Unlock() + + w.confCancelMtx.Lock() + for _, cancel := range w.confCancels { + cancel() + } + w.confCancelMtx.Unlock() }) } @@ -183,10 +203,26 @@ func (w *Watcher) expiryHandler(blockChan chan int32, errChan chan error) { } } -// WatchAccountConf watches a new account on-chain for its confirmation. +// WatchAccountConf watches a new account on-chain for its confirmation. Only +// one conf watcher per account can be used at any time. +// +// NOTE: If there is a previous conf watcher for the given account that has not +// finished yet, it will be canceled! func (w *Watcher) WatchAccountConf(traderKey *btcec.PublicKey, txHash chainhash.Hash, script []byte, numConfs, heightHint uint32) error { + w.confCancelMtx.Lock() + defer w.confCancelMtx.Unlock() + + var traderKeyRaw [33]byte + copy(traderKeyRaw[:], traderKey.SerializeCompressed()) + + // Cancel a previous conf watcher if one still exists. + cancel, ok := w.confCancels[traderKeyRaw] + if ok { + cancel() + } + ctxc, cancel := context.WithCancel(context.Background()) confChan, errChan, err := w.cfg.ChainNotifier.RegisterConfirmationsNtfn( ctxc, &txHash, script, int32(numConfs), int32(heightHint), @@ -195,10 +231,10 @@ func (w *Watcher) WatchAccountConf(traderKey *btcec.PublicKey, cancel() return err } - w.ctxCancels = append(w.ctxCancels, cancel) + w.confCancels[traderKeyRaw] = cancel w.wg.Add(1) - go w.waitForAccountConf(traderKey, confChan, errChan) + go w.waitForAccountConf(traderKey, traderKeyRaw, confChan, errChan) return nil } @@ -208,9 +244,16 @@ func (w *Watcher) WatchAccountConf(traderKey *btcec.PublicKey, // // NOTE: This method must be run as a goroutine. func (w *Watcher) waitForAccountConf(traderKey *btcec.PublicKey, - confChan chan *chainntnfs.TxConfirmation, errChan chan error) { + traderKeyRaw [33]byte, confChan chan *chainntnfs.TxConfirmation, + errChan chan error) { - defer w.wg.Done() + defer func() { + w.wg.Done() + + w.confCancelMtx.Lock() + delete(w.confCancels, traderKeyRaw) + w.confCancelMtx.Unlock() + }() select { case conf := <-confChan: @@ -231,10 +274,26 @@ func (w *Watcher) waitForAccountConf(traderKey *btcec.PublicKey, } } -// WatchAccountSpend watches for the spend of an account. +// WatchAccountSpend watches for the spend of an account. Only one spend watcher +// per account can be used at any time. +// +// NOTE: If there is a previous spend watcher for the given account that has not +// finished yet, it will be canceled! func (w *Watcher) WatchAccountSpend(traderKey *btcec.PublicKey, accountPoint wire.OutPoint, script []byte, heightHint uint32) error { + w.spendCancelMtx.Lock() + defer w.spendCancelMtx.Unlock() + + var traderKeyRaw [33]byte + copy(traderKeyRaw[:], traderKey.SerializeCompressed()) + + // Cancel a previous spend watcher if one still exists. + cancel, ok := w.spendCancels[traderKeyRaw] + if ok { + cancel() + } + ctxc, cancel := context.WithCancel(context.Background()) spendChan, errChan, err := w.cfg.ChainNotifier.RegisterSpendNtfn( ctxc, &accountPoint, script, int32(heightHint), @@ -243,10 +302,10 @@ func (w *Watcher) WatchAccountSpend(traderKey *btcec.PublicKey, cancel() return err } - w.ctxCancels = append(w.ctxCancels, cancel) + w.spendCancels[traderKeyRaw] = cancel w.wg.Add(1) - go w.waitForAccountSpend(traderKey, spendChan, errChan) + go w.waitForAccountSpend(traderKey, traderKeyRaw, spendChan, errChan) return nil } @@ -256,9 +315,16 @@ func (w *Watcher) WatchAccountSpend(traderKey *btcec.PublicKey, // // NOTE: This method must be run as a goroutine. func (w *Watcher) waitForAccountSpend(traderKey *btcec.PublicKey, - spendChan chan *chainntnfs.SpendDetail, errChan chan error) { + traderKeyRaw [33]byte, spendChan chan *chainntnfs.SpendDetail, + errChan chan error) { - defer w.wg.Done() + defer func() { + w.wg.Done() + + w.spendCancelMtx.Lock() + delete(w.spendCancels, traderKeyRaw) + w.spendCancelMtx.Unlock() + }() select { case spend := <-spendChan: @@ -294,3 +360,33 @@ func (w *Watcher) WatchAccountExpiration(traderKey *btcec.PublicKey, return errors.New("watcher shutting down") } } + +// CancelAccountSpend cancels the spend watcher of the given account, if one is +// active. +func (w *Watcher) CancelAccountSpend(traderKey *btcec.PublicKey) { + w.spendCancelMtx.Lock() + defer w.spendCancelMtx.Unlock() + + var traderKeyRaw [33]byte + copy(traderKeyRaw[:], traderKey.SerializeCompressed()) + + cancel, ok := w.spendCancels[traderKeyRaw] + if ok { + cancel() + } +} + +// CancelAccountConf cancels the conf watcher of the given account, if one is +// active. +func (w *Watcher) CancelAccountConf(traderKey *btcec.PublicKey) { + w.confCancelMtx.Lock() + defer w.confCancelMtx.Unlock() + + var traderKeyRaw [33]byte + copy(traderKeyRaw[:], traderKey.SerializeCompressed()) + + cancel, ok := w.confCancels[traderKeyRaw] + if ok { + cancel() + } +} diff --git a/rpcserver.go b/rpcserver.go index 06ef4c5..c568c34 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -820,9 +820,30 @@ func (s *rpcServer) handleServerMessage(rpcMsg *clmrpc.ServerAuctionMessage) err rpcLog.Infof("Received FinalizeMsg for batch=%x", msg.Finalize.BatchId) + // Before finalizing the batch, we want to know what accounts + // were involved so we can start watching them again. Query the + // pending batch now as BatchFinalize below will set it to nil. + batch := s.orderManager.PendingBatch() + var batchID order.BatchID copy(batchID[:], msg.Finalize.BatchId) - return s.orderManager.BatchFinalize(batchID) + err := s.orderManager.BatchFinalize(batchID) + if err != nil { + return fmt.Errorf("error finalizing batch: %v", err) + } + + // Accounts that were updated in the batch need to start new + // confirmation watchers, now that we expect a batch TX to be + // published. + matchedAccounts := make( + []*btcec.PublicKey, len(batch.AccountDiffs), + ) + for idx, acct := range batch.AccountDiffs { + matchedAccounts[idx] = acct.AccountKey + } + return s.accountManager.WatchMatchedAccounts( + context.Background(), matchedAccounts, + ) default: return fmt.Errorf("unknown server message: %v", msg)