package sweepbatcher import ( "bytes" "context" "fmt" "github.com/btcsuite/btcd/blockchain" "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/chaincfg" "github.com/btcsuite/btcd/txscript" "github.com/btcsuite/btcd/wire" "github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lnwallet/chainfee" "golang.org/x/sync/errgroup" ) // ensurePresigned checks that there is a presigned transaction spending the // inputs of this group only. If allowNonEmptyBatch is false, the batch must be // empty. func (b *batch) ensurePresigned(ctx context.Context, newSweeps []*sweep, allowNonEmptyBatch bool) error { if b.cfg.presignedHelper == nil { return fmt.Errorf("presignedHelper is not installed") } if len(b.sweeps) != 0 && !allowNonEmptyBatch { return fmt.Errorf("ensurePresigned should be done when " + "adding to an empty batch") } minRelayFeeRate, err := b.wallet.MinRelayFee(ctx) if err != nil { return fmt.Errorf("failed to get minRelayFee: %w", err) } return ensurePresigned( ctx, newSweeps, b.cfg.presignedHelper, minRelayFeeRate, b.cfg.chainParams, ) } // presignedTxChecker has methods to check if the inputs are presigned. type presignedTxChecker interface { destPkScripter presigner } // ensurePresigned checks that there is a presigned transaction spending the // inputs of this group only. func ensurePresigned(ctx context.Context, newSweeps []*sweep, presignedTxChecker presignedTxChecker, minRelayFeeRate chainfee.SatPerKWeight, chainParams *chaincfg.Params) error { sweeps := make([]sweep, len(newSweeps)) for i, s := range newSweeps { sweeps[i] = sweep{ outpoint: s.outpoint, value: s.value, presigned: s.presigned, change: s.change, } } // The sweeps are ordered inside the group, the first one is the primary // outpoint in the batch. primarySweepID := sweeps[0].outpoint // Cache the destination address. destAddr, err := getPresignedSweepsDestAddr( ctx, presignedTxChecker, primarySweepID, chainParams, ) if err != nil { return fmt.Errorf("failed to find destination address: %w", err) } // Set LockTime to 0. It is not critical. const currentHeight = 0 // Check if we can sign with minimum fee rate. feeRate := minRelayFeeRate tx, _, _, _, err := constructUnsignedTx( sweeps, destAddr, currentHeight, feeRate, minRelayFeeRate, ) if err != nil { return fmt.Errorf("failed to construct unsigned tx "+ "for feeRate %v: %w", feeRate, err) } // Check of a presigned transaction exists. var batchAmt btcutil.Amount for _, sweep := range newSweeps { batchAmt += sweep.value } const loadOnly = true signedTx, err := presignedTxChecker.SignTx( ctx, primarySweepID, tx, batchAmt, feeRate, feeRate, loadOnly, ) if err != nil { return fmt.Errorf("failed to find a presigned transaction "+ "for feeRate %v, txid of the template is %v, inputs: %d, "+ "outputs: %d: %w", feeRate, tx.TxHash(), len(tx.TxIn), len(tx.TxOut), err) } // Check the SignTx worked correctly. err = CheckSignedTx(tx, signedTx, batchAmt, feeRate) if err != nil { return fmt.Errorf("signed tx doesn't correspond the "+ "unsigned tx: %w", err) } return nil } // getOrderedSweeps returns the sweeps of the batch in the order they were // added. The method must be called from the event loop of the batch. func (b *batch) getOrderedSweeps(ctx context.Context) ([]sweep, error) { // We use the DB just to know the order. Sweeps are copied from RAM. utxo2sweep := make(map[wire.OutPoint]sweep, len(b.sweeps)) for _, s := range b.sweeps { utxo2sweep[s.outpoint] = s } dbSweeps, err := b.store.FetchBatchSweeps(ctx, b.id) if err != nil { return nil, fmt.Errorf("FetchBatchSweeps(%d) failed: %w", b.id, err) } dbSweeps = filterDbSweeps(b.cfg.skippedTxns, dbSweeps) if len(dbSweeps) != len(utxo2sweep) { return nil, fmt.Errorf("FetchBatchSweeps(%d) returned %d "+ "sweeps, len(b.sweeps) is %d", b.id, len(dbSweeps), len(utxo2sweep)) } orderedSweeps := make([]sweep, len(dbSweeps)) for i, dbSweep := range dbSweeps { // Sanity check: make sure dbSweep.ID grows. if i > 0 && dbSweep.ID <= dbSweeps[i-1].ID { return nil, fmt.Errorf("sweep ID does not grow: %d->%d", dbSweeps[i-1].ID, dbSweep.ID) } s, has := utxo2sweep[dbSweep.Outpoint] if !has { return nil, fmt.Errorf("FetchBatchSweeps(%d) returned "+ "unknown sweep %v", b.id, dbSweep.Outpoint) } orderedSweeps[i] = s } return orderedSweeps, nil } // getSweepsGroups returns groups in which sweeps were added to the batch. // All the sweeps are sorted by addition order and grouped by swap. // The method must be called from the event loop of the batch. func (b *batch) getSweepsGroups(ctx context.Context) ([][]sweep, error) { orderedSweeps, err := b.getOrderedSweeps(ctx) if err != nil { return nil, fmt.Errorf("getOrderedSweeps(%d) failed: %w", b.id, err) } groups := [][]sweep{} for _, s := range orderedSweeps { index := len(groups) - 1 // Start new group if there are no groups or new swap starts. if len(groups) == 0 || s.swapHash != groups[index][0].swapHash { groups = append(groups, []sweep{}) index++ } groups[index] = append(groups[index], s) } // Sanity check: make sure the number of groups is the same as the // number of distinct swaps. swapsSet := make(map[lntypes.Hash]struct{}, len(groups)) for _, s := range orderedSweeps { swapsSet[s.swapHash] = struct{}{} } if len(swapsSet) != len(groups) { return nil, fmt.Errorf("batch %d: there are %d groups of "+ "sweeps and %d distinct swaps", b.id, len(groups), len(swapsSet)) } return groups, nil } // presign tries to presign batch sweep transactions composed of this batch and // the sweep. In addition to that it presigns sweep transactions for any subset // of sweeps that could remain if one of the sweep transactions gets confirmed. // This can be done efficiently, since we keep track of the order in which // sweeps are added and the associated swap hashes. So we presign transactions // sweeping all the sweeps starting at some past sweeps group. For each inputs // layout it presigns many transactions with different fee rates. func (b *batch) presign(ctx context.Context, newSweeps []*sweep) error { if b.cfg.presignedHelper == nil { return fmt.Errorf("presignedHelper is not installed") } if len(b.sweeps) == 0 { return fmt.Errorf("presigning should be done when adding to " + "a non-empty batch") } // priorityConfTarget defines the confirmation target for quick // inclusion in a block. A value of 2, rather than 1, is used to prevent // fee estimator from failing. // See https://github.com/lightninglabs/loop/issues/898 const priorityConfTarget = 2 // Find the feerate needed to get into next block. nextBlockFeeRate, err := b.wallet.EstimateFeeRate( ctx, priorityConfTarget, ) if err != nil { return fmt.Errorf("failed to get nextBlockFeeRate: %w", err) } b.Infof("nextBlockFeeRate is %v", nextBlockFeeRate) // Find the minRelayFeeRate. minRelayFeeRate, err := b.wallet.MinRelayFee(ctx) if err != nil { return fmt.Errorf("failed to get minRelayFeeRate: %w", err) } b.Infof("minRelayFeeRate is %v", minRelayFeeRate) // We need to restore previously added groups. We can do it by reading // all the sweeps from DB (they must be ordered) and grouping by swap. groups, err := b.getSweepsGroups(ctx) if err != nil { return fmt.Errorf("getSweepsGroups failed: %w", err) } if len(groups) == 0 { return fmt.Errorf("getSweepsGroups returned no sweeps groups") } // Now presign a transaction spending a suffix of groups as well as new // sweeps. Any non-empty suffix of groups may remain non-swept after // some past tx is confirmed. for len(groups) != 0 { // Create the list of sweeps from the remaining groups and new // sweeps. sweeps := make([]sweep, 0, len(b.sweeps)+len(newSweeps)) for _, group := range groups { sweeps = append(sweeps, group...) } for _, sweep := range newSweeps { sweeps = append(sweeps, *sweep) } // The primarySweepID is the first sweep from the list of // remaining sweeps if previous groups are confirmed. primarySweepID := sweeps[0].outpoint // Cache the destination address. destAddr, err := getPresignedSweepsDestAddr( ctx, b.cfg.presignedHelper, primarySweepID, b.cfg.chainParams, ) if err != nil { return fmt.Errorf("failed to find destination "+ "address: %w", err) } err = presign( ctx, b.cfg.presignedHelper, destAddr, primarySweepID, sweeps, nextBlockFeeRate, minRelayFeeRate, ) if err != nil { return fmt.Errorf("failed to presign a transaction "+ "of %d sweeps: %w", len(sweeps), err) } // Cut a group to proceed to next suffix of original groups. groups = groups[1:] } // Ensure that a batch spending new sweeps only has been presigned by // PresignSweepsGroup. const allowNonEmptyBatch = true err = b.ensurePresigned(ctx, newSweeps, allowNonEmptyBatch) if err != nil { return fmt.Errorf("new sweeps were not presigned; this means "+ "that PresignSweepsGroup was not called prior to "+ "AddSweep for the group: %w", err) } return nil } // presigner tries to presign a batch transaction. type presigner interface { // SignTx signs an unsigned transaction or returns a pre-signed tx. // It is only called with loadOnly=true by ensurePresigned. SignTx(ctx context.Context, primarySweepID wire.OutPoint, tx *wire.MsgTx, inputAmt btcutil.Amount, minRelayFee, feeRate chainfee.SatPerKWeight, loadOnly bool) (*wire.MsgTx, error) } // presign tries to presign batch sweep transactions of the sweeps. It signs // multiple versions of the transaction to make sure there is a transaction to // be published if minRelayFee grows. If feerate is high, then a presigned tx // gets LockTime equal to timeout minus 50 blocks, as a precautionary measure. // A feerate is considered high if it is at least 100 sat/vbyte AND is at least // 10x of the current next block feerate. func presign(ctx context.Context, presigner presigner, destAddr btcutil.Address, primarySweepID wire.OutPoint, sweeps []sweep, nextBlockFeeRate chainfee.SatPerKWeight, minRelayFeeRate chainfee.SatPerKWeight) error { if presigner == nil { return fmt.Errorf("presigner is not installed") } if len(sweeps) == 0 { return fmt.Errorf("there are no sweeps") } if nextBlockFeeRate == 0 { return fmt.Errorf("nextBlockFeeRate is not set") } if minRelayFeeRate == 0 { return fmt.Errorf("minRelayFeeRate is not set") } // Keep track of the total amount this batch is sweeping back. batchAmt := btcutil.Amount(0) for _, sweep := range sweeps { batchAmt += sweep.value } // Find the sweep with the earliest expiry. timeout := sweeps[0].timeout for _, sweep := range sweeps[1:] { timeout = min(timeout, sweep.timeout) } if timeout <= 0 { return fmt.Errorf("timeout is invalid: %d", timeout) } // Go from minRelayFeeRate to 2k sat/vbyte with step of 1.2x. var start = minRelayFeeRate const ( stop = chainfee.AbsoluteFeePerKwFloor * 2_000 factorPPM = 1_200_000 timeoutThreshold = 50 ) // Calculate the locktime value to use for high feerate transactions. // If timeout <= timeoutThreshold, don't set LockTime (keep value 0). var highFeeRateLockTime uint32 if timeout > timeoutThreshold { highFeeRateLockTime = uint32(timeout - timeoutThreshold) } // Calculate which feerate to consider high. At least 100 sat/vbyte and // at least 10x of current nextBlockFeeRate. highFeeRate := max(100*chainfee.FeePerKwFloor, 10*nextBlockFeeRate) // Set LockTime to 0. It is not critical. const currentHeight = 0 eg, grCtx := errgroup.WithContext(ctx) for fr := start; fr <= stop; fr = (fr * factorPPM) / 1_000_000 { // Construct an unsigned transaction for this fee rate. tx, _, feeForWeight, fee, err := constructUnsignedTx( sweeps, destAddr, currentHeight, fr, minRelayFeeRate, ) if err != nil { return fmt.Errorf("failed to construct unsigned tx "+ "for feeRate %v: %w", fr, err) } // If the feerate is high enough, set locktime to prevent // broadcasting such a transaction too early by mistake. if fr >= highFeeRate { tx.LockTime = highFeeRateLockTime } // Try to presign this transaction. const loadOnly = false eg.Go(func() error { _, err := presigner.SignTx( grCtx, primarySweepID, tx, batchAmt, minRelayFeeRate, fr, loadOnly, ) if err != nil { return fmt.Errorf("failed to presign unsigned "+ "tx %v for feeRate %v: %w", tx.TxHash(), fr, err) } return nil }) // If fee was clamped, stop here, because fee rate won't grow. if fee < feeForWeight { break } } if err := eg.Wait(); err != nil { return fmt.Errorf("presigning of batch of primarySweepID %v "+ "failed: %w", primarySweepID, err) } return nil } // publishPresigned creates sweep transaction using a custom transaction signer // and publishes it. It returns the fee of the transaction, and an error (if // signing and/or publishing failed) and a boolean flag indicating signing // success. This mode is incompatible with an external address. func (b *batch) publishPresigned(ctx context.Context) (btcutil.Amount, error, bool) { // Sanity check, there should be at least 1 sweep in this batch. if len(b.sweeps) == 0 { return 0, fmt.Errorf("no sweeps in batch"), false } // Make sure that no external address is used. for _, sweep := range b.sweeps { if sweep.isExternalAddr { return 0, fmt.Errorf("external address was used with " + "a custom transaction signer"), false } } // Determine the current minimum relay fee based on our chain backend. minRelayFeeRate, err := b.wallet.MinRelayFee(ctx) if err != nil { return 0, fmt.Errorf("failed to get minRelayFeeRate: %w", err), false } // Cache current height and desired feerate of the batch. currentHeight := b.currentHeight feeRate := max(b.rbfCache.FeeRate, minRelayFeeRate) // Append this sweep to an array of sweeps. This is needed to keep the // order of sweeps stored, as iterating the sweeps map does not // guarantee same order. sweeps := make([]sweep, 0, len(b.sweeps)) for _, sweep := range b.sweeps { sweeps = append(sweeps, sweep) } // Cache the destination address. address, err := getPresignedSweepsDestAddr( ctx, b.cfg.presignedHelper, b.primarySweepID, b.cfg.chainParams, ) if err != nil { return 0, fmt.Errorf("failed to find destination address: %w", err), false } // Construct unsigned batch transaction. tx, weight, _, fee, err := constructUnsignedTx( sweeps, address, currentHeight, feeRate, minRelayFeeRate, ) if err != nil { return 0, fmt.Errorf("failed to construct tx: %w", err), false } // Adjust feeRate, because it may have been clamped. feeRate = chainfee.NewSatPerKWeight(fee, weight) // Calculate total input amount. batchAmt := btcutil.Amount(0) for _, sweep := range sweeps { batchAmt += sweep.value } // Get a pre-signed transaction. const loadOnly = false signedTx, err := b.cfg.presignedHelper.SignTx( ctx, b.primarySweepID, tx, batchAmt, minRelayFeeRate, feeRate, loadOnly, ) if err != nil { return 0, fmt.Errorf("failed to sign tx: %w", err), false } // Run sanity checks to make sure presignedHelper.SignTx complied with // all the invariants. err = CheckSignedTx(tx, signedTx, batchAmt, minRelayFeeRate) if err != nil { return 0, fmt.Errorf("signed tx doesn't correspond the "+ "unsigned tx: %w", err), false } tx = signedTx txHash := tx.TxHash() // Make sure tx weight matches the expected value. realWeight := lntypes.WeightUnit( blockchain.GetTransactionWeight(btcutil.NewTx(tx)), ) if realWeight != weight { b.Warnf("actual weight of tx %v is %v, estimated as %d", txHash, realWeight, weight) } // Find actual fee rate of the signed transaction. It may differ from // the desired fee rate, because SignTx may return a presigned tx. var output btcutil.Amount for _, txOut := range tx.TxOut { output += btcutil.Amount(txOut.Value) } fee = batchAmt - output signedFeeRate := chainfee.NewSatPerKWeight(fee, realWeight) numSweeps := len(tx.TxIn) numChange := len(tx.TxOut) - 1 b.Infof("attempting to publish custom signed tx=%v, desiredFeerate=%v,"+ " signedFeeRate=%v, weight=%v, fee=%v, sweeps=%d, "+ "changeOutputs=%d, destAddr=%s", txHash, feeRate, signedFeeRate, realWeight, fee, numSweeps, numChange, address) b.debugLogTx("serialized batch", tx) // Publish the transaction. err = b.wallet.PublishTransaction(ctx, tx, b.cfg.txLabeler(b.id)) if err != nil { return 0, fmt.Errorf("publishing tx failed: %w", err), true } // Store the batch transaction's txid and pkScript, for monitoring // purposes. b.batchTxid = &txHash b.batchPkScript = tx.TxOut[0].PkScript // Update cached FeeRate not to broadcast a tx with lower feeRate. b.rbfCache.FeeRate = max(b.rbfCache.FeeRate, signedFeeRate) return fee, nil, true } // destPkScripter returns destination pkScript used by the sweep batch. type destPkScripter interface { // DestPkScript returns destination pkScript used by the sweep batch // with the primary outpoint specified. Returns an error, if such tx // doesn't exist. If there are many such transactions, returns any of // pkScript's; all of them should have the same destination pkScript. DestPkScript(ctx context.Context, primarySweepID wire.OutPoint) ([]byte, error) } // getPresignedSweepsDestAddr returns the destination address used by the // primary outpoint. The function must be used in presigned mode only. func getPresignedSweepsDestAddr(ctx context.Context, helper destPkScripter, primarySweepID wire.OutPoint, chainParams *chaincfg.Params) (btcutil.Address, error) { // Load pkScript from the presigned helper. pkScriptBytes, err := helper.DestPkScript(ctx, primarySweepID) if err != nil { return nil, fmt.Errorf("presignedHelper.DestPkScript failed "+ "for primarySweepID %v: %w", primarySweepID, err) } // Convert pkScript to btcutil.Address. pkScript, err := txscript.ParsePkScript(pkScriptBytes) if err != nil { return nil, fmt.Errorf("txscript.ParsePkScript failed for "+ "pkScript %x returned for primarySweepID %v: %w", pkScriptBytes, primarySweepID, err) } address, err := pkScript.Address(chainParams) if err != nil { return nil, fmt.Errorf("pkScript.Address failed for "+ "pkScript %x returned for primarySweepID %v: %w", pkScriptBytes, primarySweepID, err) } return address, nil } // CheckSignedTx makes sure that signedTx matches the unsignedTx. It checks // according to criteria specified in the description of PresignedHelper.SignTx. func CheckSignedTx(unsignedTx, signedTx *wire.MsgTx, inputAmt btcutil.Amount, minRelayFee chainfee.SatPerKWeight) error { // Make sure all inputs of signedTx have a non-empty witness. for _, txIn := range signedTx.TxIn { if len(txIn.Witness) == 0 { return fmt.Errorf("input %s of signed tx is not signed", txIn.PreviousOutPoint) } } // Make sure the set of inputs is the same. unsignedMap := make(map[wire.OutPoint]uint32, len(unsignedTx.TxIn)) for _, txIn := range unsignedTx.TxIn { unsignedMap[txIn.PreviousOutPoint] = txIn.Sequence } for _, txIn := range signedTx.TxIn { seq, has := unsignedMap[txIn.PreviousOutPoint] if !has { return fmt.Errorf("input %s is new in signed tx", txIn.PreviousOutPoint) } if seq != txIn.Sequence { return fmt.Errorf("sequence mismatch in input %s: "+ "%d in unsigned, %d in signed", txIn.PreviousOutPoint, seq, txIn.Sequence) } delete(unsignedMap, txIn.PreviousOutPoint) } for outpoint := range unsignedMap { return fmt.Errorf("input %s is missing in signed tx", outpoint) } // Compare outputs. if len(unsignedTx.TxOut) != len(signedTx.TxOut) { return fmt.Errorf("unsigned tx has %d outputs, signed tx has "+ "%d outputs, should be equal", len(unsignedTx.TxOut), len(signedTx.TxOut)) } for i, o := range unsignedTx.TxOut { if !bytes.Equal(o.PkScript, signedTx.TxOut[i].PkScript) { return fmt.Errorf("mismatch of output pkScript: %x, %x", o.PkScript, signedTx.TxOut[i].PkScript) } if i != 0 && o.Value != signedTx.TxOut[i].Value { return fmt.Errorf("mismatch of output value: %d, %d", o.Value, signedTx.TxOut[i].Value) } } // Calculate the total value of all outputs to help determine the // transaction fee. totalOutputValue := btcutil.Amount(0) for _, o := range signedTx.TxOut { totalOutputValue += btcutil.Amount(o.Value) } // Find the feerate of signedTx. fee := inputAmt - totalOutputValue weight := lntypes.WeightUnit( blockchain.GetTransactionWeight(btcutil.NewTx(signedTx)), ) feeRate := chainfee.NewSatPerKWeight(fee, weight) if feeRate < minRelayFee { return fmt.Errorf("feerate (%v) of signed tx is lower than "+ "minRelayFee (%v)", feeRate, minRelayFee) } // Check LockTime. if signedTx.LockTime > unsignedTx.LockTime { return fmt.Errorf("locktime (%d) of signed tx is higher than "+ "locktime of unsigned tx (%d)", signedTx.LockTime, unsignedTx.LockTime) } // Check Version. if signedTx.Version != unsignedTx.Version { return fmt.Errorf("version (%d) of signed tx is not equal to "+ "version of unsigned tx (%d)", signedTx.Version, unsignedTx.Version) } return nil }