sweep: add new inteface InputSet to manage inputs in a tx

Previously the fee rate is tracked at cluster level, which may not be
accurate as each cluster is then divided into input sets. And these sets
are what's actually included in the final sweeping tx. To properly
reflect the final fee rate used in the sweeping tx, `InputSet` is added
so more customized clustering logic can be implemented in the future.
For intance, atm it's clustered by fee rates, while we may also cluster
by deadlines, urgencies, etc.

sweep/aggregator.go

@@ -33,10 +33,7 @@ type Cluster interface {
 	// CreateInputSets goes through the cluster's inputs and constructs
 	// sets of inputs that can be used to generate a sensible transaction.
 	CreateInputSets(wallet Wallet, maxFeeRate chainfee.SatPerKWeight,
-		maxInputs int) ([]inputSet, error)
-
-	// FeeRate returns the fee rate of the cluster.
-	FeeRate() chainfee.SatPerKWeight
+		maxInputs int) ([]InputSet, error)
 }
 
 // Compile-time constraint to ensure inputCluster implements Cluster.
@@ -50,11 +47,6 @@ type inputCluster struct {
 	inputs       pendingInputs
 }
 
-// FeeRate returns the fee rate of the cluster.
-func (c *inputCluster) FeeRate() chainfee.SatPerKWeight {
-	return c.sweepFeeRate
-}
-
 // GroupInputs goes through the cluster's inputs and constructs sets of inputs
 // that can be used to generate a sensible transaction. Each set contains up to
 // the configured maximum number of inputs. Negative yield inputs are skipped.
@@ -62,7 +54,7 @@ func (c *inputCluster) FeeRate() chainfee.SatPerKWeight {
 // returned.
 func (c *inputCluster) CreateInputSets(
 	wallet Wallet, maxFeeRate chainfee.SatPerKWeight,
-	maxInputs int) ([]inputSet, error) {
+	maxInputs int) ([]InputSet, error) {
 
 	// Turn the inputs into a slice so we can sort them.
 	inputList := make([]*pendingInput, 0, len(c.inputs))
@@ -110,7 +102,7 @@ func (c *inputCluster) CreateInputSets(
 	})
 
 	// Select blocks of inputs up to the configured maximum number.
-	var sets []inputSet
+	var sets []InputSet
 	for len(inputList) > 0 {
 		// Start building a set of positive-yield tx inputs under the
 		// condition that the tx will be published with the specified
@@ -156,7 +148,7 @@ func (c *inputCluster) CreateInputSets(
 			txInputs.totalOutput()-txInputs.walletInputTotal,
 			txInputs.weightEstimate(true).weight())
 
-		sets = append(sets, txInputs.inputs)
+		sets = append(sets, txInputs)
 		inputList = inputList[inputCount:]
 	}
 

sweep/sweeper.go

@@ -760,7 +760,7 @@ func (s *UtxoSweeper) sweepCluster(cluster Cluster) error {
 
 		// Create sweeping transaction for each set.
 		for _, inputs := range sets {
-			err := s.sweep(inputs, cluster.FeeRate())
+			err := s.sweep(inputs)
 			if err != nil {
 				log.Errorf("sweep new inputs: %w", err)
 			}
@@ -803,9 +803,7 @@ func (s *UtxoSweeper) signalResult(pi *pendingInput, result Result) {
 
 // sweep takes a set of preselected inputs, creates a sweep tx and publishes the
 // tx. The output address is only marked as used if the publish succeeds.
-func (s *UtxoSweeper) sweep(inputs inputSet,
-	feeRate chainfee.SatPerKWeight) error {
-
+func (s *UtxoSweeper) sweep(set InputSet) error {
 	// Generate an output script if there isn't an unused script available.
 	if s.currentOutputScript == nil {
 		pkScript, err := s.cfg.GenSweepScript()
@@ -817,8 +815,9 @@ func (s *UtxoSweeper) sweep(inputs inputSet,
 
 	// Create sweep tx.
 	tx, fee, err := createSweepTx(
-		inputs, nil, s.currentOutputScript, uint32(s.currentHeight),
-		feeRate, s.cfg.MaxFeeRate.FeePerKWeight(), s.cfg.Signer,
+		set.Inputs(), nil, s.currentOutputScript,
+		uint32(s.currentHeight), set.FeeRate(),
+		s.cfg.MaxFeeRate.FeePerKWeight(), s.cfg.Signer,
 	)
 	if err != nil {
 		return fmt.Errorf("create sweep tx: %w", err)
@@ -826,7 +825,7 @@ func (s *UtxoSweeper) sweep(inputs inputSet,
 
 	tr := &TxRecord{
 		Txid:    tx.TxHash(),
-		FeeRate: uint64(feeRate),
+		FeeRate: uint64(set.FeeRate()),
 		Fee:     uint64(fee),
 	}
 
@@ -1195,6 +1194,8 @@ func (s *UtxoSweeper) handleUpdateReq(req *updateReq) (
 // - Make handling re-orgs easier.
 // - Thwart future possible fee sniping attempts.
 // - Make us blend in with the bitcoind wallet.
+//
+// TODO(yy): remove this method and only allow sweeping via requests.
 func (s *UtxoSweeper) CreateSweepTx(inputs []input.Input,
 	feePref FeeEstimateInfo) (*wire.MsgTx, error) {
 

sweep/tx_input_set.go

@@ -30,6 +30,16 @@ const (
 	constraintsForce
 )
 
+// InputSet defines an interface that's responsible for filtering a set of
+// inputs that can be swept economically.
+type InputSet interface {
+	// Inputs returns the set of inputs that should be used to create a tx.
+	Inputs() []input.Input
+
+	// FeeRate returns the fee rate that should be used for the tx.
+	FeeRate() chainfee.SatPerKWeight
+}
+
 type txInputSetState struct {
 	// feeRate is the fee rate to use for the sweep transaction.
 	feeRate chainfee.SatPerKWeight
@@ -121,6 +131,9 @@ type txInputSet struct {
 	wallet Wallet
 }
 
+// Compile-time constraint to ensure txInputSet implements InputSet.
+var _ InputSet = (*txInputSet)(nil)
+
 // newTxInputSet constructs a new, empty input set.
 func newTxInputSet(wallet Wallet, feePerKW, maxFeeRate chainfee.SatPerKWeight,
 	maxInputs int) *txInputSet {
@@ -139,6 +152,16 @@ func newTxInputSet(wallet Wallet, feePerKW, maxFeeRate chainfee.SatPerKWeight,
 	return &b
 }
 
+// Inputs returns the inputs that should be used to create a tx.
+func (t *txInputSet) Inputs() []input.Input {
+	return t.inputs
+}
+
+// FeeRate returns the fee rate that should be used for the tx.
+func (t *txInputSet) FeeRate() chainfee.SatPerKWeight {
+	return t.feeRate
+}
+
 // enoughInput returns true if we've accumulated enough inputs to pay the fees
 // and have at least one output that meets the dust limit.
 func (t *txInputSet) enoughInput() bool {

sweep/txgenerator.go

@@ -30,22 +30,15 @@ var (
 	ErrLocktimeConflict = errors.New("incompatible locktime")
 )
 
-// txInput is an interface that provides the input data required for tx
-// generation.
-type txInput interface {
-	input.Input
-	parameters() Params
-}
-
 // createSweepTx builds a signed tx spending the inputs to the given outputs,
 // sending any leftover change to the change script.
 func createSweepTx(inputs []input.Input, outputs []*wire.TxOut,
 	changePkScript []byte, currentBlockHeight uint32,
-	feePerKw, maxFeeRate chainfee.SatPerKWeight,
+	feeRate, maxFeeRate chainfee.SatPerKWeight,
 	signer input.Signer) (*wire.MsgTx, btcutil.Amount, error) {
 
 	inputs, estimator, err := getWeightEstimate(
-		inputs, outputs, feePerKw, maxFeeRate, changePkScript,
+		inputs, outputs, feeRate, maxFeeRate, changePkScript,
 	)
 	if err != nil {
 		return nil, 0, err
@@ -216,7 +209,7 @@ func createSweepTx(inputs []input.Input, outputs []*wire.TxOut,
 		"using %v sat/kw, tx_weight=%v, tx_fee=%v, parents_count=%v, "+
 		"parents_fee=%v, parents_weight=%v",
 		sweepTx.TxHash(), len(inputs),
-		inputTypeSummary(inputs), int64(feePerKw),
+		inputTypeSummary(inputs), feeRate,
 		estimator.weight(), txFee,
 		len(estimator.parents), estimator.parentsFee,
 		estimator.parentsWeight,
@@ -229,8 +222,7 @@ func createSweepTx(inputs []input.Input, outputs []*wire.TxOut,
 // Additionally, it returns counts for the number of csv and cltv inputs.
 func getWeightEstimate(inputs []input.Input, outputs []*wire.TxOut,
 	feeRate, maxFeeRate chainfee.SatPerKWeight,
-	outputPkScript []byte) ([]input.Input,
-	*weightEstimator, error) {
+	outputPkScript []byte) ([]input.Input, *weightEstimator, error) {
 
 	// We initialize a weight estimator so we can accurately asses the
 	// amount of fees we need to pay for this sweep transaction.