sweep: handle inputs locally instead of relying on the tx

This commit changes how inputs are handled upon receiving a bump result.
Previously the inputs are taken from the `BumpResult.Tx`, which is now
instead being handled locally as we will remember the input set when
sending the bump request, and handle this input set when a result is
received.

sweep/sweeper.go

@@ -309,9 +309,9 @@ type UtxoSweeper struct {
 	// updated whenever a new block epoch is received.
 	currentHeight int32
 
-	// bumpResultChan is a channel that receives broadcast results from the
+	// bumpRespChan is a channel that receives broadcast results from the
 	// TxPublisher.
-	bumpResultChan chan *BumpResult
+	bumpRespChan chan *bumpResp
 }
 
 // UtxoSweeperConfig contains dependencies of UtxoSweeper.
@@ -395,7 +395,7 @@ func New(cfg *UtxoSweeperConfig) *UtxoSweeper {
 		pendingSweepsReqs: make(chan *pendingSweepsReq),
 		quit:              make(chan struct{}),
 		inputs:            make(InputsMap),
-		bumpResultChan:    make(chan *BumpResult, 100),
+		bumpRespChan:      make(chan *bumpResp, 100),
 	}
 }
 
@@ -681,9 +681,9 @@ func (s *UtxoSweeper) collector(blockEpochs <-chan *chainntnfs.BlockEpoch) {
 				s.sweepPendingInputs(inputs)
 			}
 
-		case result := <-s.bumpResultChan:
+		case resp := <-s.bumpRespChan:
 			// Handle the bump event.
-			err := s.handleBumpEvent(result)
+			err := s.handleBumpEvent(resp)
 			if err != nil {
 				log.Errorf("Failed to handle bump event: %v",
 					err)
@@ -840,16 +840,11 @@ func (s *UtxoSweeper) sweep(set InputSet) error {
 	// this publish result and future RBF attempt.
 	resp, err := s.cfg.Publisher.Broadcast(req)
 	if err != nil {
-		outpoints := make([]wire.OutPoint, len(set.Inputs()))
-		for i, inp := range set.Inputs() {
-			outpoints[i] = inp.OutPoint()
-		}
-
 		log.Errorf("Initial broadcast failed: %v, inputs=\n%v", err,
 			inputTypeSummary(set.Inputs()))
 
 		// TODO(yy): find out which input is causing the failure.
-		s.markInputsPublishFailed(outpoints)
+		s.markInputsPublishFailed(set)
 
 		return err
 	}
@@ -858,7 +853,7 @@ func (s *UtxoSweeper) sweep(set InputSet) error {
 	// subscribing to the result chan and listen for future updates about
 	// this tx.
 	s.wg.Add(1)
-	go s.monitorFeeBumpResult(resp)
+	go s.monitorFeeBumpResult(set, resp)
 
 	return nil
 }
@@ -868,14 +863,14 @@ func (s *UtxoSweeper) sweep(set InputSet) error {
 func (s *UtxoSweeper) markInputsPendingPublish(set InputSet) {
 	// Reschedule sweep.
 	for _, input := range set.Inputs() {
-		pi, ok := s.inputs[input.OutPoint()]
+		op := input.OutPoint()
+		pi, ok := s.inputs[op]
 		if !ok {
 			// It could be that this input is an additional wallet
 			// input that was attached. In that case there also
 			// isn't a pending input to update.
 			log.Tracef("Skipped marking input as pending "+
-				"published: %v not found in pending inputs",
+				"published: %v not found in pending inputs", op)
-				input.OutPoint())
 
 			continue
 		}
@@ -886,8 +881,7 @@ func (s *UtxoSweeper) markInputsPendingPublish(set InputSet) {
 		// publish.
 		if pi.terminated() {
 			log.Errorf("Expect input %v to not have terminated "+
-				"state, instead it has %v",
+				"state, instead it has %v", op, pi.state)
-				input.OutPoint, pi.state)
 
 			continue
 		}
@@ -902,9 +896,7 @@ func (s *UtxoSweeper) markInputsPendingPublish(set InputSet) {
 
 // markInputsPublished updates the sweeping tx in db and marks the list of
 // inputs as published.
-func (s *UtxoSweeper) markInputsPublished(tr *TxRecord,
+func (s *UtxoSweeper) markInputsPublished(tr *TxRecord, set InputSet) error {
-	inputs []*wire.TxIn) error {
-
 	// Mark this tx in db once successfully published.
 	//
 	// NOTE: this will behave as an overwrite, which is fine as the record
@@ -916,15 +908,15 @@ func (s *UtxoSweeper) markInputsPublished(tr *TxRecord,
 	}
 
 	// Reschedule sweep.
-	for _, input := range inputs {
+	for _, input := range set.Inputs() {
-		pi, ok := s.inputs[input.PreviousOutPoint]
+		op := input.OutPoint()
+		pi, ok := s.inputs[op]
 		if !ok {
 			// It could be that this input is an additional wallet
 			// input that was attached. In that case there also
 			// isn't a pending input to update.
 			log.Tracef("Skipped marking input as published: %v "+
-				"not found in pending inputs",
+				"not found in pending inputs", op)
-				input.PreviousOutPoint)
 
 			continue
 		}
@@ -933,8 +925,7 @@ func (s *UtxoSweeper) markInputsPublished(tr *TxRecord,
 		if pi.state != PendingPublish {
 			// We may get a Published if this is a replacement tx.
 			log.Debugf("Expect input %v to have %v, instead it "+
-				"has %v", input.PreviousOutPoint,
+				"has %v", op, PendingPublish, pi.state)
-				PendingPublish, pi.state)
 
 			continue
 		}
@@ -950,9 +941,10 @@ func (s *UtxoSweeper) markInputsPublished(tr *TxRecord,
 }
 
 // markInputsPublishFailed marks the list of inputs as failed to be published.
-func (s *UtxoSweeper) markInputsPublishFailed(outpoints []wire.OutPoint) {
+func (s *UtxoSweeper) markInputsPublishFailed(set InputSet) {
 	// Reschedule sweep.
-	for _, op := range outpoints {
+	for _, inp := range set.Inputs() {
+		op := inp.OutPoint()
 		pi, ok := s.inputs[op]
 		if !ok {
 			// It could be that this input is an additional wallet
@@ -1540,6 +1532,8 @@ func (s *UtxoSweeper) updateSweeperInputs() InputsMap {
 // sweepPendingInputs is called when the ticker fires. It will create clusters
 // and attempt to create and publish the sweeping transactions.
 func (s *UtxoSweeper) sweepPendingInputs(inputs InputsMap) {
+	log.Debugf("Sweeping %v inputs", len(inputs))
+
 	// Cluster all of our inputs based on the specific Aggregator.
 	sets := s.cfg.Aggregator.ClusterInputs(inputs)
 
@@ -1581,11 +1575,24 @@ func (s *UtxoSweeper) sweepPendingInputs(inputs InputsMap) {
 	}
 }
 
+// bumpResp wraps the result of a bump attempt returned from the fee bumper and
+// the inputs being used.
+type bumpResp struct {
+	// result is the result of the bump attempt returned from the fee
+	// bumper.
+	result *BumpResult
+
+	// set is the input set that was used in the bump attempt.
+	set InputSet
+}
+
 // monitorFeeBumpResult subscribes to the passed result chan to listen for
 // future updates about the sweeping tx.
 //
 // NOTE: must run as a goroutine.
-func (s *UtxoSweeper) monitorFeeBumpResult(resultChan <-chan *BumpResult) {
+func (s *UtxoSweeper) monitorFeeBumpResult(set InputSet,
+	resultChan <-chan *BumpResult) {
+
 	defer s.wg.Done()
 
 	for {
@@ -1597,9 +1604,14 @@ func (s *UtxoSweeper) monitorFeeBumpResult(resultChan <-chan *BumpResult) {
 				continue
 			}
 
+			resp := &bumpResp{
+				result: r,
+				set:    set,
+			}
+
 			// Send the result back to the main event loop.
 			select {
-			case s.bumpResultChan <- r:
+			case s.bumpRespChan <- resp:
 			case <-s.quit:
 				log.Debug("Sweeper shutting down, skip " +
 					"sending bump result")
@@ -1635,25 +1647,25 @@ func (s *UtxoSweeper) monitorFeeBumpResult(resultChan <-chan *BumpResult) {
 
 // handleBumpEventTxFailed handles the case where the tx has been failed to
 // publish.
-func (s *UtxoSweeper) handleBumpEventTxFailed(r *BumpResult) error {
+func (s *UtxoSweeper) handleBumpEventTxFailed(resp *bumpResp) {
+	r := resp.result
 	tx, err := r.Tx, r.Err
 
 	log.Errorf("Fee bump attempt failed for tx=%v: %v", tx.TxHash(), err)
 
-	outpoints := make([]wire.OutPoint, 0, len(tx.TxIn))
+	// NOTE: When marking the inputs as failed, we are using the input set
-	for _, inp := range tx.TxIn {
+	// instead of the inputs found in the tx. This is fine for current
-		outpoints = append(outpoints, inp.PreviousOutPoint)
+	// version of the sweeper because we always create a tx using ALL of
-	}
+	// the inputs specified by the set.
-
+	//
 	// TODO(yy): should we also remove the failed tx from db?
-	s.markInputsPublishFailed(outpoints)
+	s.markInputsPublishFailed(resp.set)
-
-	return err
 }
 
 // handleBumpEventTxReplaced handles the case where the sweeping tx has been
 // replaced by a new one.
-func (s *UtxoSweeper) handleBumpEventTxReplaced(r *BumpResult) error {
+func (s *UtxoSweeper) handleBumpEventTxReplaced(resp *bumpResp) error {
+	r := resp.result
 	oldTx := r.ReplacedTx
 	newTx := r.Tx
 
@@ -1693,12 +1705,13 @@ func (s *UtxoSweeper) handleBumpEventTxReplaced(r *BumpResult) error {
 	}
 
 	// Mark the inputs as published using the replacing tx.
-	return s.markInputsPublished(tr, r.Tx.TxIn)
+	return s.markInputsPublished(tr, resp.set)
 }
 
 // handleBumpEventTxPublished handles the case where the sweeping tx has been
 // successfully published.
-func (s *UtxoSweeper) handleBumpEventTxPublished(r *BumpResult) error {
+func (s *UtxoSweeper) handleBumpEventTxPublished(resp *bumpResp) error {
+	r := resp.result
 	tx := r.Tx
 	tr := &TxRecord{
 		Txid:    tx.TxHash(),
@@ -1708,7 +1721,7 @@ func (s *UtxoSweeper) handleBumpEventTxPublished(r *BumpResult) error {
 
 	// Inputs have been successfully published so we update their
 	// states.
-	err := s.markInputsPublished(tr, tx.TxIn)
+	err := s.markInputsPublished(tr, resp.set)
 	if err != nil {
 		return err
 	}
@@ -1729,10 +1742,10 @@ func (s *UtxoSweeper) handleBumpEventTxPublished(r *BumpResult) error {
 //
 // NOTE: TxConfirmed event is not handled, since we already subscribe to the
 // input's spending event, we don't need to do anything here.
-func (s *UtxoSweeper) handleBumpEvent(r *BumpResult) error {
+func (s *UtxoSweeper) handleBumpEvent(r *bumpResp) error {
-	log.Debugf("Received bump result %v", r)
+	log.Debugf("Received bump result %v", r.result)
 
-	switch r.Event {
+	switch r.result.Event {
 	// The tx has been published, we update the inputs' state and create a
 	// record to be stored in the sweeper db.
 	case TxPublished:
@@ -1740,7 +1753,8 @@ func (s *UtxoSweeper) handleBumpEvent(r *BumpResult) error {
 
 	// The tx has failed, we update the inputs' state.
 	case TxFailed:
-		return s.handleBumpEventTxFailed(r)
+		s.handleBumpEventTxFailed(r)
+		return nil
 
 	// The tx has been replaced, we will remove the old tx and replace it
 	// with the new one.

sweep/sweeper_test.go

@@ -1,6 +1,7 @@
 package sweep
 
 import (
+	"crypto/rand"
 	"errors"
 	"testing"
 	"time"
@@ -12,6 +13,7 @@ import (
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/input"
+	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
 	"github.com/stretchr/testify/mock"
@@ -33,6 +35,41 @@ var (
 	})
 )
 
+// createMockInput creates a mock input and saves it to the sweeper's inputs
+// map. The created input has the specified state and a random outpoint. It
+// will assert the method `OutPoint` is called at least once.
+func createMockInput(t *testing.T, s *UtxoSweeper,
+	state SweepState) *input.MockInput {
+
+	inp := &input.MockInput{}
+	t.Cleanup(func() {
+		inp.AssertExpectations(t)
+	})
+
+	randBuf := make([]byte, lntypes.HashSize)
+	_, err := rand.Read(randBuf)
+	require.NoError(t, err, "internal error, cannot generate random bytes")
+
+	randHash, err := chainhash.NewHash(randBuf)
+	require.NoError(t, err)
+
+	inp.On("OutPoint").Return(wire.OutPoint{
+		Hash:  *randHash,
+		Index: 0,
+	})
+
+	// We don't do branch switches based on the witness type here so we
+	// just mock it.
+	inp.On("WitnessType").Return(input.CommitmentTimeLock).Maybe()
+
+	s.inputs[inp.OutPoint()] = &SweeperInput{
+		Input: inp,
+		state: state,
+	}
+
+	return inp
+}
+
 // TestMarkInputsPendingPublish checks that given a list of inputs with
 // different states, only the non-terminal state will be marked as `Published`.
 func TestMarkInputsPendingPublish(t *testing.T) {
@@ -47,50 +84,21 @@ func TestMarkInputsPendingPublish(t *testing.T) {
 	set := &MockInputSet{}
 	defer set.AssertExpectations(t)
 
-	// Create three testing inputs.
+	// Create three inputs with different states.
-	//
+	// - inputInit specifies a newly created input.
-	// inputNotExist specifies an input that's not found in the sweeper's
+	// - inputPendingPublish specifies an input about to be published.
-	// `pendingInputs` map.
+	// - inputTerminated specifies an input that's terminated.
-	inputNotExist := &input.MockInput{}
+	var (
-	defer inputNotExist.AssertExpectations(t)
+		inputInit           = createMockInput(t, s, Init)
-
+		inputPendingPublish = createMockInput(t, s, PendingPublish)
-	inputNotExist.On("OutPoint").Return(wire.OutPoint{Index: 0})
+		inputTerminated     = createMockInput(t, s, Excluded)
-
+	)
-	// inputInit specifies a newly created input.
-	inputInit := &input.MockInput{}
-	defer inputInit.AssertExpectations(t)
-
-	inputInit.On("OutPoint").Return(wire.OutPoint{Index: 1})
-
-	s.inputs[inputInit.OutPoint()] = &SweeperInput{
-		state: Init,
-	}
-
-	// inputPendingPublish specifies an input that's about to be published.
-	inputPendingPublish := &input.MockInput{}
-	defer inputPendingPublish.AssertExpectations(t)
-
-	inputPendingPublish.On("OutPoint").Return(wire.OutPoint{Index: 2})
-
-	s.inputs[inputPendingPublish.OutPoint()] = &SweeperInput{
-		state: PendingPublish,
-	}
-
-	// inputTerminated specifies an input that's terminated.
-	inputTerminated := &input.MockInput{}
-	defer inputTerminated.AssertExpectations(t)
-
-	inputTerminated.On("OutPoint").Return(wire.OutPoint{Index: 3})
-
-	s.inputs[inputTerminated.OutPoint()] = &SweeperInput{
-		state: Excluded,
-	}
 
 	// Mark the test inputs. We expect the non-exist input and the
 	// inputTerminated to be skipped, and the rest to be marked as pending
 	// publish.
 	set.On("Inputs").Return([]input.Input{
-		inputNotExist, inputInit, inputPendingPublish, inputTerminated,
+		inputInit, inputPendingPublish, inputTerminated,
 	})
 	s.markInputsPendingPublish(set)
 
@@ -122,36 +130,22 @@ func TestMarkInputsPublished(t *testing.T) {
 	dummyTR := &TxRecord{}
 	dummyErr := errors.New("dummy error")
 
+	// Create a mock input set.
+	set := &MockInputSet{}
+	defer set.AssertExpectations(t)
+
 	// Create a test sweeper.
 	s := New(&UtxoSweeperConfig{
 		Store: mockStore,
 	})
 
-	// Create three testing inputs.
+	// Create two inputs with different states.
-	//
+	// - inputInit specifies a newly created input.
-	// inputNotExist specifies an input that's not found in the sweeper's
+	// - inputPendingPublish specifies an input about to be published.
-	// `inputs` map.
+	var (
-	inputNotExist := &wire.TxIn{
+		inputInit           = createMockInput(t, s, Init)
-		PreviousOutPoint: wire.OutPoint{Index: 1},
+		inputPendingPublish = createMockInput(t, s, PendingPublish)
-	}
+	)
-
-	// inputInit specifies a newly created input. When marking this as
-	// published, we should see an error log as this input hasn't been
-	// published yet.
-	inputInit := &wire.TxIn{
-		PreviousOutPoint: wire.OutPoint{Index: 2},
-	}
-	s.inputs[inputInit.PreviousOutPoint] = &SweeperInput{
-		state: Init,
-	}
-
-	// inputPendingPublish specifies an input that's about to be published.
-	inputPendingPublish := &wire.TxIn{
-		PreviousOutPoint: wire.OutPoint{Index: 3},
-	}
-	s.inputs[inputPendingPublish.PreviousOutPoint] = &SweeperInput{
-		state: PendingPublish,
-	}
 
 	// First, check that when an error is returned from db, it's properly
 	// returned here.
@@ -171,9 +165,9 @@ func TestMarkInputsPublished(t *testing.T) {
 	// Mark the test inputs. We expect the non-exist input and the
 	// inputInit to be skipped, and the final input to be marked as
 	// published.
-	err = s.markInputsPublished(dummyTR, []*wire.TxIn{
+	set.On("Inputs").Return([]input.Input{inputInit, inputPendingPublish})
-		inputNotExist, inputInit, inputPendingPublish,
+
-	})
+	err = s.markInputsPublished(dummyTR, set)
 	require.NoError(err)
 
 	// We expect unchanged number of pending inputs.
@@ -181,11 +175,11 @@ func TestMarkInputsPublished(t *testing.T) {
 
 	// We expect the init input's state to stay unchanged.
 	require.Equal(Init,
-		s.inputs[inputInit.PreviousOutPoint].state)
+		s.inputs[inputInit.OutPoint()].state)
 
 	// We expect the pending-publish input's is now marked as published.
 	require.Equal(Published,
-		s.inputs[inputPendingPublish.PreviousOutPoint].state)
+		s.inputs[inputPendingPublish.OutPoint()].state)
 
 	// Assert mocked statements are executed as expected.
 	mockStore.AssertExpectations(t)
@@ -202,117 +196,75 @@ func TestMarkInputsPublishFailed(t *testing.T) {
 	// Create a mock sweeper store.
 	mockStore := NewMockSweeperStore()
 
+	// Create a mock input set.
+	set := &MockInputSet{}
+	defer set.AssertExpectations(t)
+
 	// Create a test sweeper.
 	s := New(&UtxoSweeperConfig{
 		Store: mockStore,
 	})
 
-	// Create testing inputs for each state.
+	// Create inputs with different states.
-	//
+	// - inputInit specifies a newly created input. When marking this as
-	// inputNotExist specifies an input that's not found in the sweeper's
+	//   published, we should see an error log as this input hasn't been
-	// `inputs` map.
+	//   published yet.
-	inputNotExist := &wire.TxIn{
+	// - inputPendingPublish specifies an input about to be published.
-		PreviousOutPoint: wire.OutPoint{Index: 1},
+	// - inputPublished specifies an input that's published.
-	}
+	// - inputPublishFailed specifies an input that's failed to be
+	//   published.
+	// - inputSwept specifies an input that's swept.
+	// - inputExcluded specifies an input that's excluded.
+	// - inputFailed specifies an input that's failed.
+	var (
+		inputInit           = createMockInput(t, s, Init)
+		inputPendingPublish = createMockInput(t, s, PendingPublish)
+		inputPublished      = createMockInput(t, s, Published)
+		inputPublishFailed  = createMockInput(t, s, PublishFailed)
+		inputSwept          = createMockInput(t, s, Swept)
+		inputExcluded       = createMockInput(t, s, Excluded)
+		inputFailed         = createMockInput(t, s, Failed)
+	)
 
-	// inputInit specifies a newly created input. When marking this as
+	// Gather all inputs.
-	// published, we should see an error log as this input hasn't been
+	set.On("Inputs").Return([]input.Input{
-	// published yet.
+		inputInit, inputPendingPublish, inputPublished,
-	inputInit := &wire.TxIn{
+		inputPublishFailed, inputSwept, inputExcluded, inputFailed,
-		PreviousOutPoint: wire.OutPoint{Index: 2},
+	})
-	}
-	s.inputs[inputInit.PreviousOutPoint] = &SweeperInput{
-		state: Init,
-	}
-
-	// inputPendingPublish specifies an input that's about to be published.
-	inputPendingPublish := &wire.TxIn{
-		PreviousOutPoint: wire.OutPoint{Index: 3},
-	}
-	s.inputs[inputPendingPublish.PreviousOutPoint] = &SweeperInput{
-		state: PendingPublish,
-	}
-
-	// inputPublished specifies an input that's published.
-	inputPublished := &wire.TxIn{
-		PreviousOutPoint: wire.OutPoint{Index: 4},
-	}
-	s.inputs[inputPublished.PreviousOutPoint] = &SweeperInput{
-		state: Published,
-	}
-
-	// inputPublishFailed specifies an input that's failed to be published.
-	inputPublishFailed := &wire.TxIn{
-		PreviousOutPoint: wire.OutPoint{Index: 5},
-	}
-	s.inputs[inputPublishFailed.PreviousOutPoint] = &SweeperInput{
-		state: PublishFailed,
-	}
-
-	// inputSwept specifies an input that's swept.
-	inputSwept := &wire.TxIn{
-		PreviousOutPoint: wire.OutPoint{Index: 6},
-	}
-	s.inputs[inputSwept.PreviousOutPoint] = &SweeperInput{
-		state: Swept,
-	}
-
-	// inputExcluded specifies an input that's excluded.
-	inputExcluded := &wire.TxIn{
-		PreviousOutPoint: wire.OutPoint{Index: 7},
-	}
-	s.inputs[inputExcluded.PreviousOutPoint] = &SweeperInput{
-		state: Excluded,
-	}
-
-	// inputFailed specifies an input that's failed.
-	inputFailed := &wire.TxIn{
-		PreviousOutPoint: wire.OutPoint{Index: 8},
-	}
-	s.inputs[inputFailed.PreviousOutPoint] = &SweeperInput{
-		state: Failed,
-	}
-
-	// Gather all inputs' outpoints.
-	pendingOps := make([]wire.OutPoint, 0, len(s.inputs)+1)
-	for op := range s.inputs {
-		pendingOps = append(pendingOps, op)
-	}
-	pendingOps = append(pendingOps, inputNotExist.PreviousOutPoint)
 
 	// Mark the test inputs. We expect the non-exist input and the
 	// inputInit to be skipped, and the final input to be marked as
 	// published.
-	s.markInputsPublishFailed(pendingOps)
+	s.markInputsPublishFailed(set)
 
 	// We expect unchanged number of pending inputs.
 	require.Len(s.inputs, 7)
 
 	// We expect the init input's state to stay unchanged.
 	require.Equal(Init,
-		s.inputs[inputInit.PreviousOutPoint].state)
+		s.inputs[inputInit.OutPoint()].state)
 
 	// We expect the pending-publish input's is now marked as publish
 	// failed.
 	require.Equal(PublishFailed,
-		s.inputs[inputPendingPublish.PreviousOutPoint].state)
+		s.inputs[inputPendingPublish.OutPoint()].state)
 
 	// We expect the published input's is now marked as publish failed.
 	require.Equal(PublishFailed,
-		s.inputs[inputPublished.PreviousOutPoint].state)
+		s.inputs[inputPublished.OutPoint()].state)
 
 	// We expect the publish failed input to stay unchanged.
 	require.Equal(PublishFailed,
-		s.inputs[inputPublishFailed.PreviousOutPoint].state)
+		s.inputs[inputPublishFailed.OutPoint()].state)
 
 	// We expect the swept input to stay unchanged.
-	require.Equal(Swept, s.inputs[inputSwept.PreviousOutPoint].state)
+	require.Equal(Swept, s.inputs[inputSwept.OutPoint()].state)
 
 	// We expect the excluded input to stay unchanged.
-	require.Equal(Excluded, s.inputs[inputExcluded.PreviousOutPoint].state)
+	require.Equal(Excluded, s.inputs[inputExcluded.OutPoint()].state)
 
 	// We expect the failed input to stay unchanged.
-	require.Equal(Failed, s.inputs[inputFailed.PreviousOutPoint].state)
+	require.Equal(Failed, s.inputs[inputFailed.OutPoint()].state)
 
 	// Assert mocked statements are executed as expected.
 	mockStore.AssertExpectations(t)
@@ -738,33 +690,33 @@ func TestSweepPendingInputs(t *testing.T) {
 func TestHandleBumpEventTxFailed(t *testing.T) {
 	t.Parallel()
 
+	// Create a mock input set.
+	set := &MockInputSet{}
+	defer set.AssertExpectations(t)
+
 	// Create a test sweeper.
 	s := New(&UtxoSweeperConfig{})
 
-	var (
+	// inputNotExist specifies an input that's not found in the sweeper's
-		// Create four testing outpoints.
+	// `pendingInputs` map.
-		op1        = wire.OutPoint{Hash: chainhash.Hash{1}}
+	inputNotExist := &input.MockInput{}
-		op2        = wire.OutPoint{Hash: chainhash.Hash{2}}
+	defer inputNotExist.AssertExpectations(t)
-		op3        = wire.OutPoint{Hash: chainhash.Hash{3}}
+	inputNotExist.On("OutPoint").Return(wire.OutPoint{Index: 0})
-		opNotExist = wire.OutPoint{Hash: chainhash.Hash{4}}
+	opNotExist := inputNotExist.OutPoint()
-	)
 
 	// Create three mock inputs.
-	input1 := &input.MockInput{}
+	var (
-	defer input1.AssertExpectations(t)
+		input1 = createMockInput(t, s, PendingPublish)
+		input2 = createMockInput(t, s, PendingPublish)
+		input3 = createMockInput(t, s, PendingPublish)
+	)
 
-	input2 := &input.MockInput{}
+	op1 := input1.OutPoint()
-	defer input2.AssertExpectations(t)
+	op2 := input2.OutPoint()
-
+	op3 := input3.OutPoint()
-	input3 := &input.MockInput{}
-	defer input3.AssertExpectations(t)
 
 	// Construct the initial state for the sweeper.
-	s.inputs = InputsMap{
+	set.On("Inputs").Return([]input.Input{input1, input2, input3})
-		op1: &SweeperInput{Input: input1, state: PendingPublish},
-		op2: &SweeperInput{Input: input2, state: PendingPublish},
-		op3: &SweeperInput{Input: input3, state: PendingPublish},
-	}
 
 	// Create a testing tx that spends the first two inputs.
 	tx := &wire.MsgTx{
@@ -782,16 +734,26 @@ func TestHandleBumpEventTxFailed(t *testing.T) {
 		Err:   errDummy,
 	}
 
+	// Create a testing bump response.
+	resp := &bumpResp{
+		result: br,
+		set:    set,
+	}
+
 	// Call the method under test.
-	err := s.handleBumpEvent(br)
+	err := s.handleBumpEvent(resp)
 	require.ErrorIs(t, err, errDummy)
 
 	// Assert the states of the first two inputs are updated.
 	require.Equal(t, PublishFailed, s.inputs[op1].state)
 	require.Equal(t, PublishFailed, s.inputs[op2].state)
 
-	// Assert the state of the third input is not updated.
+	// Assert the state of the third input.
-	require.Equal(t, PendingPublish, s.inputs[op3].state)
+	//
+	// NOTE: Although the tx doesn't spend it, we still mark this input as
+	// failed as we are treating the input set as the single source of
+	// truth.
+	require.Equal(t, PublishFailed, s.inputs[op3].state)
 
 	// Assert the non-existing input is not added to the pending inputs.
 	require.NotContains(t, s.inputs, opNotExist)
@@ -810,23 +772,21 @@ func TestHandleBumpEventTxReplaced(t *testing.T) {
 	wallet := &MockWallet{}
 	defer wallet.AssertExpectations(t)
 
+	// Create a mock input set.
+	set := &MockInputSet{}
+	defer set.AssertExpectations(t)
+
 	// Create a test sweeper.
 	s := New(&UtxoSweeperConfig{
 		Store:  store,
 		Wallet: wallet,
 	})
 
-	// Create a testing outpoint.
-	op := wire.OutPoint{Hash: chainhash.Hash{1}}
-
 	// Create a mock input.
-	inp := &input.MockInput{}
+	inp := createMockInput(t, s, PendingPublish)
-	defer inp.AssertExpectations(t)
+	set.On("Inputs").Return([]input.Input{inp})
 
-	// Construct the initial state for the sweeper.
+	op := inp.OutPoint()
-	s.inputs = InputsMap{
-		op: &SweeperInput{Input: inp, state: PendingPublish},
-	}
 
 	// Create a testing tx that spends the input.
 	tx := &wire.MsgTx{
@@ -851,12 +811,18 @@ func TestHandleBumpEventTxReplaced(t *testing.T) {
 		Event:      TxReplaced,
 	}
 
+	// Create a testing bump response.
+	resp := &bumpResp{
+		result: br,
+		set:    set,
+	}
+
 	// Mock the store to return an error.
 	dummyErr := errors.New("dummy error")
 	store.On("GetTx", tx.TxHash()).Return(nil, dummyErr).Once()
 
 	// Call the method under test and assert the error is returned.
-	err := s.handleBumpEventTxReplaced(br)
+	err := s.handleBumpEventTxReplaced(resp)
 	require.ErrorIs(t, err, dummyErr)
 
 	// Mock the store to return the old tx record.
@@ -871,7 +837,7 @@ func TestHandleBumpEventTxReplaced(t *testing.T) {
 	store.On("DeleteTx", tx.TxHash()).Return(dummyErr).Once()
 
 	// Call the method under test and assert the error is returned.
-	err = s.handleBumpEventTxReplaced(br)
+	err = s.handleBumpEventTxReplaced(resp)
 	require.ErrorIs(t, err, dummyErr)
 
 	// Mock the store to return the old tx record and delete it without
@@ -891,7 +857,7 @@ func TestHandleBumpEventTxReplaced(t *testing.T) {
 	wallet.On("CancelRebroadcast", tx.TxHash()).Once()
 
 	// Call the method under test.
-	err = s.handleBumpEventTxReplaced(br)
+	err = s.handleBumpEventTxReplaced(resp)
 	require.NoError(t, err)
 
 	// Assert the state of the input is updated.
@@ -907,22 +873,20 @@ func TestHandleBumpEventTxPublished(t *testing.T) {
 	store := &MockSweeperStore{}
 	defer store.AssertExpectations(t)
 
+	// Create a mock input set.
+	set := &MockInputSet{}
+	defer set.AssertExpectations(t)
+
 	// Create a test sweeper.
 	s := New(&UtxoSweeperConfig{
 		Store: store,
 	})
 
-	// Create a testing outpoint.
-	op := wire.OutPoint{Hash: chainhash.Hash{1}}
-
 	// Create a mock input.
-	inp := &input.MockInput{}
+	inp := createMockInput(t, s, PendingPublish)
-	defer inp.AssertExpectations(t)
+	set.On("Inputs").Return([]input.Input{inp})
 
-	// Construct the initial state for the sweeper.
+	op := inp.OutPoint()
-	s.inputs = InputsMap{
-		op: &SweeperInput{Input: inp, state: PendingPublish},
-	}
 
 	// Create a testing tx that spends the input.
 	tx := &wire.MsgTx{
@@ -938,6 +902,12 @@ func TestHandleBumpEventTxPublished(t *testing.T) {
 		Event: TxPublished,
 	}
 
+	// Create a testing bump response.
+	resp := &bumpResp{
+		result: br,
+		set:    set,
+	}
+
 	// Mock the store to save the new tx record.
 	store.On("StoreTx", &TxRecord{
 		Txid:      tx.TxHash(),
@@ -945,7 +915,7 @@ func TestHandleBumpEventTxPublished(t *testing.T) {
 	}).Return(nil).Once()
 
 	// Call the method under test.
-	err := s.handleBumpEventTxPublished(br)
+	err := s.handleBumpEventTxPublished(resp)
 	require.NoError(t, err)
 
 	// Assert the state of the input is updated.
@@ -963,25 +933,21 @@ func TestMonitorFeeBumpResult(t *testing.T) {
 	wallet := &MockWallet{}
 	defer wallet.AssertExpectations(t)
 
+	// Create a mock input set.
+	set := &MockInputSet{}
+	defer set.AssertExpectations(t)
+
 	// Create a test sweeper.
 	s := New(&UtxoSweeperConfig{
 		Store:  store,
 		Wallet: wallet,
 	})
 
-	// Create a testing outpoint.
-	op := wire.OutPoint{Hash: chainhash.Hash{1}}
-
 	// Create a mock input.
-	inp := &input.MockInput{}
+	inp := createMockInput(t, s, PendingPublish)
-	defer inp.AssertExpectations(t)
-
-	// Construct the initial state for the sweeper.
-	s.inputs = InputsMap{
-		op: &SweeperInput{Input: inp, state: PendingPublish},
-	}
 
 	// Create a testing tx that spends the input.
+	op := inp.OutPoint()
 	tx := &wire.MsgTx{
 		LockTime: 1,
 		TxIn: []*wire.TxIn{
@@ -1060,7 +1026,8 @@ func TestMonitorFeeBumpResult(t *testing.T) {
 				return resultChan
 			},
 			shouldExit: false,
-		}, {
+		},
+		{
 			// When the sweeper is shutting down, the monitor loop
 			// should exit.
 			name: "exit on sweeper shutdown",
@@ -1087,7 +1054,7 @@ func TestMonitorFeeBumpResult(t *testing.T) {
 
 			s.wg.Add(1)
 			go func() {
-				s.monitorFeeBumpResult(resultChan)
+				s.monitorFeeBumpResult(set, resultChan)
 				close(done)
 			}()