diff --git a/sweep/aggregator.go b/sweep/aggregator.go index fdc4a27f1..a65fa39c2 100644 --- a/sweep/aggregator.go +++ b/sweep/aggregator.go @@ -122,7 +122,7 @@ func (c *inputCluster) createInputSets(maxFeeRate chainfee.SatPerKWeight, type UtxoAggregator interface { // ClusterInputs takes a list of inputs and groups them into input // sets. Each input set will be used to create a sweeping transaction. - ClusterInputs(inputs InputsMap, defaultDeadline int32) []InputSet + ClusterInputs(inputs InputsMap) []InputSet } // SimpleAggregator aggregates inputs known by the Sweeper based on each @@ -174,7 +174,7 @@ func NewSimpleUtxoAggregator(estimator chainfee.Estimator, // inputs known by the UtxoSweeper. It clusters inputs by // 1) Required tx locktime // 2) Similar fee rates. -func (s *SimpleAggregator) ClusterInputs(inputs InputsMap, _ int32) []InputSet { +func (s *SimpleAggregator) ClusterInputs(inputs InputsMap) []InputSet { // We start by getting the inputs clusters by locktime. Since the // inputs commit to the locktime, they can only be clustered together // if the locktime is equal. @@ -501,9 +501,7 @@ type clusterGroup map[int32][]SweeperInput // 5. optionally split a cluster if it exceeds the max input limit. // 6. create input sets from each of the clusters. // 7. create input sets for each of the exclusive inputs. -func (b *BudgetAggregator) ClusterInputs(inputs InputsMap, - defaultDeadline int32) []InputSet { - +func (b *BudgetAggregator) ClusterInputs(inputs InputsMap) []InputSet { // Filter out inputs that have a budget below min relay fee. filteredInputs := b.filterInputs(inputs) @@ -521,7 +519,7 @@ func (b *BudgetAggregator) ClusterInputs(inputs InputsMap, for _, input := range filteredInputs { // Get deadline height, and use the specified default deadline // height if it's not set. - height := input.params.DeadlineHeight.UnwrapOr(defaultDeadline) + height := input.DeadlineHeight // Put exclusive inputs in their own set. if input.params.ExclusiveGroup != nil { diff --git a/sweep/aggregator_test.go b/sweep/aggregator_test.go index e002a14b6..b9b86379b 100644 --- a/sweep/aggregator_test.go +++ b/sweep/aggregator_test.go @@ -821,9 +821,9 @@ func TestBudgetInputSetClusterInputs(t *testing.T) { // Create three deadline heights, which means there are three // groups of inputs to be expected. - deadlineNone = fn.None[int32]() - deadline1 = fn.Some(int32(1)) - deadline2 = fn.Some(int32(2)) + defaultDeadline = testHeight + DefaultDeadlineDelta + deadline1 = int32(1) + deadline2 = int32(2) ) // Create testing pending inputs. @@ -854,16 +854,16 @@ func TestBudgetInputSetClusterInputs(t *testing.T) { Input: inpExclusive, params: Params{ Budget: budgetHigh, - DeadlineHeight: deadline1, ExclusiveGroup: &exclusiveGroup, }, + DeadlineHeight: deadline1, } // For each deadline height, create two inputs with different budgets, // one below the min fee rate and one above it. We should see the lower // one being filtered out. - for i, deadline := range []fn.Option[int32]{ - deadlineNone, deadline1, deadline2, + for i, deadline := range []int32{ + defaultDeadline, deadline1, deadline2, } { // Define three outpoints. opLow := wire.OutPoint{ @@ -917,25 +917,25 @@ func TestBudgetInputSetClusterInputs(t *testing.T) { inputs[opLow] = &SweeperInput{ Input: inpLow, params: Params{ - Budget: budgetLow, - DeadlineHeight: deadline, + Budget: budgetLow, }, + DeadlineHeight: deadline, } // Add the high inputs, which should be included. inputs[opHigh1] = &SweeperInput{ Input: inpHigh1, params: Params{ - Budget: budgetHigh, - DeadlineHeight: deadline, + Budget: budgetHigh, }, + DeadlineHeight: deadline, } inputs[opHigh2] = &SweeperInput{ Input: inpHigh2, params: Params{ - Budget: budgetHigh, - DeadlineHeight: deadline, + Budget: budgetHigh, }, + DeadlineHeight: deadline, } } @@ -943,8 +943,7 @@ func TestBudgetInputSetClusterInputs(t *testing.T) { b := NewBudgetAggregator(estimator, DefaultMaxInputsPerTx) // Call the method under test. - defaultDeadline := testHeight + DefaultDeadlineDelta - result := b.ClusterInputs(inputs, defaultDeadline) + result := b.ClusterInputs(inputs) // We expect four input sets to be returned, one for each deadline and // extra one for the exclusive input. @@ -969,8 +968,8 @@ func TestBudgetInputSetClusterInputs(t *testing.T) { // We expect to see all three deadlines. require.Contains(t, deadlines, defaultDeadline) - require.Contains(t, deadlines, deadline1.UnwrapOrFail(t)) - require.Contains(t, deadlines, deadline2.UnwrapOrFail(t)) + require.Contains(t, deadlines, deadline1) + require.Contains(t, deadlines, deadline2) } // TestSplitOnLocktime asserts `splitOnLocktime` works as expected. diff --git a/sweep/mock_test.go b/sweep/mock_test.go index 6160923c3..356d2e3a8 100644 --- a/sweep/mock_test.go +++ b/sweep/mock_test.go @@ -346,10 +346,8 @@ type mockUtxoAggregator struct { var _ UtxoAggregator = (*mockUtxoAggregator)(nil) // ClusterInputs takes a list of inputs and groups them into clusters. -func (m *mockUtxoAggregator) ClusterInputs(inputs InputsMap, - defaultDeadline int32) []InputSet { - - args := m.Called(inputs, defaultDeadline) +func (m *mockUtxoAggregator) ClusterInputs(inputs InputsMap) []InputSet { + args := m.Called(inputs) return args.Get(0).([]InputSet) } diff --git a/sweep/sweeper.go b/sweep/sweeper.go index 82fdfe4b6..4abf59d1a 100644 --- a/sweep/sweeper.go +++ b/sweep/sweeper.go @@ -200,10 +200,10 @@ type SweeperInput struct { // rbf records the RBF constraints. rbf fn.Option[RBFInfo] - // deadlineHeight is the deadline height for this input. This is + // DeadlineHeight is the deadline height for this input. This is // different from the DeadlineHeight in its params as it's an actual // value than an option. - deadlineHeight int32 + DeadlineHeight int32 } // String returns a human readable interpretation of the pending input. @@ -872,10 +872,6 @@ func (s *UtxoSweeper) sweep(set InputSet) error { // markInputsPendingPublish updates the pending inputs with the given tx // inputs. It also increments the `publishAttempts`. func (s *UtxoSweeper) markInputsPendingPublish(set InputSet) { - // Create a default deadline height, which will be used when there's no - // DeadlineHeight specified for a given input. - defaultDeadline := s.currentHeight + int32(s.cfg.NoDeadlineConfTarget) - // Reschedule sweep. for _, input := range set.Inputs() { pi, ok := s.inputs[input.OutPoint()] @@ -907,11 +903,6 @@ func (s *UtxoSweeper) markInputsPendingPublish(set InputSet) { // Record another publish attempt. pi.publishAttempts++ - - // Set the acutal deadline height. - pi.deadlineHeight = pi.params.DeadlineHeight.UnwrapOr( - defaultDeadline, - ) } } @@ -1082,7 +1073,7 @@ func (s *UtxoSweeper) handlePendingSweepsReq( LastFeeRate: inp.lastFeeRate, BroadcastAttempts: inp.publishAttempts, Params: inp.params, - DeadlineHeight: uint32(inp.deadlineHeight), + DeadlineHeight: uint32(inp.DeadlineHeight), } } @@ -1173,6 +1164,11 @@ func (s *UtxoSweeper) handleUpdateReq(req *updateReq) ( // TODO(yy): a dedicated state? sweeperInput.state = Init + // If the new input specifies a deadline, update the deadline height. + sweeperInput.DeadlineHeight = req.params.DeadlineHeight.UnwrapOr( + sweeperInput.DeadlineHeight, + ) + resultChan := make(chan Result, 1) sweeperInput.listeners = append(sweeperInput.listeners, resultChan) @@ -1204,6 +1200,10 @@ func (s *UtxoSweeper) mempoolLookup(op wire.OutPoint) fn.Option[wire.MsgTx] { // handleNewInput processes a new input by registering spend notification and // scheduling sweeping for it. func (s *UtxoSweeper) handleNewInput(input *sweepInputMessage) error { + // Create a default deadline height, which will be used when there's no + // DeadlineHeight specified for a given input. + defaultDeadline := s.currentHeight + int32(s.cfg.NoDeadlineConfTarget) + outpoint := input.input.OutPoint() pi, pending := s.inputs[outpoint] if pending { @@ -1228,6 +1228,10 @@ func (s *UtxoSweeper) handleNewInput(input *sweepInputMessage) error { Input: input.input, params: input.params, rbf: rbfInfo, + // Set the acutal deadline height. + DeadlineHeight: input.params.DeadlineHeight.UnwrapOr( + defaultDeadline, + ), } s.inputs[outpoint] = pi @@ -1343,6 +1347,11 @@ func (s *UtxoSweeper) handleExistingInput(input *sweepInputMessage, oldInput.params = input.params oldInput.Input = input.input + // If the new input specifies a deadline, update the deadline height. + oldInput.DeadlineHeight = input.params.DeadlineHeight.UnwrapOr( + oldInput.DeadlineHeight, + ) + // Add additional result channel to signal spend of this input. oldInput.listeners = append(oldInput.listeners, input.resultChan) @@ -1541,12 +1550,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) { - // Create a default deadline height, which will be used when there's no - // DeadlineHeight specified for a given input. - defaultDeadline := s.currentHeight + int32(s.cfg.NoDeadlineConfTarget) - // Cluster all of our inputs based on the specific Aggregator. - sets := s.cfg.Aggregator.ClusterInputs(inputs, defaultDeadline) + sets := s.cfg.Aggregator.ClusterInputs(inputs) // sweepWithLock is a helper closure that executes the sweep within a // coin select lock to prevent the coins being selected for other diff --git a/sweep/sweeper_test.go b/sweep/sweeper_test.go index 6a164ccd0..bfa3ff778 100644 --- a/sweep/sweeper_test.go +++ b/sweep/sweeper_test.go @@ -2747,9 +2747,7 @@ func TestSweepPendingInputs(t *testing.T) { pis := make(InputsMap) // Mock the aggregator to return the mocked input sets. - expectedDeadlineUsed := testHeight + DefaultDeadlineDelta - aggregator.On("ClusterInputs", pis, - expectedDeadlineUsed).Return([]InputSet{ + aggregator.On("ClusterInputs", pis).Return([]InputSet{ setNeedWallet, normalSet, })