sweep: add handleInitialBroadcast to handle initial broadcast

This commit adds a new method `handleInitialBroadcast` to handle the initial broadcast. Previously we'd broadcast immediately inside `Broadcast`, which soon will not work after the `blockbeat` is implemented as the action to publish is now always triggered by a new block. Meanwhile, we still keep the option to bypass the block trigger so users can broadcast immediately by setting `Immediate` to true.
2024-12-20 17:53:59 +08:00
1 parent 2479dc7f2e
commit 77ff2c0585
@@ -376,40 +376,52 @@ func (t *TxPublisher) isNeutrinoBackend() bool {
 	return t.cfg.Wallet.BackEnd() == "neutrino"
 }

-// Broadcast is used to publish the tx created from the given inputs. It will,
-// 1. init a fee function based on the given strategy.
-// 2. create an RBF-compliant tx and monitor it for confirmation.
-// 3. notify the initial broadcast result back to the caller.
-// The initial broadcast is guaranteed to be RBF-compliant unless the budget
-// specified cannot cover the fee.
+// Broadcast is used to publish the tx created from the given inputs. It will
+// register the broadcast request and return a chan to the caller to subscribe
+// the broadcast result. The initial broadcast is guaranteed to be
+// RBF-compliant unless the budget specified cannot cover the fee.
 //
 // NOTE: part of the Bumper interface.
 func (t *TxPublisher) Broadcast(req *BumpRequest) (<-chan *BumpResult, error) {
 	log.Tracef("Received broadcast request: %s", lnutils.SpewLogClosure(
 		req))

-	// Attempt an initial broadcast which is guaranteed to comply with the
-	// RBF rules.
-	result, err := t.initialBroadcast(req)
-	if err != nil {
-		log.Errorf("Initial broadcast failed: %v", err)
-
-		return nil, err
-	}
+	// Store the request.
+	requestID, record := t.storeInitialRecord(req)

 	// Create a chan to send the result to the caller.
 	subscriber := make(chan *BumpResult, 1)
-	t.subscriberChans.Store(result.requestID, subscriber)
+	t.subscriberChans.Store(requestID, subscriber)

-	// Send the initial broadcast result to the caller.
-	t.handleResult(result)
+	// Publish the tx immediately if specified.
+	if req.Immediate {
+		t.handleInitialBroadcast(record, requestID)
+	}

 	return subscriber, nil
 }

+// storeInitialRecord initializes a monitor record and saves it in the map.
+func (t *TxPublisher) storeInitialRecord(req *BumpRequest) (
+	uint64, *monitorRecord) {
+
+	// Increase the request counter.
+	//
+	// NOTE: this is the only place where we increase the counter.
+	requestID := t.requestCounter.Add(1)
+
+	// Register the record.
+	record := &monitorRecord{req: req}
+	t.records.Store(requestID, record)
+
+	return requestID, record
+}
+
 // initialBroadcast initializes a fee function, creates an RBF-compliant tx and
 // broadcasts it.
-func (t *TxPublisher) initialBroadcast(req *BumpRequest) (*BumpResult, error) {
+func (t *TxPublisher) initialBroadcast(requestID uint64,
+	req *BumpRequest) (*BumpResult, error) {
+
 	// Create a fee bumping algorithm to be used for future RBF.
 	feeAlgo, err := t.initializeFeeFunction(req)
 	if err != nil {
@@ -418,7 +430,7 @@ func (t *TxPublisher) initialBroadcast(req *BumpRequest) (*BumpResult, error) {

 	// Create the initial tx to be broadcasted. This tx is guaranteed to
 	// comply with the RBF restrictions.
-	requestID, err := t.createRBFCompliantTx(req, feeAlgo)
+	err = t.createRBFCompliantTx(requestID, req, feeAlgo)
 	if err != nil {
 		return nil, fmt.Errorf("create RBF-compliant tx: %w", err)
 	}
@@ -465,8 +477,8 @@ func (t *TxPublisher) initializeFeeFunction(
 // so by creating a tx, validate it using `TestMempoolAccept`, and bump its fee
 // and redo the process until the tx is valid, or return an error when non-RBF
 // related errors occur or the budget has been used up.
-func (t *TxPublisher) createRBFCompliantTx(req *BumpRequest,
-	f FeeFunction) (uint64, error) {
+func (t *TxPublisher) createRBFCompliantTx(requestID uint64, req *BumpRequest,
+	f FeeFunction) error {

 	for {
 		// Create a new tx with the given fee rate and check its
@@ -475,18 +487,19 @@ func (t *TxPublisher) createRBFCompliantTx(req *BumpRequest,

 		switch {
 		case err == nil:
-			// The tx is valid, return the request ID.
-			requestID := t.storeRecord(
-				sweepCtx.tx, req, f, sweepCtx.fee,
+			// The tx is valid, store it.
+			t.storeRecord(
+				requestID, sweepCtx.tx, req, f, sweepCtx.fee,
 				sweepCtx.outpointToTxIndex,
 			)

-			log.Infof("Created tx %v for %v inputs: feerate=%v, "+
-				"fee=%v, inputs=%v", sweepCtx.tx.TxHash(),
-				len(req.Inputs), f.FeeRate(), sweepCtx.fee,
+			log.Infof("Created initial sweep tx=%v for %v inputs: "+
+				"feerate=%v, fee=%v, inputs:\n%v",
+				sweepCtx.tx.TxHash(), len(req.Inputs),
+				f.FeeRate(), sweepCtx.fee,
 				inputTypeSummary(req.Inputs))

-			return requestID, nil
+			return nil

 		// If the error indicates the fees paid is not enough, we will
 		// ask the fee function to increase the fee rate and retry.
@@ -517,7 +530,7 @@ func (t *TxPublisher) createRBFCompliantTx(req *BumpRequest,
 				// cluster these inputs differetly.
 				increased, err = f.Increment()
 				if err != nil {
-					return 0, err
+					return err
 				}
 			}

@@ -527,21 +540,15 @@ func (t *TxPublisher) createRBFCompliantTx(req *BumpRequest,
 		// mempool acceptance.
 		default:
 			log.Debugf("Failed to create RBF-compliant tx: %v", err)
-			return 0, err
+			return err
 		}
 	}
 }

 // storeRecord stores the given record in the records map.
-func (t *TxPublisher) storeRecord(tx *wire.MsgTx, req *BumpRequest,
-	f FeeFunction, fee btcutil.Amount,
-	outpointToTxIndex map[wire.OutPoint]int) uint64 {
-
-	// Increase the request counter.
-	//
-	// NOTE: this is the only place where we increase the
-	// counter.
-	requestID := t.requestCounter.Add(1)
+func (t *TxPublisher) storeRecord(requestID uint64, tx *wire.MsgTx,
+	req *BumpRequest, f FeeFunction, fee btcutil.Amount,
+	outpointToTxIndex map[wire.OutPoint]int) {

 	// Register the record.
 	t.records.Store(requestID, &monitorRecord{
@@ -551,8 +558,6 @@ func (t *TxPublisher) storeRecord(tx *wire.MsgTx, req *BumpRequest,
 		fee:               fee,
 		outpointToTxIndex: outpointToTxIndex,
 	})
-
-	return requestID
 }

 // createAndCheckTx creates a tx based on the given inputs, change output
@@ -852,18 +857,27 @@ func (t *TxPublisher) processRecords() {
 	// confirmed.
 	confirmedRecords := make(map[uint64]*monitorRecord)

-	// feeBumpRecords stores a map of the records which need to be bumped.
+	// feeBumpRecords stores a map of records which need to be bumped.
 	feeBumpRecords := make(map[uint64]*monitorRecord)

-	// failedRecords stores a map of the records which has inputs being
-	// spent by a third party.
+	// failedRecords stores a map of records which has inputs being spent
+	// by a third party.
 	//
 	// NOTE: this is only used for neutrino backend.
 	failedRecords := make(map[uint64]*monitorRecord)

+	// initialRecords stores a map of records which are being created and
+	// published for the first time.
+	initialRecords := make(map[uint64]*monitorRecord)
+
 	// visitor is a helper closure that visits each record and divides them
 	// into two groups.
 	visitor := func(requestID uint64, r *monitorRecord) error {
+		if r.tx == nil {
+			initialRecords[requestID] = r
+			return nil
+		}
+
 		log.Tracef("Checking monitor recordID=%v for tx=%v", requestID,
 			r.tx.TxHash())

@@ -891,9 +905,14 @@ func (t *TxPublisher) processRecords() {
 		return nil
 	}

-	// Iterate through all the records and divide them into two groups.
+	// Iterate through all the records and divide them into four groups.
 	t.records.ForEach(visitor)

+	// Handle the initial broadcast.
+	for requestID, r := range initialRecords {
+		t.handleInitialBroadcast(r, requestID)
+	}
+
 	// For records that are confirmed, we'll notify the caller about this
 	// result.
 	for requestID, r := range confirmedRecords {
@@ -949,6 +968,69 @@ func (t *TxPublisher) handleTxConfirmed(r *monitorRecord, requestID uint64) {
 	t.handleResult(result)
 }

+// handleInitialBroadcast is called when a new request is received. It will
+// handle the initial tx creation and broadcast. In details,
+// 1. init a fee function based on the given strategy.
+// 2. create an RBF-compliant tx and monitor it for confirmation.
+// 3. notify the initial broadcast result back to the caller.
+func (t *TxPublisher) handleInitialBroadcast(r *monitorRecord,
+	requestID uint64) {
+
+	log.Debugf("Initial broadcast for requestID=%v", requestID)
+
+	var (
+		result *BumpResult
+		err    error
+	)
+
+	// Attempt an initial broadcast which is guaranteed to comply with the
+	// RBF rules.
+	result, err = t.initialBroadcast(requestID, r.req)
+	if err != nil {
+		log.Errorf("Initial broadcast failed: %v", err)
+
+		// We now decide what type of event to send.
+		var event BumpEvent
+
+		switch {
+		// When the error is due to a dust output, we'll send a
+		// TxFailed so these inputs can be retried with a different
+		// group in the next block.
+		case errors.Is(err, ErrTxNoOutput):
+			event = TxFailed
+
+		// When the error is due to budget being used up, we'll send a
+		// TxFailed so these inputs can be retried with a different
+		// group in the next block.
+		case errors.Is(err, ErrMaxPosition):
+			event = TxFailed
+
+		// When the error is due to zero fee rate delta, we'll send a
+		// TxFailed so these inputs can be retried in the next block.
+		case errors.Is(err, ErrZeroFeeRateDelta):
+			event = TxFailed
+
+		// Otherwise this is not a fee-related error and the tx cannot
+		// be retried. In that case we will fail ALL the inputs in this
+		// tx, which means they will be removed from the sweeper and
+		// never be tried again.
+		//
+		// TODO(yy): Find out which input is causing the failure and
+		// fail that one only.
+		default:
+			event = TxFatal
+		}
+
+		result = &BumpResult{
+			Event:     event,
+			Err:       err,
+			requestID: requestID,
+		}
+	}
+
+	t.handleResult(result)
+}
+
 // handleFeeBumpTx checks if the tx needs to be bumped, and if so, it will
 // attempt to bump the fee of the tx.
 //