package main import ( "sync" "github.com/davecgh/go-spew/spew" "github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/channeldb" "github.com/lightningnetwork/lnd/lnwallet" "github.com/roasbeef/btcd/txscript" "github.com/roasbeef/btcd/wire" "github.com/roasbeef/btcutil" ) // utxoNursery is a system dedicated to incubating time-locked outputs created // by the broadcast of a commitment transaction either by us, or the remote // peer. The nursery accepts outputs and "incubates" them until they've reached // maturity, then sweep the outputs into the source wallet. An output is // considered mature after the relative time-lock within the pkScript has // passed. As outputs reach their maturity age, they're sweeped in batches into // the source wallet, returning the outputs so they can be used within future // channels, or regular Bitcoin transactions. type utxoNursery struct { sync.RWMutex notifier chainntnfs.ChainNotifier wallet *lnwallet.LightningWallet db channeldb.DB requests chan *incubationRequest // TODO(roasbeef): persist to disk afterwards unstagedOutputs map[wire.OutPoint]*immatureOutput stagedOutputs map[uint32][]*immatureOutput started uint32 stopped uint32 quit chan struct{} wg sync.WaitGroup } // newUtxoNursery creates a new instance of the utxoNursery from a // ChainNotifier and LightningWallet instance. func newUtxoNursery(notifier chainntnfs.ChainNotifier, wallet *lnwallet.LightningWallet) *utxoNursery { return &utxoNursery{ notifier: notifier, wallet: wallet, requests: make(chan *incubationRequest), unstagedOutputs: make(map[wire.OutPoint]*immatureOutput), stagedOutputs: make(map[uint32][]*immatureOutput), quit: make(chan struct{}), } } // Start launches all goroutines the utxoNursery needs to properly carry out // its duties. func (u *utxoNursery) Start() error { u.wg.Add(1) go u.incubator() return nil } // Stop gracefully shutsdown any lingering goroutines launched during normal // operation of the utxoNursery. func (u *utxoNursery) Stop() error { close(u.quit) u.wg.Wait() return nil } // incubator is tasked with watching over all immature outputs until they've // reached "maturity", after which they'll be sweeped into the underlying // wallet in batches within a single transaction. Immature outputs can be // divided into three stages: early stage, mid stage, and final stage. During // the early stage, the transaction containing the output has not yet been // confirmed. Once the txn creating the output is confirmed, then output moves // to the mid stage wherein a dedicated goroutine waits until it has reached // "maturity". Once an output is mature, it will be sweeped into the wallet at // the earlier possible height. func (u *utxoNursery) incubator() { // Register with the notifier to receive notifications for each newly // connected block. newBlocks, err := u.notifier.RegisterBlockEpochNtfn() if err != nil { utxnLog.Errorf("unable to register for block epoch "+ "notifications: %v", err) } // Outputs that are transitioning from early to mid-stage are sent over // this channel by each output's dedicated watcher goroutine. midStageOutputs := make(chan *immatureOutput) out: for { select { case earlyStagers := <-u.requests: utxnLog.Infof("Incubating %v new outputs", len(earlyStagers.outputs)) for _, immatureUtxo := range earlyStagers.outputs { outpoint := immatureUtxo.outPoint sourceTXID := outpoint.Hash // Register for a confirmation once the // generating txn has been confirmed. confChan, err := u.notifier.RegisterConfirmationsNtfn(&sourceTXID, 1) if err != nil { utxnLog.Errorf("unable to register for confirmations "+ "for txid: %v", sourceTXID) continue } // TODO(roasbeef): should be an on-disk // at-least-once task queue u.unstagedOutputs[outpoint] = immatureUtxo // Launch a dedicated goroutine which will send // the output back to the incubator once the // source txn has been confirmed. go func() { confHeight, ok := <-confChan.Confirmed if !ok { utxnLog.Errorf("notification chan "+ "closed, can't advance output %v", outpoint) } utxnLog.Infof("Outpoint %v confirmed in "+ "block %v moving to mid-stage", outpoint, confHeight) immatureUtxo.confHeight = uint32(confHeight) midStageOutputs <- immatureUtxo }() } case midUtxo := <-midStageOutputs: // The transaction creating the output has been // created, so we move it from early stage to // mid-stage. delete(u.unstagedOutputs, midUtxo.outPoint) // TODO(roasbeef): your off-by-one sense are tingling... maturityHeight := midUtxo.confHeight + midUtxo.blocksToMaturity u.stagedOutputs[maturityHeight] = append(u.stagedOutputs[maturityHeight], midUtxo) utxnLog.Infof("Outpoint %v now mid-stage, will mature "+ "at height %v (delay of %v)", midUtxo.outPoint, maturityHeight, midUtxo.blocksToMaturity) case epoch := <-newBlocks.Epochs: // A new block has just been connected, check to see if // we have any new outputs that can be swept into the // wallet. newHeight := uint32(epoch.Height) matureOutputs, ok := u.stagedOutputs[newHeight] if !ok { continue } utxnLog.Infof("New block: height=%v hash=%v, "+ "sweeping %v mature outputs", newHeight, epoch.Hash, len(matureOutputs)) // Create a transation which sweeps all the newly // mature outputs into a output controlled by the // wallet. // TODO(roasbeef): can be more intelligent about // buffering outputs to be more efficient on-chain. sweepTx, err := u.createSweepTx(matureOutputs) if err != nil { // TODO(roasbeef): retry logic? utxnLog.Errorf("unable to create sweep tx: %v", err) continue } utxnLog.Infof("Sweeping %v time-locked outputs "+ "with sweep tx: %v", len(matureOutputs), newLogClosure(func() string { return spew.Sdump(sweepTx) })) // With the sweep transaction fully signed, broadcast // the transaction to the network. Additionally, we can // stop tracking these outputs as they've just been // sweeped. err = u.wallet.PublishTransaction(sweepTx) if err != nil { utxnLog.Errorf("unable to broadcast sweep tx: %v, %v", err, spew.Sdump(sweepTx)) continue } delete(u.stagedOutputs, newHeight) case <-u.quit: break out } } u.wg.Done() } // createSweepTx creates a final sweeping transaction with all witnesses // inplace for all inputs. The created transaction has a single output sending // all the funds back to the source wallet. func (u *utxoNursery) createSweepTx(matureOutputs []*immatureOutput) (*wire.MsgTx, error) { sweepAddr, err := u.wallet.NewAddress(lnwallet.WitnessPubKey, false) if err != nil { return nil, err } pkScript, err := txscript.PayToAddrScript(sweepAddr) if err != nil { return nil, err } var totalSum btcutil.Amount for _, o := range matureOutputs { totalSum += o.amt } sweepTx := wire.NewMsgTx() sweepTx.Version = 2 sweepTx.AddTxOut(&wire.TxOut{ PkScript: pkScript, Value: int64(totalSum - 1000), }) for _, utxo := range matureOutputs { sweepTx.AddTxIn(&wire.TxIn{ PreviousOutPoint: utxo.outPoint, // TODO(roasbeef): assumes pure block delays Sequence: utxo.blocksToMaturity, }) } // TODO(roasbeef): insert fee calculation // * remove hardcoded fee above // With all the inputs in place, use each output's unique witness // function to generate the final witness required for spending. hashCache := txscript.NewTxSigHashes(sweepTx) for i, txIn := range sweepTx.TxIn { witness, err := matureOutputs[i].witnessFunc(sweepTx, hashCache, i) if err != nil { return nil, err } txIn.Witness = witness } return sweepTx, nil } // witnessFunc represents a function which is able to generate the final // witness for a particular public key script. This function acts as an // abstraction layer, hidiing the details of the underlying script from the // utxoNursery. type witnessGenerator func(tx *wire.MsgTx, hc *txscript.TxSigHashes, inputIndex int) ([][]byte, error) // immatureOutput encapsulates an immature output. The struct includes a // witnessGenerator closure which will be used to generate the witness required // to sweep the output once it's mature. // TODO(roasbeef): make into interface? can't gob functions type immatureOutput struct { amt btcutil.Amount outPoint wire.OutPoint witnessFunc witnessGenerator // TODO(roasbeef): using block timeouts everywhere currently, will need // to modify logic later to account for MTP based timeouts. blocksToMaturity uint32 confHeight uint32 } // incubationRequest is a request to the utxoNursery to incubate a set of // outputs until their mature, finally sweeping them into the wallet once // available. type incubationRequest struct { outputs []*immatureOutput } // incubateOutputs sends a request to utxoNursery to incubate the outputs // defined within the summary of a closed channel. Induvidually, as all outputs // reach maturity they'll be sweeped back into the wallet. func (u *utxoNursery) incubateOutputs(closeSummary *lnwallet.ForceCloseSummary) { // TODO(roasbeef): should use factory func here based on an interface // * spend here also assumes delay is blocked bsaed, and in range witnessFunc := func(tx *wire.MsgTx, hc *txscript.TxSigHashes, inputIndex int) ([][]byte, error) { desc := closeSummary.SelfOutputSignDesc desc.SigHashes = hc desc.InputIndex = inputIndex return lnwallet.CommitSpendTimeout(u.wallet.Signer, desc, tx) } outputAmt := btcutil.Amount(closeSummary.SelfOutputSignDesc.Output.Value) selfOutput := &immatureOutput{ amt: outputAmt, outPoint: closeSummary.SelfOutpoint, witnessFunc: witnessFunc, blocksToMaturity: closeSummary.SelfOutputMaturity, } u.requests <- &incubationRequest{ outputs: []*immatureOutput{selfOutput}, } }