sweep: refactor IsOurTx to not return an error

Before this commit, the only error returned from `IsOurTx` is when the
root bucket was not created. In that case, we should consider the tx to
be not found in our db, since technically our db is empty.

A future PR may consider treating our wallet as the single source of
truth and query the wallet instead to check for past sweeping txns.
This commit is contained in:
yyforyongyu committed 2025-02-20 14:41:52 +08:00
1 parent 8d49246a54
commit 353f208031
5 files changed
+22 -47

No files matched your search

+2 -2
View File
@@ -24,10 +24,10 @@ func NewMockSweeperStore() *MockSweeperStore {
} }
// IsOurTx determines whether a tx is published by us, based on its hash. // IsOurTx determines whether a tx is published by us, based on its hash.
func (s *MockSweeperStore) IsOurTx(hash chainhash.Hash) (bool, error) { func (s *MockSweeperStore) IsOurTx(hash chainhash.Hash) bool {
args := s.Called(hash) args := s.Called(hash)
return args.Bool(0), args.Error(1) return args.Bool(0)
} }
// StoreTx stores a tx we are about to publish. // StoreTx stores a tx we are about to publish.
+9 -7
View File
@@ -121,7 +121,7 @@ func deserializeTxRecord(r io.Reader) (*TxRecord, error) {
type SweeperStore interface { type SweeperStore interface {
// IsOurTx determines whether a tx is published by us, based on its // IsOurTx determines whether a tx is published by us, based on its
// hash. // hash.
IsOurTx(hash chainhash.Hash) (bool, error) IsOurTx(hash chainhash.Hash) bool
// StoreTx stores a tx hash we are about to publish. // StoreTx stores a tx hash we are about to publish.
StoreTx(*TxRecord) error StoreTx(*TxRecord) error
@@ -276,15 +276,17 @@ func (s *sweeperStore) StoreTx(tr *TxRecord) error {
}, func() {}) }, func() {})
} }
// IsOurTx determines whether a tx is published by us, based on its // IsOurTx determines whether a tx is published by us, based on its hash.
// hash. func (s *sweeperStore) IsOurTx(hash chainhash.Hash) bool {
func (s *sweeperStore) IsOurTx(hash chainhash.Hash) (bool, error) {
var ours bool var ours bool
err := kvdb.View(s.db, func(tx kvdb.RTx) error { err := kvdb.View(s.db, func(tx kvdb.RTx) error {
txHashesBucket := tx.ReadBucket(txHashesBucketKey) txHashesBucket := tx.ReadBucket(txHashesBucketKey)
// If the root bucket cannot be found, we consider the tx to be
// not found in our db.
if txHashesBucket == nil { if txHashesBucket == nil {
return errNoTxHashesBucket log.Error("Tx hashes bucket not found in sweeper store")
return nil
} }
ours = txHashesBucket.Get(hash[:]) != nil ours = txHashesBucket.Get(hash[:]) != nil
@@ -294,10 +296,10 @@ func (s *sweeperStore) IsOurTx(hash chainhash.Hash) (bool, error) {
ours = false ours = false
}) })
if err != nil { if err != nil {
return false, err return false
} }
return ours, nil return ours
} }
// ListSweeps lists all the sweep transactions we have in the sweeper store. // ListSweeps lists all the sweep transactions we have in the sweeper store.
+3 -6
View File
@@ -57,18 +57,15 @@ func TestStore(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
// Assert that both txes are recognized as our own. // Assert that both txes are recognized as our own.
ours, err := store.IsOurTx(tx1.TxHash()) ours := store.IsOurTx(tx1.TxHash())
require.NoError(t, err)
require.True(t, ours, "expected tx to be ours") require.True(t, ours, "expected tx to be ours")
ours, err = store.IsOurTx(tx2.TxHash()) ours = store.IsOurTx(tx2.TxHash())
require.NoError(t, err)
require.True(t, ours, "expected tx to be ours") require.True(t, ours, "expected tx to be ours")
// An different hash should be reported as not being ours. // An different hash should be reported as not being ours.
var unknownHash chainhash.Hash var unknownHash chainhash.Hash
ours, err = store.IsOurTx(unknownHash) ours = store.IsOurTx(unknownHash)
require.NoError(t, err)
require.False(t, ours, "expected tx to not be ours") require.False(t, ours, "expected tx to not be ours")
txns, err := store.ListSweeps() txns, err := store.ListSweeps()
+4 -28
View File
@@ -1394,12 +1394,7 @@ func (s *UtxoSweeper) handleExistingInput(input *sweepInputMessage,
func (s *UtxoSweeper) handleInputSpent(spend *chainntnfs.SpendDetail) { func (s *UtxoSweeper) handleInputSpent(spend *chainntnfs.SpendDetail) {
// Query store to find out if we ever published this tx. // Query store to find out if we ever published this tx.
spendHash := *spend.SpenderTxHash spendHash := *spend.SpenderTxHash
isOurTx, err := s.cfg.Store.IsOurTx(spendHash) isOurTx := s.cfg.Store.IsOurTx(spendHash)
if err != nil {
log.Errorf("cannot determine if tx %v is ours: %v",
spendHash, err)
return
}
// If this isn't our transaction, it means someone else swept outputs // If this isn't our transaction, it means someone else swept outputs
// that we were attempting to sweep. This can happen for anchor outputs // that we were attempting to sweep. This can happen for anchor outputs
@@ -1879,22 +1874,7 @@ func (s *UtxoSweeper) handleBumpEvent(r *bumpResp) error {
// NOTE: It is enough to check the txid because the sweeper will create // NOTE: It is enough to check the txid because the sweeper will create
// outpoints which solely belong to the internal LND wallet. // outpoints which solely belong to the internal LND wallet.
func (s *UtxoSweeper) IsSweeperOutpoint(op wire.OutPoint) bool { func (s *UtxoSweeper) IsSweeperOutpoint(op wire.OutPoint) bool {
found, err := s.cfg.Store.IsOurTx(op.Hash) return s.cfg.Store.IsOurTx(op.Hash)
// In case there is an error fetching the transaction details from the
// sweeper store we assume the outpoint is still used by the sweeper
// (worst case scenario).
//
// TODO(ziggie): Ensure that confirmed outpoints are deleted from the
// bucket.
if err != nil && !errors.Is(err, errNoTxHashesBucket) {
log.Errorf("failed to fetch info for outpoint(%v:%d) "+
"with: %v, we assume it is still in use by the sweeper",
op.Hash, op.Index, err)
return true
}
return found
} }
// markInputSwept marks the given input as swept by the tx. It will also notify // markInputSwept marks the given input as swept by the tx. It will also notify
@@ -1923,11 +1903,7 @@ func (s *UtxoSweeper) handleUnknownSpendTx(inp *SweeperInput, tx *wire.MsgTx) {
op := inp.OutPoint() op := inp.OutPoint()
txid := tx.TxHash() txid := tx.TxHash()
isOurTx, err := s.cfg.Store.IsOurTx(txid) isOurTx := s.cfg.Store.IsOurTx(txid)
if err != nil {
log.Errorf("Cannot determine if tx %v is ours: %v", txid, err)
return
}
// If this is our tx, it means it's a previous sweeping tx that got // If this is our tx, it means it's a previous sweeping tx that got
// confirmed, which could happen when a restart happens during the // confirmed, which could happen when a restart happens during the
@@ -1955,7 +1931,7 @@ func (s *UtxoSweeper) handleUnknownSpendTx(inp *SweeperInput, tx *wire.MsgTx) {
spentInputs[txIn.PreviousOutPoint] = struct{}{} spentInputs[txIn.PreviousOutPoint] = struct{}{}
} }
err = s.removeConflictSweepDescendants(spentInputs) err := s.removeConflictSweepDescendants(spentInputs)
if err != nil { if err != nil {
log.Warnf("unable to remove descendant transactions "+ log.Warnf("unable to remove descendant transactions "+
"due to tx %v: ", txid) "due to tx %v: ", txid)
+4 -4
View File
@@ -1227,7 +1227,7 @@ func TestHandleUnknownSpendTxOurs(t *testing.T) {
txid := tx.TxHash() txid := tx.TxHash()
// Mock the store to return true when calling IsOurTx. // Mock the store to return true when calling IsOurTx.
store.On("IsOurTx", txid).Return(true, nil).Once() store.On("IsOurTx", txid).Return(true).Once()
// Call the method under test. // Call the method under test.
s.handleUnknownSpendTx(si, tx) s.handleUnknownSpendTx(si, tx)
@@ -1271,7 +1271,7 @@ func TestHandleInputSpendTxThirdParty(t *testing.T) {
txid := tx.TxHash() txid := tx.TxHash()
// Mock the store to return false when calling IsOurTx. // Mock the store to return false when calling IsOurTx.
store.On("IsOurTx", txid).Return(false, nil).Once() store.On("IsOurTx", txid).Return(false).Once()
// Mock `ListSweeps` to return an empty slice as we are testing the // Mock `ListSweeps` to return an empty slice as we are testing the
// workflow here, not the method `removeConflictSweepDescendants`. // workflow here, not the method `removeConflictSweepDescendants`.
@@ -1333,7 +1333,7 @@ func TestHandleBumpEventTxUnknownSpendNoRetry(t *testing.T) {
} }
// Mock the store to return true when calling IsOurTx. // Mock the store to return true when calling IsOurTx.
store.On("IsOurTx", txid).Return(true, nil).Once() store.On("IsOurTx", txid).Return(true).Once()
// Call the method under test. // Call the method under test.
s.handleBumpEventTxUnknownSpend(resp) s.handleBumpEventTxUnknownSpend(resp)
@@ -1419,7 +1419,7 @@ func TestHandleBumpEventTxUnknownSpendWithRetry(t *testing.T) {
} }
// Mock the store to return true when calling IsOurTx. // Mock the store to return true when calling IsOurTx.
store.On("IsOurTx", txid).Return(true, nil).Once() store.On("IsOurTx", txid).Return(true).Once()
// Mock the aggregator to return an empty slice as we are not testing // Mock the aggregator to return an empty slice as we are not testing
// the actual sweeping behavior. // the actual sweeping behavior.