sweep: add method handleBumpEventError and fix markInputFailed

Previously in `markInputFailed`, we'd remove all inputs under the same
group via `removeExclusiveGroup`. This is wrong as when the current
sweep fails for this input, it shouldn't affect other inputs.

Author: yyforyongyu

sweep/sweeper.go

@@ -1433,11 +1433,6 @@ func (s *UtxoSweeper) markInputFailed(pi *SweeperInput, err error) {
 
 	pi.state = Failed
 
-	// Remove all other inputs in this exclusive group.
-	if pi.params.ExclusiveGroup != nil {
-		s.removeExclusiveGroup(*pi.params.ExclusiveGroup)
-	}
-
 	s.signalResult(pi, Result{Err: err})
 }
 
@@ -1701,6 +1696,60 @@ func (s *UtxoSweeper) handleBumpEventTxPublished(r *BumpResult) error {
 	return nil
 }
 
+// handleBumpEventError handles the case where there's an unexpected error when
+// creating or publishing the sweeping tx. In this case, the tx will be removed
+// from the sweeper store and the inputs will be marked as `Failed`.
+func (s *UtxoSweeper) handleBumpEventError(r *BumpResult) error {
+	txid := r.Tx.TxHash()
+	log.Infof("Tx=%v failed with unexpected error: %v", txid, r.Err)
+
+	// Remove the tx from the sweeper db if it exists.
+	if err := s.cfg.Store.DeleteTx(txid); err != nil {
+		return fmt.Errorf("delete tx record for %v: %w", txid, err)
+	}
+
+	// Mark the inputs as failed.
+	s.markInputsFailed(r.Tx, r.Err)
+
+	return nil
+}
+
+// markInputsFailed marks all inputs found in the tx as failed. It will also
+// notify all the subscribers of these inputs.
+func (s *UtxoSweeper) markInputsFailed(tx *wire.MsgTx, err error) {
+	for _, txIn := range tx.TxIn {
+		outpoint := txIn.PreviousOutPoint
+
+		input, ok := s.inputs[outpoint]
+		if !ok {
+			// It's very likely that a spending tx contains inputs
+			// that we don't know.
+			log.Tracef("Skipped marking input as failed: %v not "+
+				"found in pending inputs", outpoint)
+
+			continue
+		}
+
+		// If the input is already in a terminal state, we don't want
+		// to rewrite it, which also indicates an error as we only get
+		// an error event during the initial broadcast.
+		if input.terminated() {
+			log.Errorf("Skipped marking input=%v as failed due to "+
+				"unexpected state=%v", outpoint, input.state)
+
+			continue
+		}
+
+		input.state = Failed
+
+		// Signal result channels.
+		s.signalResult(input, Result{
+			Tx:  tx,
+			Err: err,
+		})
+	}
+}
+
 // handleBumpEvent handles the result sent from the bumper based on its event
 // type.
 //
@@ -1724,8 +1773,11 @@ func (s *UtxoSweeper) handleBumpEvent(r *BumpResult) error {
 	case TxReplaced:
 		return s.handleBumpEventTxReplaced(r)
 
+	// There's an unexpected error in creating or publishing the tx, we
+	// will remove the tx from the sweeper db and mark the inputs as
+	// failed.
 	case TxError:
-		// TODO(yy): create a method to remove this input.
+		return s.handleBumpEventError(r)
 	}
 
 	return nil

sweep/sweeper_test.go

@@ -1104,3 +1104,179 @@ func TestMonitorFeeBumpResult(t *testing.T) {
 		})
 	}
 }
+
+// TestMarkInputsFailed checks that given a list of inputs with different
+// states, the method `markInputsFailed` correctly marks the inputs as failed.
+func TestMarkInputsFailed(t *testing.T) {
+	t.Parallel()
+
+	require := require.New(t)
+
+	// Create a test sweeper.
+	s := New(&UtxoSweeperConfig{})
+
+	// Create a mock input.
+	mockInput := &input.MockInput{}
+	defer mockInput.AssertExpectations(t)
+
+	// Mock the `OutPoint` to return a dummy outpoint.
+	mockInput.On("OutPoint").Return(wire.OutPoint{Hash: chainhash.Hash{1}})
+
+	// Create testing inputs for each state.
+	//
+	// inputNotExist specifies an input that's not found in the sweeper's
+	// `inputs` map.
+	inputNotExist := &wire.TxIn{
+		PreviousOutPoint: wire.OutPoint{Index: 1},
+	}
+
+	// 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,
+		Input: mockInput,
+	}
+
+	// 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,
+		Input: mockInput,
+	}
+
+	// inputPublished specifies an input that's published.
+	inputPublished := &wire.TxIn{
+		PreviousOutPoint: wire.OutPoint{Index: 4},
+	}
+	s.inputs[inputPublished.PreviousOutPoint] = &SweeperInput{
+		state: Published,
+		Input: mockInput,
+	}
+
+	// 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,
+		Input: mockInput,
+	}
+
+	// inputSwept specifies an input that's swept.
+	inputSwept := &wire.TxIn{
+		PreviousOutPoint: wire.OutPoint{Index: 6},
+	}
+	s.inputs[inputSwept.PreviousOutPoint] = &SweeperInput{
+		state: Swept,
+		Input: mockInput,
+	}
+
+	// inputExcluded specifies an input that's excluded.
+	inputExcluded := &wire.TxIn{
+		PreviousOutPoint: wire.OutPoint{Index: 7},
+	}
+	s.inputs[inputExcluded.PreviousOutPoint] = &SweeperInput{
+		state: Excluded,
+		Input: mockInput,
+	}
+
+	// inputFailed specifies an input that's failed.
+	inputFailed := &wire.TxIn{
+		PreviousOutPoint: wire.OutPoint{Index: 8},
+	}
+	s.inputs[inputFailed.PreviousOutPoint] = &SweeperInput{
+		state: Failed,
+		Input: mockInput,
+	}
+
+	// Create a test tx.
+	tx := &wire.MsgTx{
+		TxIn: []*wire.TxIn{
+			inputNotExist,
+			inputInit,
+			inputPendingPublish,
+			inputPublished,
+			inputPublishFailed,
+			inputSwept,
+			inputExcluded,
+			inputFailed,
+		},
+	}
+
+	// Mark the test inputs. We expect the non-exist input and
+	// inputSwept/inputExcluded/inputFailed to be skipped.
+	s.markInputsFailed(tx, errDummy)
+
+	// We expect unchanged number of pending inputs.
+	require.Len(s.inputs, 7)
+
+	// We expect the init input's to be marked as failed.
+	require.Equal(Failed, s.inputs[inputInit.PreviousOutPoint].state)
+
+	// We expect the pending-publish input to be marked as failed.
+	require.Equal(Failed,
+		s.inputs[inputPendingPublish.PreviousOutPoint].state)
+
+	// We expect the published input to be marked as failed.
+	require.Equal(Failed, s.inputs[inputPublished.PreviousOutPoint].state)
+
+	// We expect the publish failed input to be markd as failed.
+	require.Equal(Failed,
+		s.inputs[inputPublishFailed.PreviousOutPoint].state)
+
+	// We expect the swept input to stay unchanged.
+	require.Equal(Swept, s.inputs[inputSwept.PreviousOutPoint].state)
+
+	// We expect the excluded input to stay unchanged.
+	require.Equal(Excluded, s.inputs[inputExcluded.PreviousOutPoint].state)
+
+	// We expect the failed input to stay unchanged.
+	require.Equal(Failed, s.inputs[inputFailed.PreviousOutPoint].state)
+}
+
+// TestHandleBumpEventError checks that `handleBumpEventError` correctly
+// handles a `TxError` event.
+func TestHandleBumpEventError(t *testing.T) {
+	t.Parallel()
+
+	rt := require.New(t)
+
+	// Create a mock store.
+	store := &MockSweeperStore{}
+	defer store.AssertExpectations(t)
+
+	// Create a test sweeper.
+	s := New(&UtxoSweeperConfig{
+		Store: store,
+	})
+
+	// Create a testing tx.
+	//
+	// We are not testing `markInputFailed` here, so the actual tx doesn't
+	// matter.
+	tx := &wire.MsgTx{}
+	result := &BumpResult{
+		Tx:  tx,
+		Err: errDummy,
+	}
+
+	// Mock the store to return an error.
+	store.On("DeleteTx", mock.Anything).Return(errDummy).Once()
+
+	// Call the method under test and assert the error is returned.
+	err := s.handleBumpEventError(result)
+	rt.ErrorIs(err, errDummy)
+
+	// Mock the store to return nil.
+	store.On("DeleteTx", mock.Anything).Return(nil).Once()
+
+	// Call the method under test and assert no error is returned.
+	err = s.handleBumpEventError(result)
+	rt.NoError(err)
+}