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)