diff --git a/Makefile b/Makefile index c4ee66aa2ef..c8477904de2 100644 --- a/Makefile +++ b/Makefile @@ -175,8 +175,8 @@ flakehunter: build flake-unit: @$(call print, "Flake hunting unit tests.") - $(UNIT) -count=1 - while [ $$? -eq 0 ]; do /bin/sh -c "$(UNIT) -count=1"; done + GOTRACEBACK=all $(UNIT) -count=1 + while [ $$? -eq 0 ]; do /bin/sh -c "GOTRACEBACK=all $(UNIT) -count=1"; done # ====== # TRAVIS diff --git a/contractcourt/briefcase.go b/contractcourt/briefcase.go index d7da21ea138..fce16f3a377 100644 --- a/contractcourt/briefcase.go +++ b/contractcourt/briefcase.go @@ -355,8 +355,6 @@ func (b *boltArbitratorLog) writeResolver(contractBucket *bolt.Bucket, rType = resolverTimeout case *htlcSuccessResolver: rType = resolverSuccess - case *htlcOutgoingContestResolver: - rType = resolverOutgoingContest case *htlcIncomingContestResolver: rType = resolverIncomingContest case *commitSweepResolver: @@ -467,16 +465,6 @@ func (b *boltArbitratorLog) FetchUnresolvedContracts() ([]ContractResolver, erro res = successRes - case resolverOutgoingContest: - outContestRes := &htlcOutgoingContestResolver{ - htlcTimeoutResolver: htlcTimeoutResolver{}, - } - if err := outContestRes.Decode(resReader); err != nil { - return err - } - - res = outContestRes - case resolverIncomingContest: inContestRes := &htlcIncomingContestResolver{ htlcSuccessResolver: htlcSuccessResolver{}, diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 55a6e238168..58670bf2e0f 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -1155,7 +1155,7 @@ func (c *ChannelArbitrator) prepContractResolutions(htlcActions ChainActionMap, // If we can timeout the HTLC directly, then we'll create the // proper resolver to do so, who will then cancel the packet // backwards. - case HtlcTimeoutAction: + case HtlcTimeoutAction, HtlcOutgoingWatchAction: for _, htlc := range htlcs { htlcOp := wire.OutPoint{ Hash: commitHash, @@ -1212,36 +1212,6 @@ func (c *ChannelArbitrator) prepContractResolutions(htlcActions ChainActionMap, } htlcResolvers = append(htlcResolvers, resolver) } - - // Finally, if this is an outgoing HTLC we've sent, then we'll - // launch a resolver to watch for the pre-image (and settle - // backwards), or just timeout. - case HtlcOutgoingWatchAction: - for _, htlc := range htlcs { - htlcOp := wire.OutPoint{ - Hash: commitHash, - Index: uint32(htlc.OutputIndex), - } - - resolution, ok := outResolutionMap[htlcOp] - if !ok { - log.Errorf("ChannelArbitrator(%v) unable to find "+ - "outgoing resolution: %v", - c.cfg.ChanPoint, htlcOp) - continue - } - - resKit.Quit = make(chan struct{}) - resolver := &htlcOutgoingContestResolver{ - htlcTimeoutResolver{ - htlcResolution: resolution, - broadcastHeight: height, - htlcIndex: htlc.HtlcIndex, - ResolverKit: resKit, - }, - } - htlcResolvers = append(htlcResolvers, resolver) - } } } diff --git a/contractcourt/contract_resolvers.go b/contractcourt/contract_resolvers.go index 7b8be486663..da971c2eaf1 100644 --- a/contractcourt/contract_resolvers.go +++ b/contractcourt/contract_resolvers.go @@ -112,6 +112,8 @@ type htlcTimeoutResolver struct { htlcIndex uint64 ResolverKit + + sweeper Sweeper } // ResolverKey returns an identifier which should be globally unique for this @@ -134,140 +136,272 @@ func (h *htlcTimeoutResolver) ResolverKey() []byte { return key[:] } -// Resolve kicks off full resolution of an outgoing HTLC output. If it's our -// commitment, it isn't resolved until we see the second level HTLC txn -// confirmed. If it's the remote party's commitment, we don't resolve until we -// see a direct sweep via the timeout clause. +type outgoingState uint8 + +const ( + cltvWait uint8 = iota + secondLevelConf + csvWait + sweepWait +) + +// Resolve commences the resolution of this contract. As this contract hasn't +// yet timed out, we'll wait for one of two things to happen // -// NOTE: Part of the ContractResolver interface. +// 1. The HTLC expires. In this case, we'll sweep the funds and send a clean +// up cancel message to outside sub-systems. +// +// 2. The remote party sweeps this HTLC on-chain, in which case we'll add the +// pre-image to our global cache, then send a clean up settle message +// backwards. +// +// When either of these two things happens, we'll create a new resolver which +// is able to handle the final resolution of the contract. We're only the pivot +// point. func (h *htlcTimeoutResolver) Resolve() (ContractResolver, error) { - // If we're already resolved, then we can exit early. + // If we're already full resolved, then we don't have anything further + // to do. if h.resolved { return nil, nil } - // If we haven't already sent the output to the utxo nursery, then - // we'll do so now. - if !h.outputIncubating { - log.Tracef("%T(%v): incubating htlc output", h, - h.htlcResolution.ClaimOutpoint) + // claimCleanUp is a helper function that's called once the HTLC output + // is spent by the remote party. It'll extract the preimage, add it to + // the global cache, and finally send the appropriate clean up message. + claimCleanUp := func(commitSpend *chainntnfs.SpendDetail) (ContractResolver, error) { + // Depending on if this is our commitment or not, then we'll be + // looking for a different witness pattern. + spenderIndex := commitSpend.SpenderInputIndex + spendingInput := commitSpend.SpendingTx.TxIn[spenderIndex] - err := h.IncubateOutputs(h.ChanPoint, nil, &h.htlcResolution, nil) - if err != nil { - return nil, err + log.Infof("%T(%v): extracting preimage! remote party spent "+ + "HTLC with tx=%v", h, h.htlcResolution.ClaimOutpoint, + spew.Sdump(commitSpend.SpendingTx)) + + // If this is the remote party's commitment, then we'll be + // looking for them to spend using the second-level success + // transaction. + var preimage [32]byte + if h.htlcResolution.SignedTimeoutTx == nil { + // The witness stack when the remote party sweeps the + // output to them looks like: + // + // * + copy(preimage[:], spendingInput.Witness[3]) + } else { + // Otherwise, they'll be spending directly from our + // commitment output. In which case the witness stack + // looks like: + // + // * + copy(preimage[:], spendingInput.Witness[1]) } - h.outputIncubating = true + log.Infof("%T(%v): extracting preimage=%x from on-chain "+ + "spend!", h, h.htlcResolution.ClaimOutpoint, preimage[:]) - if err := h.Checkpoint(h); err != nil { - log.Errorf("unable to Checkpoint: %v", err) + // With the preimage obtained, we can now add it to the global + // cache. + if err := h.PreimageDB.AddPreimage(preimage[:]); err != nil { + log.Errorf("%T(%v): unable to add witness to cache", + h, h.htlcResolution.ClaimOutpoint) + } + + // Finally, we'll send the clean up message, mark ourselves as + // resolved, then exit. + if err := h.DeliverResolutionMsg(ResolutionMsg{ + SourceChan: h.ShortChanID, + HtlcIndex: h.htlcIndex, + PreImage: &preimage, + }); err != nil { + return nil, err } + h.resolved = true + return nil, h.Checkpoint(h) } - // waitForOutputResolution waits for the HTLC output to be fully - // resolved. The output is considered fully resolved once it has been - // spent, and the spending transaction has been fully confirmed. - waitForOutputResolution := func() error { - // We first need to register to see when the HTLC output itself - // has been spent by a confirmed transaction. - spendNtfn, err := h.Notifier.RegisterSpendNtfn( - &h.htlcResolution.ClaimOutpoint, - h.htlcResolution.SweepSignDesc.Output.PkScript, - h.broadcastHeight, + // Otherwise, we'll watch for two external signals to decide if we'll + // morph into another resolver, or fully resolve the contract. + + // The output we'll be watching for is the *direct* spend from the HTLC + // output. If this isn't our commitment transaction, it'll be right on + // the resolution. Otherwise, we fetch this pointer from the input of + // the time out transaction. + var ( + outPointToWatch wire.OutPoint + scriptToWatch []byte + err error + firstLevelSpendChan <-chan *chainntnfs.SpendDetail + ) + if h.htlcResolution.SignedTimeoutTx == nil { + // Create dummy channel that will never receive a notification. + firstLevelSpendChan = make(chan *chainntnfs.SpendDetail) + } else { + // If this is the remote party's commitment, then we'll need to + // grab watch the output that our timeout transaction points + // to. We can directly grab the outpoint, then also extract the + // witness script (the last element of the witness stack) to + // re-construct the pkScipt we need to watch. + outPointToWatch = h.htlcResolution.SignedTimeoutTx.TxIn[0].PreviousOutPoint + witness := h.htlcResolution.SignedTimeoutTx.TxIn[0].Witness + scriptToWatch, err = lnwallet.WitnessScriptHash( + witness[len(witness)-1], ) if err != nil { - return err + return nil, err } - select { - case _, ok := <-spendNtfn.Spend: - if !ok { - return fmt.Errorf("notifier quit") - } - - case <-h.Quit: - return fmt.Errorf("quitting") + secondLevelSpendNtfn, err := h.Notifier.RegisterSpendNtfn( + &outPointToWatch, scriptToWatch, h.broadcastHeight, + ) + if err != nil { + return nil, err } + firstLevelSpendChan = secondLevelSpendNtfn.Spend + } - return nil + // TODO: Cancel spend notifications on quit? + + _, currentHeight, err := h.ChainIO.GetBestBlock() + if err != nil { + return nil, err } - // With the output sent to the nursery, we'll now wait until the output - // has been fully resolved before sending the clean up message. - // - // TODO(roasbeef): need to be able to cancel nursery? - // * if they pull on-chain while we're waiting + state := cltvWait + var secondLevelExpiry int32 + sweepDoneChan := make(chan struct{}) - // If we don't have a second layer transaction, then this is a remote - // party's commitment, so we'll watch for a direct spend. - if h.htlcResolution.SignedTimeoutTx == nil { - // We'll block until: the HTLC output has been spent, and the - // transaction spending that output is sufficiently confirmed. - log.Infof("%T(%v): waiting for nursery to spend CLTV-locked "+ - "output", h, h.htlcResolution.ClaimOutpoint) - if err := waitForOutputResolution(); err != nil { - return nil, err + evaluateSweep := func(sweeperCall SweeperCall) bool { + height := sweeperCall.TargetHeight + + switch { + case state == cltvWait: + // Nothing to sweep if not expired or second level + // required. + if h.htlcResolution.SignedTimeoutTx != nil { + return false + } + + if uint32(height) < h.htlcResolution.Expiry-1 { + return false + } + case state == csvWait: + // Nothing to sweep if second level not expired yet. + if height < secondLevelExpiry-1 { + return false + } + default: + return false } - } else { - // Otherwise, this is our commitment, so we'll watch for the - // second-level transaction to be sufficiently confirmed. - secondLevelTXID := h.htlcResolution.SignedTimeoutTx.TxHash() - sweepScript := h.htlcResolution.SignedTimeoutTx.TxOut[0].PkScript - confNtfn, err := h.Notifier.RegisterConfirmationsNtfn( - &secondLevelTXID, sweepScript, 1, h.broadcastHeight, - ) - if err != nil { - return nil, err + + // Signal sweeper that this input can be picked + // up now. + sweeperCall.InputChan <- SweepInput{ + OutPoint: h.htlcResolution.ClaimOutpoint, + ResultChan: sweepDoneChan, } + close(sweeperCall.InputChan) + + state = sweepWait + + return true + } - log.Infof("%T(%v): waiting second-level tx (txid=%v) to be "+ - "fully confirmed", h, h.htlcResolution.ClaimOutpoint, - secondLevelTXID) + // defer sweeperCallChan.Close() ? + sweeperCallChan := h.sweeper.RegisterForCalls() + blockEpochs, err := h.Notifier.RegisterBlockEpochNtfn(nil) + if err != nil { + return nil, err + } + defer blockEpochs.Cancel() + + // Early publish of timeout tx + +loop: + for { select { - case _, ok := <-confNtfn.Confirmed: + + // A new block has arrived, we'll check to see if this leads to + // HTLC expiration and the need to publish the timeout tx. + case newBlock, ok := <-blockEpochs.Epochs: if !ok { return nil, fmt.Errorf("quitting") } - case <-h.Quit: - return nil, fmt.Errorf("quitting") - } - } + if state == cltvWait && + uint32(newBlock.Height) >= h.htlcResolution.Expiry-1 && + h.htlcResolution.SignedTimeoutTx != nil { - // TODO(roasbeef): need to watch for remote party sweeping with pre-image? - // * have another waiting on spend above, will check the type, if it's - // pre-image, then we'll cancel, and send a clean up back with - // pre-image, also add to preimage cache + // TODO: Publish timeout tx - log.Infof("%T(%v): resolving htlc with incoming fail msg, fully "+ - "confirmed", h, h.htlcResolution.ClaimOutpoint) + state = secondLevelConf + } - // At this point, the second-level transaction is sufficiently - // confirmed, or a transaction directly spending the output is. - // Therefore, we can now send back our clean up message. - failureMsg := &lnwire.FailPermanentChannelFailure{} - if err := h.DeliverResolutionMsg(ResolutionMsg{ - SourceChan: h.ShortChanID, - HtlcIndex: h.htlcIndex, - Failure: failureMsg, - }); err != nil { - return nil, err - } + // Sweeper is constructing a new sweep. Evaluate if we have + // anything to add. + case newBlock, ok := <-sweeperCallChan: + if !ok { + return nil, fmt.Errorf("quitting") + } - // Finally, if this was an output on our commitment transaction, we'll - // for the second-level HTLC output to be spent, and for that - // transaction itself to confirm. - if h.htlcResolution.SignedTimeoutTx != nil { - log.Infof("%T(%v): waiting for nursery to spend CSV delayed "+ - "output", h, h.htlcResolution.ClaimOutpoint) - if err := waitForOutputResolution(); err != nil { - return nil, err + // If this new height expires the HTLC, then we can + // exit early and create a resolver that's capable of + // handling the time locked output. + evaluateSweep(newBlock) + + // The output has been spent! This means the preimage has been + // revealed on-chain or we used the output ourselves. + case commitSpend, ok := <-firstLevelSpendChan: + if !ok { + return nil, fmt.Errorf("quitting") + } + + // The only way this output can be spent by the remote + // party is by revealing the preimage. So we'll perform + // our duties to clean up the contract once it has been + // claimed. + timeoutTxHash := h.htlcResolution.SignedTimeoutTx.TxHash() + + // If spend not by our own timeout tx, it must + // be a remote spend. + isRemoteSpend := !bytes.Equal(commitSpend.SpenderTxHash[:], + timeoutTxHash[:]) + + if isRemoteSpend { + return claimCleanUp(commitSpend) + } + + // At this point, the second-level transaction is + // sufficiently confirmed, or a transaction directly + // spending the output is. Therefore, we can now send + // back our clean up message. + failureMsg := &lnwire.FailPermanentChannelFailure{} + if err := h.DeliverResolutionMsg(ResolutionMsg{ + SourceChan: h.ShortChanID, + HtlcIndex: h.htlcIndex, + Failure: failureMsg, + }); err != nil { + return nil, err + } + + state = csvWait + secondLevelExpiry = currentHeight + + int32(h.htlcResolution.CsvDelay) + case <-sweepDoneChan: + // Check for remote spend + + // Detect remote success tx by witness length. + // isRemoteSpend = len(commitSpend.SpendingTx.TxIn[0].Witness) == 5 + + // Deliver failure message + + break loop + case <-h.Quit: + return nil, fmt.Errorf("resolver cancelled") } - } - // With the clean up message sent, we'll now mark the contract - // resolved, and wait. + } h.resolved = true return nil, h.Checkpoint(h) } @@ -693,265 +827,6 @@ func (h *htlcSuccessResolver) AttachResolverKit(r ResolverKit) { // ContractResolver interface. var _ ContractResolver = (*htlcSuccessResolver)(nil) -// htlcOutgoingContestResolver is a ContractResolver that's able to resolve an -// outgoing HTLC that is still contested. An HTLC is still contested, if at the -// time that we broadcast the commitment transaction, it isn't able to be fully -// resolved. In this case, we'll either wait for the HTLC to timeout, or for -// us to learn of the preimage. -type htlcOutgoingContestResolver struct { - // htlcTimeoutResolver is the inner solver that this resolver may turn - // into. This only happens if the HTLC expires on-chain. - htlcTimeoutResolver -} - -// Resolve commences the resolution of this contract. As this contract hasn't -// yet timed out, we'll wait for one of two things to happen -// -// 1. The HTLC expires. In this case, we'll sweep the funds and send a clean -// up cancel message to outside sub-systems. -// -// 2. The remote party sweeps this HTLC on-chain, in which case we'll add the -// pre-image to our global cache, then send a clean up settle message -// backwards. -// -// When either of these two things happens, we'll create a new resolver which -// is able to handle the final resolution of the contract. We're only the pivot -// point. -func (h *htlcOutgoingContestResolver) Resolve() (ContractResolver, error) { - // If we're already full resolved, then we don't have anything further - // to do. - if h.resolved { - return nil, nil - } - - // claimCleanUp is a helper function that's called once the HTLC output - // is spent by the remote party. It'll extract the preimage, add it to - // the global cache, and finally send the appropriate clean up message. - claimCleanUp := func(commitSpend *chainntnfs.SpendDetail) (ContractResolver, error) { - // Depending on if this is our commitment or not, then we'll be - // looking for a different witness pattern. - spenderIndex := commitSpend.SpenderInputIndex - spendingInput := commitSpend.SpendingTx.TxIn[spenderIndex] - - log.Infof("%T(%v): extracting preimage! remote party spent "+ - "HTLC with tx=%v", h, h.htlcResolution.ClaimOutpoint, - spew.Sdump(commitSpend.SpendingTx)) - - // If this is the remote party's commitment, then we'll be - // looking for them to spend using the second-level success - // transaction. - var preimage [32]byte - if h.htlcResolution.SignedTimeoutTx == nil { - // The witness stack when the remote party sweeps the - // output to them looks like: - // - // * - copy(preimage[:], spendingInput.Witness[3]) - } else { - // Otherwise, they'll be spending directly from our - // commitment output. In which case the witness stack - // looks like: - // - // * - copy(preimage[:], spendingInput.Witness[1]) - } - - log.Infof("%T(%v): extracting preimage=%x from on-chain "+ - "spend!", h, h.htlcResolution.ClaimOutpoint, preimage[:]) - - // With the preimage obtained, we can now add it to the global - // cache. - if err := h.PreimageDB.AddPreimage(preimage[:]); err != nil { - log.Errorf("%T(%v): unable to add witness to cache", - h, h.htlcResolution.ClaimOutpoint) - } - - // Finally, we'll send the clean up message, mark ourselves as - // resolved, then exit. - if err := h.DeliverResolutionMsg(ResolutionMsg{ - SourceChan: h.ShortChanID, - HtlcIndex: h.htlcIndex, - PreImage: &preimage, - }); err != nil { - return nil, err - } - h.resolved = true - return nil, h.Checkpoint(h) - } - - // Otherwise, we'll watch for two external signals to decide if we'll - // morph into another resolver, or fully resolve the contract. - - // The output we'll be watching for is the *direct* spend from the HTLC - // output. If this isn't our commitment transaction, it'll be right on - // the resolution. Otherwise, we fetch this pointer from the input of - // the time out transaction. - var ( - outPointToWatch wire.OutPoint - scriptToWatch []byte - err error - ) - if h.htlcResolution.SignedTimeoutTx == nil { - outPointToWatch = h.htlcResolution.ClaimOutpoint - scriptToWatch = h.htlcResolution.SweepSignDesc.Output.PkScript - } else { - // If this is the remote party's commitment, then we'll need to - // grab watch the output that our timeout transaction points - // to. We can directly grab the outpoint, then also extract the - // witness script (the last element of the witness stack) to - // re-construct the pkScipt we need to watch. - outPointToWatch = h.htlcResolution.SignedTimeoutTx.TxIn[0].PreviousOutPoint - witness := h.htlcResolution.SignedTimeoutTx.TxIn[0].Witness - scriptToWatch, err = lnwallet.WitnessScriptHash( - witness[len(witness)-1], - ) - if err != nil { - return nil, err - } - } - - // First, we'll register for a spend notification for this output. If - // the remote party sweeps with the pre-image, we'll be notified. - spendNtfn, err := h.Notifier.RegisterSpendNtfn( - &outPointToWatch, scriptToWatch, h.broadcastHeight, - ) - if err != nil { - return nil, err - } - - // We'll quickly check to see if the output has already been spent. - select { - // If the output has already been spent, then we can stop early and - // sweep the pre-image from the output. - case commitSpend, ok := <-spendNtfn.Spend: - if !ok { - return nil, fmt.Errorf("quitting") - } - - // TODO(roasbeef): Checkpoint? - return claimCleanUp(commitSpend) - - // If it hasn't, then we'll watch for both the expiration, and the - // sweeping out this output. - default: - } - - // We'll check the current height, if the HTLC has already expired, - // then we'll morph immediately into a resolver that can sweep the - // HTLC. - // - // TODO(roasbeef): use grace period instead? - _, currentHeight, err := h.ChainIO.GetBestBlock() - if err != nil { - return nil, err - } - - // If the current height is >= expiry-1, then a spend will be valid to - // be included in the next block, and we can immediately return the - // resolver. - if uint32(currentHeight) >= h.htlcResolution.Expiry-1 { - log.Infof("%T(%v): HTLC has expired (height=%v, expiry=%v), "+ - "transforming into timeout resolver", h, - h.htlcResolution.ClaimOutpoint) - return &h.htlcTimeoutResolver, nil - } - - // If we reach this point, then we can't fully act yet, so we'll await - // either of our signals triggering: the HTLC expires, or we learn of - // the preimage. - blockEpochs, err := h.Notifier.RegisterBlockEpochNtfn(nil) - if err != nil { - return nil, err - } - defer blockEpochs.Cancel() - - for { - select { - - // A new block has arrived, we'll check to see if this leads to - // HTLC expiration. - case newBlock, ok := <-blockEpochs.Epochs: - if !ok { - return nil, fmt.Errorf("quitting") - } - - // If this new height expires the HTLC, then we can - // exit early and create a resolver that's capable of - // handling the time locked output. - newHeight := uint32(newBlock.Height) - if newHeight >= h.htlcResolution.Expiry-1 { - log.Infof("%T(%v): HTLC has expired "+ - "(height=%v, expiry=%v), transforming "+ - "into timeout resolver", h, - h.htlcResolution.ClaimOutpoint, - newHeight, h.htlcResolution.Expiry) - return &h.htlcTimeoutResolver, nil - } - - // The output has been spent! This means the preimage has been - // revealed on-chain. - case commitSpend, ok := <-spendNtfn.Spend: - if !ok { - return nil, fmt.Errorf("quitting") - } - - // The only way this output can be spent by the remote - // party is by revealing the preimage. So we'll perform - // our duties to clean up the contract once it has been - // claimed. - return claimCleanUp(commitSpend) - - case <-h.Quit: - return nil, fmt.Errorf("resolver cancelled") - } - } -} - -// Stop signals the resolver to cancel any current resolution processes, and -// suspend. -// -// NOTE: Part of the ContractResolver interface. -func (h *htlcOutgoingContestResolver) Stop() { - close(h.Quit) -} - -// IsResolved returns true if the stored state in the resolve is fully -// resolved. In this case the target output can be forgotten. -// -// NOTE: Part of the ContractResolver interface. -func (h *htlcOutgoingContestResolver) IsResolved() bool { - return h.resolved -} - -// Encode writes an encoded version of the ContractResolver into the passed -// Writer. -// -// NOTE: Part of the ContractResolver interface. -func (h *htlcOutgoingContestResolver) Encode(w io.Writer) error { - return h.htlcTimeoutResolver.Encode(w) -} - -// Decode attempts to decode an encoded ContractResolver from the passed Reader -// instance, returning an active ContractResolver instance. -// -// NOTE: Part of the ContractResolver interface. -func (h *htlcOutgoingContestResolver) Decode(r io.Reader) error { - return h.htlcTimeoutResolver.Decode(r) -} - -// AttachResolverKit should be called once a resolved is successfully decoded -// from its stored format. This struct delivers a generic tool kit that -// resolvers need to complete their duty. -// -// NOTE: Part of the ContractResolver interface. -func (h *htlcOutgoingContestResolver) AttachResolverKit(r ResolverKit) { - h.ResolverKit = r -} - -// A compile time assertion to ensure htlcOutgoingContestResolver meets the -// ContractResolver interface. -var _ ContractResolver = (*htlcOutgoingContestResolver)(nil) - // htlcIncomingContestResolver is a ContractResolver that's able to resolve an // incoming HTLC that is still contested. An HTLC is still contested, if at the // time of commitment broadcast, we don't know of the preimage for it yet, and diff --git a/contractcourt/sweeper.go b/contractcourt/sweeper.go new file mode 100644 index 00000000000..c67578add02 --- /dev/null +++ b/contractcourt/sweeper.go @@ -0,0 +1,105 @@ +package contractcourt + +import ( + "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/chainntnfs" +) + +// Sweeper is responsible for sweeping outputs back into the wallet +type Sweeper struct { + Notifier chainntnfs.ChainNotifier + quit chan struct{} +} + +// Start starts the process of constructing and publish sweep txes. +func (s *Sweeper) Start() error { + s.quit = make(chan struct{}) + + newBlockChan, err := s.Notifier.RegisterBlockEpochNtfn(nil) + if err != nil { + return err + } + + go s.collector(newBlockChan) + + return nil +} + +// SweepInput is what is pushed to the sweeper to describe the input to sweep. +type SweepInput struct { + OutPoint wire.OutPoint + ResultChan chan struct{} +} + +// SweeperCall is push in the channel to collect sweep inputs. +type SweeperCall struct { + InputChan chan SweepInput + TargetHeight int32 +} + +// RegisterForCalls returns a channel on which the caller can listen for sweeper +// events. +func (s *Sweeper) RegisterForCalls() chan SweeperCall { + + return make(chan SweeperCall) +} + +func (s *Sweeper) collector(newBlockChan *chainntnfs.BlockEpochEvent) { + defer newBlockChan.Cancel() + + for { + select { + case _, ok := <-newBlockChan.Epochs: + // If the epoch channel has been closed, then the + // ChainNotifier is exiting which means the daemon is + // as well. Therefore, we exit early also in order to + // ensure the daemon shuts down gracefully, yet + // swiftly. + if !ok { + return + } + + // Fetch all input channel from priority queue that have + // min target height <= epoch + 1 + + // Read outpoints from input channels. If channel is + // closed, ignore. This means the sweep for that output + // has been cancelled. + + // Construct sweep tx with the outputs. + + // Publish sweep tx. + + // Process publish error (double spend) -> ? + + // Remove from priority queue + + // Log in database outputs that are currently part of an + // unconfirmed sweep tx. This is to prevent republishing + // in a new sweep tx after restart. + + case <-s.quit: + return + } + } + +} + +// AnnounceSweep schedules a sweep for a specific minimum target height. The +// return value is a channel that is expected to provide the outpoint to sweep +// by the time the block height is high enough. The reason to use a blocking +// channel is that we want to make sure we collected all inputs before +// constructing the sweep tx. Otherwise, if block events would reach sweeper and +// the providers of sweep inputs (like resolvers) in the wrong order, inputs +// would be unnecessarily delayed until the next block. +// +// TODO: Maybe we can already pass in the outpoint at this moment? +func (s *Sweeper) AnnounceSweep(minTargetHeight int32) chan wire.OutPoint { + inputChan := make(chan wire.OutPoint) + + // Insert channel into priority queue keyed with minTargetHeight + + // TODO: publish sweep immediately sometimes? + + return inputChan +} diff --git a/nursery_store.go b/nursery_store.go index 4b6edbc0189..14a75a36490 100644 --- a/nursery_store.go +++ b/nursery_store.go @@ -109,6 +109,12 @@ type NurseryStore interface { // kidOutput's CSV delay. CribToKinder(*babyOutput) error + // CribToRemoteSpend marks an output as spend by the remote party. + CribToRemoteSpend(*babyOutput) error + + // KidToRemoteSpend marks an output as spend by the remote party. + KidToRemoteSpend(*kidOutput) error + // PreschoolToKinder atomically moves a kidOutput from the preschool // bucket to the kindergarten bucket. This transition should be // executed after receiving confirmation of the preschool output. @@ -128,6 +134,14 @@ type NurseryStore interface { // the preschool bucket. FetchPreschools() ([]kidOutput, error) + // FetchKinder returns a list of all outputs currently stored in + // the kindergarten bucket. + FetchKinder() ([]kidOutput, error) + + // FetchCribs returns a list of all outputs currently stored in the + // cribs bucket. + FetchCribs() ([]babyOutput, error) + // FetchClass returns a list of kindergarten and crib outputs whose // timelocks expire at the given height. If the kindergarten class at // this height hash been finalized previously, via FinalizeKinder, it @@ -231,6 +245,8 @@ var ( // this serves as a persistent marker that the nursery should mark the // channel fully closed in the channeldb. gradPrefix = []byte("grad") + + spndPrefix = []byte("spnd") ) // prefixChainKey creates the root level keys for the nursery store. The keys @@ -412,6 +428,132 @@ func (ns *nurseryStore) CribToKinder(bby *babyOutput) error { }) } +func (ns *nurseryStore) CribToRemoteSpend(bby *babyOutput) error { + return ns.db.Update(func(tx *bolt.Tx) error { + + // First, retrieve or create the channel bucket corresponding to + // the baby output's origin channel point. + chanPoint := bby.OriginChanPoint() + chanBucket, err := ns.createChannelBucket(tx, chanPoint) + if err != nil { + return err + } + + // The babyOutput should currently be stored in the crib bucket. + // So, we create a key that prefixes the babyOutput's outpoint + // with the crib prefix, allowing us to reference it in the + // store. + pfxOutputKey, err := prefixOutputKey(cribPrefix, bby.OutPoint()) + if err != nil { + return err + } + + // Since the babyOutput is being moved to the spend bucket, we + // remove the entry from the channel bucket under the + // crib-prefixed outpoint key. + if err := chanBucket.Delete(pfxOutputKey); err != nil { + return err + } + + // Remove the crib output's entry in the height index. + err = ns.removeOutputFromHeight(tx, bby.expiry, chanPoint, + pfxOutputKey) + if err != nil { + return err + } + + // Since we are moving this output from the crib bucket to the + // spend bucket, we overwrite the existing prefix of this key + // with the spend prefix. + copy(pfxOutputKey, spndPrefix) + + // Now, serialize babyOutput's encapsulated kidOutput such that + // it can be written to the channel bucket under the new + // spend-prefixed key. + var kidBuffer bytes.Buffer + if err := bby.kidOutput.Encode(&kidBuffer); err != nil { + return err + } + kidBytes := kidBuffer.Bytes() + + // Persist the serialized kidOutput under the spend-prefixed + // outpoint key. + if err := chanBucket.Put(pfxOutputKey, kidBytes); err != nil { + return err + } + + utxnLog.Tracef("Transitioning (crib -> spnd) output for "+ + "chan_point=%v", chanPoint) + + return nil + }) +} + +func (ns *nurseryStore) KidToRemoteSpend(kid *kidOutput) error { + return ns.db.Update(func(tx *bolt.Tx) error { + + // First, retrieve or create the channel bucket corresponding to + // the baby output's origin channel point. + chanPoint := kid.OriginChanPoint() + chanBucket, err := ns.createChannelBucket(tx, chanPoint) + if err != nil { + return err + } + + // The kidOutput should currently be stored in the kndr bucket. + // So, we create a key that prefixes the kidOutput's outpoint + // with the crib prefix, allowing us to reference it in the + // store. + pfxOutputKey, err := prefixOutputKey(kndrPrefix, kid.OutPoint()) + if err != nil { + return err + } + + // Since the kidOutput is being moved to the spend bucket, we + // remove the entry from the channel bucket under the + // crib-prefixed outpoint key. + if err := chanBucket.Delete(pfxOutputKey); err != nil { + return err + } + + // Remove the crib output's entry in the height index. + // + // TODO(joostjager): Here we assume no late graduation into kid + // stage has happened! Probably ok, because channel_arb holds + // back until commit tx is confirmed. + err = ns.removeOutputFromHeight(tx, kid.absoluteMaturity, + chanPoint, pfxOutputKey) + if err != nil { + return err + } + + // Since we are moving this output from the crib bucket to the + // spend bucket, we overwrite the existing prefix of this key + // with the spend prefix. + copy(pfxOutputKey, spndPrefix) + + // Now, serialize babyOutput's encapsulated kidOutput such that + // it can be written to the channel bucket under the new + // spend-prefixed key. + var kidBuffer bytes.Buffer + if err := kid.Encode(&kidBuffer); err != nil { + return err + } + kidBytes := kidBuffer.Bytes() + + // Persist the serialized kidOutput under the spend-prefixed + // outpoint key. + if err := chanBucket.Put(pfxOutputKey, kidBytes); err != nil { + return err + } + + utxnLog.Tracef("Transitioning (kid -> spnd) output for "+ + "chan_point=%v", chanPoint) + + return nil + }) +} + // PreschoolToKinder atomically moves a kidOutput from the preschool bucket to // the kindergarten bucket. This transition should be executed after receiving // confirmation of the preschool output's commitment transaction. @@ -494,7 +636,7 @@ func (ns *nurseryStore) PreschoolToKinder(kid *kidOutput) error { maturityHeight = lastGradHeight + 1 } - utxnLog.Infof("Transitioning (crib -> kid) output for "+ + utxnLog.Infof("Transitioning (preschool -> kid) output for "+ "chan_point=%v at height_index=%v", chanPoint, maturityHeight) @@ -692,6 +834,16 @@ func (ns *nurseryStore) FetchClass( // FetchPreschools returns a list of all outputs currently stored in the // preschool bucket. func (ns *nurseryStore) FetchPreschools() ([]kidOutput, error) { + return ns.fetchKidsByPrefix(psclPrefix) +} + +// FetchKinder returns a list of all outputs currently stored in the +// kindergarten bucket. +func (ns *nurseryStore) FetchKinder() ([]kidOutput, error) { + return ns.fetchKidsByPrefix(kndrPrefix) +} + +func (ns *nurseryStore) fetchKidsByPrefix(prefix []byte) ([]kidOutput, error) { var kids []kidOutput if err := ns.db.View(func(tx *bolt.Tx) error { @@ -731,26 +883,26 @@ func (ns *nurseryStore) FetchPreschools() ([]kidOutput, error) { } // All of the outputs of interest will start with the - // "pscl" prefix. So, we will perform a prefix scan of + // specified prefix. So, we will perform a prefix scan of // the channel bucket to efficiently enumerate all the // desired outputs. c := chanBucket.Cursor() - for k, v := c.Seek(psclPrefix); bytes.HasPrefix( - k, psclPrefix); k, v = c.Next() { + for k, v := c.Seek(prefix); bytes.HasPrefix( + k, prefix); k, v = c.Next() { // Deserialize each output as a kidOutput, since // this should have been the type that was // serialized when it was written to disk. - var psclOutput kidOutput - psclReader := bytes.NewReader(v) - err := psclOutput.Decode(psclReader) + var kidOutput kidOutput + kidReader := bytes.NewReader(v) + err := kidOutput.Decode(kidReader) if err != nil { return err } // Add the deserialized output to our list of // preschool outputs. - kids = append(kids, psclOutput) + kids = append(kids, kidOutput) } } @@ -762,6 +914,79 @@ func (ns *nurseryStore) FetchPreschools() ([]kidOutput, error) { return kids, nil } +// FetchPreschools returns a list of all outputs currently stored in the +// preschool bucket. +func (ns *nurseryStore) FetchCribs() ([]babyOutput, error) { + var babies []babyOutput + if err := ns.db.View(func(tx *bolt.Tx) error { + + // Retrieve the existing chain bucket for this nursery store. + chainBucket := tx.Bucket(ns.pfxChainKey) + if chainBucket == nil { + return nil + } + + // Load the existing channel index from the chain bucket. + chanIndex := chainBucket.Bucket(channelIndexKey) + if chanIndex == nil { + return nil + } + + // Construct a list of all channels in the channel index that + // are currently being tracked by the nursery store. + var activeChannels [][]byte + if err := chanIndex.ForEach(func(chanBytes, _ []byte) error { + activeChannels = append(activeChannels, chanBytes) + return nil + }); err != nil { + return err + } + + // Iterate over all of the accumulated channels, and do a prefix + // scan inside of each channel bucket. Each output found that + // has a preschool prefix will be deserialized into a kidOutput, + // and added to our list of preschool outputs to return to the + // caller. + for _, chanBytes := range activeChannels { + // Retrieve the channel bucket associated with this + // channel. + chanBucket := chanIndex.Bucket(chanBytes) + if chanBucket == nil { + continue + } + + // All of the outputs of interest will start with the + // "pscl" prefix. So, we will perform a prefix scan of + // the channel bucket to efficiently enumerate all the + // desired outputs. + c := chanBucket.Cursor() + for k, v := c.Seek(cribPrefix); bytes.HasPrefix( + k, cribPrefix); k, v = c.Next() { + + // Deserialize each output as a kidOutput, since + // this should have been the type that was + // serialized when it was written to disk. + var cribOutput babyOutput + cribReader := bytes.NewReader(v) + err := cribOutput.Decode(cribReader) + if err != nil { + return err + } + + // Add the deserialized output to our list of + // preschool outputs. + babies = append(babies, cribOutput) + } + } + + return nil + }); err != nil { + return nil, err + } + + return babies, nil +} + // HeightsBelowOrEqual returns a slice of all non-empty heights in the height // index at or below the provided upper bound. func (ns *nurseryStore) HeightsBelowOrEqual(height uint32) ([]uint32, error) { @@ -1457,6 +1682,7 @@ func (ns *nurseryStore) getLastGraduatedHeight(tx *bolt.Tx) (uint32, error) { // the last graduated height key. func (ns *nurseryStore) putLastGraduatedHeight(tx *bolt.Tx, height uint32) error { + utxnLog.Infof("Log last graduated height at %v", height) // Ensure that the chain bucket for this nursery store exists. chainBucket, err := tx.CreateBucketIfNotExists(ns.pfxChainKey) if err != nil { diff --git a/server.go b/server.go index 35ef3689b16..73caacc9539 100644 --- a/server.go +++ b/server.go @@ -630,20 +630,33 @@ func newServer(listenAddrs []net.Addr, chanDB *channeldb.DB, cc *chainControl, outHtlcRes *lnwallet.OutgoingHtlcResolution, inHtlcRes *lnwallet.IncomingHtlcResolution) error { - var ( - inRes []lnwallet.IncomingHtlcResolution - outRes []lnwallet.OutgoingHtlcResolution - ) - if inHtlcRes != nil { - inRes = append(inRes, *inHtlcRes) + // TODO(joostjager): Propagate function split further + // up. To prevent adding three function typed fields in + // the config struct, replace by nursery typed field. + // But before this can be done, nursery needs to be put + // in its own package to fix dependencies. + if commitRes == nil && + outHtlcRes == nil && inHtlcRes != nil { + + return s.utxoNursery.IncubateIncomingHtlcOutput( + chanPoint, *inHtlcRes) } - if outHtlcRes != nil { - outRes = append(outRes, *outHtlcRes) + + if commitRes == nil && + outHtlcRes != nil && inHtlcRes == nil { + + return s.utxoNursery.IncubateOutgoingHtlcOutput( + chanPoint, *outHtlcRes) } - return s.utxoNursery.IncubateOutputs( - chanPoint, commitRes, outRes, inRes, - ) + if commitRes != nil && + outHtlcRes == nil && inHtlcRes == nil { + + return s.utxoNursery.IncubateCommitOutput( + chanPoint, commitRes) + } + + panic("invalid combination of inputs") }, PreimageDB: s.witnessBeacon, Notifier: cc.chainNotifier, diff --git a/utxonursery.go b/utxonursery.go index f3bf362ec0e..12740eeb7a8 100644 --- a/utxonursery.go +++ b/utxonursery.go @@ -298,7 +298,7 @@ func (u *utxoNursery) Start() error { // for the force closed commitment txn to confirm, or any second-layer // HTLC success transactions. // - // NOTE: The next two steps *may* spawn go routines, thus from this + // NOTE: The next four steps *may* spawn go routines, thus from this // point forward, we must close the nursery's quit channel if we detect // any failures during startup to ensure they terminate. if err := u.reloadPreschool(); err != nil { @@ -307,7 +307,23 @@ func (u *utxoNursery) Start() error { return err } - // 3. Replay all crib and kindergarten outputs from last pruned to + // 3. Restart spend ntfns for any kinder outputs, which are waiting + // for the CTLV to expire. + if err := u.reloadKinder(); err != nil { + newBlockChan.Cancel() + close(u.quit) + return err + } + + // 4. Restart spend ntfns for any crib outputs, which are waiting + // for the CTLV to expire. + if err := u.reloadCrib(); err != nil { + newBlockChan.Cancel() + close(u.quit) + return err + } + + // 5. Replay all crib and kindergarten outputs from last pruned to // current best height. if err := u.reloadClasses(lastGraduatedHeight); err != nil { newBlockChan.Cancel() @@ -333,112 +349,155 @@ func (u *utxoNursery) Stop() error { close(u.quit) u.wg.Wait() + utxnLog.Infof("UTXO nursery shut down finished") + return nil } -// IncubateOutputs sends a request to the utxoNursery to incubate a set of -// outputs from an existing commitment transaction. Outputs need to incubate if -// they're CLTV absolute time locked, or if they're CSV relative time locked. -// Once all outputs reach maturity, they'll be swept back into the wallet. -func (u *utxoNursery) IncubateOutputs(chanPoint wire.OutPoint, - commitResolution *lnwallet.CommitOutputResolution, - outgoingHtlcs []lnwallet.OutgoingHtlcResolution, - incomingHtlcs []lnwallet.IncomingHtlcResolution) error { +// IncubateCommitOutput sends a request to the utxoNursery to incubate the +// to-self output from an existing commitment transaction. The output needs to +// incubate it is CSV relative time locked. Once the output reaches maturity, +// it is swept back into the wallet. +func (u *utxoNursery) IncubateCommitOutput(chanPoint wire.OutPoint, + commitResolution *lnwallet.CommitOutputResolution) error { + + // Add to wait group because nursery might shut down during execution of + // this function. Otherwise it could happen that nursery thinks it is + // shut down, but in this function new goroutines were started and stay + // around. + u.wg.Add(1) + defer u.wg.Done() - numHtlcs := len(incomingHtlcs) + len(outgoingHtlcs) - var ( - hasCommit bool - - // Kid outputs can be swept after an initial confirmation - // followed by a maturity period.Baby outputs are two stage and - // will need to wait for an absolute time out to reach a - // confirmation, then require a relative confirmation delay. - kidOutputs = make([]kidOutput, 0, 1+len(incomingHtlcs)) - babyOutputs = make([]babyOutput, 0, len(outgoingHtlcs)) + selfOutput := makeKidOutput( + &commitResolution.SelfOutPoint, + &chanPoint, + commitResolution.MaturityDelay, + lnwallet.CommitmentTimeLock, + &commitResolution.SelfOutputSignDesc, + 0, ) - // 1. Build all the spendable outputs that we will try to incubate. - - // It could be that our to-self output was below the dust limit. In - // that case the commit resolution would be nil and we would not have - // that output to incubate. - if commitResolution != nil { - hasCommit = true - selfOutput := makeKidOutput( - &commitResolution.SelfOutPoint, - &chanPoint, - commitResolution.MaturityDelay, - lnwallet.CommitmentTimeLock, - &commitResolution.SelfOutputSignDesc, - 0, - ) + // We'll skip any zero valued outputs as this indicates we + // don't have a settled balance within the commitment + // transaction. + if selfOutput.Amount() == 0 { + return nil + } - // We'll skip any zero valued outputs as this indicates we - // don't have a settled balance within the commitment - // transaction. - if selfOutput.Amount() > 0 { - kidOutputs = append(kidOutputs, selfOutput) - } + utxnLog.Infof("Incubating Channel(%s) commit output") + + u.mu.Lock() + defer u.mu.Unlock() + + // Persist the output we intend to sweep in the nursery store + if err := u.cfg.Store.Incubate([]kidOutput{selfOutput}, + []babyOutput{}); err != nil { + + utxnLog.Errorf("unable to begin incubation of Channel(%s): %v", + chanPoint, err) + return err } - // TODO(roasbeef): query and see if we already have, if so don't add? + err := u.registerPreschoolConf(&selfOutput, u.bestHeight) + if err != nil { + return err + } - // For each incoming HTLC, we'll register a kid output marked as a - // second-layer HTLC output. We effectively skip the baby stage (as the - // timelock is zero), and enter the kid stage. - for _, htlcRes := range incomingHtlcs { - htlcOutput := makeKidOutput( - &htlcRes.ClaimOutpoint, &chanPoint, htlcRes.CsvDelay, - lnwallet.HtlcAcceptedSuccessSecondLevel, - &htlcRes.SweepSignDesc, 0, - ) + return nil +} - if htlcOutput.Amount() > 0 { - kidOutputs = append(kidOutputs, htlcOutput) - } +// IncubateOutgoingHtlcOutput sends a request to the utxoNursery to incubate a +// on outgoing htlc output from an existing commitment transaction. This +// function delegates incubation based on whether the commit tx was published by +// local or remote. +func (u *utxoNursery) IncubateOutgoingHtlcOutput(chanPoint wire.OutPoint, + outgoingHtlc lnwallet.OutgoingHtlcResolution) error { + + u.wg.Add(1) + defer u.wg.Done() + + if outgoingHtlc.SignedTimeoutTx != nil { + return u.incubateOutgoingHtlcOnLocal(chanPoint, outgoingHtlc) } - // For each outgoing HTLC, we'll create a baby output. If this is our - // commitment transaction, then we'll broadcast a second-layer - // transaction to transition to a kid output. Otherwise, we'll directly - // spend once the CLTV delay us up. - for _, htlcRes := range outgoingHtlcs { - // If this HTLC is on our commitment transaction, then it'll be - // a baby output as we need to go to the second level to sweep - // it. - if htlcRes.SignedTimeoutTx != nil { - htlcOutput := makeBabyOutput(&chanPoint, &htlcRes) + return u.incubateOutgoingHtlcOnRemote(chanPoint, outgoingHtlc) +} - if htlcOutput.Amount() > 0 { - babyOutputs = append(babyOutputs, htlcOutput) - } - continue - } +// incubateOutgoingHtlcOnRemote sends a request to the utxoNursery to incubate +// an outgoing htlc from an existing remote commitment transaction. The output +// needs to incubate because it is CLTV absolute time locked. Once the output +// reaches maturity, it is swept back into the wallet. +func (u *utxoNursery) incubateOutgoingHtlcOnRemote(chanPoint wire.OutPoint, + outgoingHtlc lnwallet.OutgoingHtlcResolution) error { - // Otherwise, this is actually a kid output as we can sweep it - // once the commitment transaction confirms, and the absolute - // CLTV lock has expired. We set the CSV delay to zero to - // indicate this is actually a CLTV output. - htlcOutput := makeKidOutput( - &htlcRes.ClaimOutpoint, &chanPoint, 0, - lnwallet.HtlcOfferedRemoteTimeout, - &htlcRes.SweepSignDesc, htlcRes.Expiry, - ) - kidOutputs = append(kidOutputs, htlcOutput) + // We set the CSV delay to zero to indicate this is actually a CLTV + // output. + htlcOutput := makeKidOutput( + &outgoingHtlc.ClaimOutpoint, &chanPoint, 0, + lnwallet.HtlcOfferedRemoteTimeout, + &outgoingHtlc.SweepSignDesc, outgoingHtlc.Expiry, + ) + + if htlcOutput.Amount() == 0 { + return nil } // TODO(roasbeef): if want to handle outgoing on remote commit // * need ability to cancel in the case that we learn of pre-image or // remote party pulls - utxnLog.Infof("Incubating Channel(%s) has-commit=%v, num-htlcs=%d", - chanPoint, hasCommit, numHtlcs) + utxnLog.Infof("Incubating Channel(%s) outgoing htlc on remote commit", + chanPoint) u.mu.Lock() defer u.mu.Unlock() - // 2. Persist the outputs we intended to sweep in the nursery store - if err := u.cfg.Store.Incubate(kidOutputs, babyOutputs); err != nil { + // Persist the output we intend to sweep in the nursery store + if err := u.cfg.Store.Incubate([]kidOutput{htlcOutput}, + []babyOutput{}); err != nil { + + utxnLog.Errorf("unable to begin incubation of Channel(%s): %v", + chanPoint, err) + return err + } + + err := u.registerPreschoolConf(&htlcOutput, u.bestHeight) + if err != nil { + return err + } + + return nil +} + +// IncubateOutputs sends a request to the utxoNursery to incubate an outgoing +// htlc from an existing local commitment transaction. The output needs to +// incubate because it is CLTV absolute time locked and the second level +// transaction is CSV time locked. Once the output reaches maturity, it is swept +// back into the wallet. +// +// NOTE: For this crib output, it is assumed that the commit tx is confirmed +// when incubateOutgoingHtlcOnLocal is called. +func (u *utxoNursery) incubateOutgoingHtlcOnLocal(chanPoint wire.OutPoint, + outgoingHtlc lnwallet.OutgoingHtlcResolution) error { + + // This is a HTLC on our commitment transaction,we need to go to the + // second level to sweep it. + htlcOutput := makeBabyOutput(&chanPoint, &outgoingHtlc) + + if htlcOutput.Amount() == 0 { + return nil + } + + utxnLog.Infof("Incubating Channel(%s) outgoing htlc on local commit", + chanPoint) + + u.mu.Lock() + defer u.mu.Unlock() + + // Persist the output we intend to sweep in the nursery store + if err := u.cfg.Store.Incubate([]kidOutput{}, + []babyOutput{htlcOutput}); err != nil { + utxnLog.Errorf("unable to begin incubation of Channel(%s): %v", chanPoint, err) return err @@ -447,33 +506,64 @@ func (u *utxoNursery) IncubateOutputs(chanPoint wire.OutPoint, // As an intermediate step, we'll now check to see if any of the baby // outputs has actually _already_ expired. This may be the case if // blocks were mined while we processed this message. - _, bestHeight, err := u.cfg.ChainIO.GetBestBlock() + if uint32(u.bestHeight) >= htlcOutput.expiry { + err := u.sweepCribOutput(uint32(u.bestHeight), &htlcOutput) + if err != nil { + return err + } + } + + // Start watch this output for remote spends. + err := u.registerCribSpend(&htlcOutput, uint32(u.bestHeight)) if err != nil { return err } - // We'll examine all the baby outputs just inserted into the database, - // if the output has already expired, then we'll *immediately* sweep - // it. This may happen if the caller raced a block to call this method. - for _, babyOutput := range babyOutputs { - if uint32(bestHeight) >= babyOutput.expiry { - err = u.sweepCribOutput(uint32(bestHeight), &babyOutput) - if err != nil { - return err - } - } + return nil +} + +// IncubateOutputs sends a request to the utxoNursery to incubate an incoming +// htlc output from an existing commitment transaction. The outputs needs to +// incubate if it is CSV relative time locked. Once the output reaches +// maturity, it is swept back into the wallet. +func (u *utxoNursery) IncubateIncomingHtlcOutput(chanPoint wire.OutPoint, + incomingHtlcs lnwallet.IncomingHtlcResolution) error { + + u.wg.Add(1) + defer u.wg.Done() + + // TODO(roasbeef): query and see if we already have, if so don't add? + + // For each incoming HTLC, we'll register a kid output marked as a + // second-layer HTLC output. We effectively skip the baby stage (as the + // timelock is zero), and enter the kid stage. + htlcOutput := makeKidOutput( + &incomingHtlcs.ClaimOutpoint, &chanPoint, incomingHtlcs.CsvDelay, + lnwallet.HtlcAcceptedSuccessSecondLevel, + &incomingHtlcs.SweepSignDesc, 0, + ) + + if htlcOutput.Amount() == 0 { + return nil } - // 3. If we are incubating any preschool outputs, register for a - // confirmation notification that will transition it to the - // kindergarten bucket. - if len(kidOutputs) != 0 { - for _, kidOutput := range kidOutputs { - err := u.registerPreschoolConf(&kidOutput, u.bestHeight) - if err != nil { - return err - } - } + utxnLog.Infof("Incubating Channel(%s) incoming htlc", chanPoint) + + u.mu.Lock() + defer u.mu.Unlock() + + // Persist the output we intend to sweep in the nursery store + if err := u.cfg.Store.Incubate([]kidOutput{htlcOutput}, + []babyOutput{}); err != nil { + + utxnLog.Errorf("unable to begin incubation of Channel(%s): %v", + chanPoint, err) + return err + } + + err := u.registerPreschoolConf(&htlcOutput, u.bestHeight) + if err != nil { + return err } return nil @@ -510,6 +600,19 @@ func (u *utxoNursery) NurseryReport( // Each crib output represents a stage one htlc, and // will contribute towards the limbo balance. report.AddLimboStage1TimeoutHtlc(&baby) + case bytes.HasPrefix(k, spndPrefix): + // Cribs outputs are the only kind currently stored as + // baby outputs. + var kid kidOutput + err := kid.Decode(bytes.NewReader(v)) + if err != nil { + return err + } + + // Remote spend outputs will be mentioned in the report, + // but do not contribute towards the limbo balance + // anymore. + report.AddLostHtlc(&kid) case bytes.HasPrefix(k, psclPrefix), bytes.HasPrefix(k, kndrPrefix), @@ -534,11 +637,18 @@ func (u *utxoNursery) NurseryReport( case lnwallet.CommitmentTimeLock: report.AddLimboCommitment(&kid) - // An HTLC output on our commitment transaction - // where the second-layer transaction hasn't - // yet confirmed. case lnwallet.HtlcAcceptedSuccessSecondLevel: + // An HTLC output on our commitment transaction + // where the second-layer transaction hasn't + // yet confirmed. report.AddLimboStage1SuccessHtlc(&kid) + + case lnwallet.HtlcOfferedRemoteTimeout: + // This is an HTLC output on the + // commitment transaction of the remote + // party. We are waiting for the CLTV + // timelock expire. + report.AddLimboDirectHtlc(&kid) } case bytes.HasPrefix(k, kndrPrefix): @@ -648,6 +758,56 @@ func (u *utxoNursery) reloadPreschool() error { return nil } +// reloadPreschool re-initializes the chain notifier with all of the outputs +// that had been saved to the "preschool" database bucket prior to shutdown. +func (u *utxoNursery) reloadKinder() error { + kndrOutputs, err := u.cfg.Store.FetchKinder() + if err != nil { + return err + } + + // For each of the preschool outputs stored in the nursery store, load + // its close summary from disk so that we can get an accurate height + // hint from which to start our range for spend notifications. + + for i := range kndrOutputs { + kid := &kndrOutputs[i] + + // TODO(joostjager): fix height hint + heightHint := uint32(0) + err = u.registerKinderSpend(kid, heightHint) + if err != nil { + return err + } + } + + return nil +} + +// reloadCrib re-initializes the chain notifier with all of the outputs +// that had been saved to the "crib" database bucket prior to shutdown. +func (u *utxoNursery) reloadCrib() error { + cribOutputs, err := u.cfg.Store.FetchCribs() + if err != nil { + return err + } + + // For each of the crib outputs stored in the nursery store, load + // its close summary from disk so that we can get an accurate height + // hint from which to start our range for spend notifications. + for i := range cribOutputs { + baby := &cribOutputs[i] + + // TODO(joostjager): set height hint after #1847 is merged. + err = u.registerCribSpend(baby, 0) + if err != nil { + return err + } + } + + return nil +} + // reloadClasses reinitializes any height-dependent state transitions for which // the utxonursery has not received confirmation, and replays the graduation of // all kindergarten and crib outputs for heights that have not been finalized. @@ -1308,6 +1468,157 @@ func (u *utxoNursery) waitForTimeoutConf(baby *babyOutput, "kindergarten", baby.OutPoint()) } +func (u *utxoNursery) registerCribSpend(baby *babyOutput, + heightHint uint32) error { + + outPointToWatch := baby.timeoutTx.TxIn[0].PreviousOutPoint + witness := baby.timeoutTx.TxIn[0].Witness + scriptToWatch, err := lnwallet.WitnessScriptHash( + witness[len(witness)-1], + ) + if err != nil { + return err + } + + // First, we'll register for a spend notification for this output. If + // the remote party sweeps with the pre-image, we'll be notified. + spendNtfn, err := u.cfg.Notifier.RegisterSpendNtfn( + &outPointToWatch, scriptToWatch, heightHint, + ) + if err != nil { + return err + } + + u.wg.Add(1) + go u.waitForCribSpend(baby, spendNtfn) + + return nil +} + +func (u *utxoNursery) waitForCribSpend(baby *babyOutput, + spendNtfn *chainntnfs.SpendEvent) { + + defer u.wg.Done() + + timeoutTxHash := baby.timeoutTx.TxHash() + + select { + case spendDetail, ok := <-spendNtfn.Spend: + if !ok { + utxnLog.Errorf("Notification chan "+ + "closed, can't monitor output %v", + baby.OutPoint()) + return + } + + // Check if it is our own timeout transaction that spends. If + // so, no action is needed. Output handling will continue when + // transaction sufficiently confirms. + if bytes.Equal(spendDetail.SpenderTxHash[:], timeoutTxHash[:]) { + utxnLog.Trace("Timeout tx spent output") + return + } + case <-u.quit: + spendNtfn.Cancel() + return + } + + // Remote party has spent the output. + utxnLog.Trace("Remote party spend detected!") + + err := u.cfg.Store.CribToRemoteSpend(baby) + if err != nil { + utxnLog.Errorf("Unable to move %v output "+ + "to spend bucket", baby.OutPoint()) + + // TODO(joostjager): error channel to signal incubator to quit? + return + } + + // TODO(joostjager): somehow notify the caller of IncubateOutput of the + // remote spend. In case of a resolver, it can proceed with extracting + // the pre-image and resolving the contract. +} + +func (u *utxoNursery) registerKinderSpend(kid *kidOutput, + heightHint uint32) error { + + // For CLTV locked direct outputs on the remote commit tx, register a + // spend notification. These output are still contested and can also be + // swept by the remote party. Other kid outputs don't need to be watched + // for spends. + if kid.WitnessType() != lnwallet.HtlcOfferedRemoteTimeout { + return nil + } + + outPointToWatch := kid.OutPoint() + scriptToWatch := kid.SignDesc().Output.PkScript + + // First, we'll register for a spend notification for this output. If + // the remote party sweeps with the pre-image, we'll be notified. + spendNtfn, err := u.cfg.Notifier.RegisterSpendNtfn( + outPointToWatch, scriptToWatch, heightHint, + ) + if err != nil { + return err + } + + u.wg.Add(1) + go u.waitForKidSpend(kid, spendNtfn) + + return nil +} + +func (u *utxoNursery) waitForKidSpend(kid *kidOutput, + spendNtfn *chainntnfs.SpendEvent) { + + defer u.wg.Done() + + select { + case spendDetail, ok := <-spendNtfn.Spend: + if !ok { + utxnLog.Errorf("Notification chan "+ + "closed, can't monitor output %v", + kid.OutPoint()) + return + } + + utxnLog.Tracef("Spend details: %v", spendDetail) + + // Recognize success tx based on script length. + isRemoteSuccessTx := + len(spendDetail.SpendingTx.TxIn[0].Witness) == 5 + + // If it is not the remote tx, it must be our own sweep tx. + // Output handling will continue when transaction sufficiently + // confirms. + + if !isRemoteSuccessTx { + utxnLog.Trace("Timeout tx spent output") + return + } + case <-u.quit: + spendNtfn.Cancel() + return + } + + // Remote party has spent the output. + utxnLog.Trace("Remote party spend detected!") + + err := u.cfg.Store.KidToRemoteSpend(kid) + if err != nil { + utxnLog.Errorf("Unable to move %v output "+ + "to spend bucket", kid.OutPoint()) + + // TODO(joostjager): error channel to signal incubator to quit? + return + } + + // TODO(joostjager): somehow notify the caller of IncubateOutput of the + // remote spend. In case of a resolver, it can proceed with extracting + // the pre-image and resolving the contract. +} + // registerPreschoolConf is responsible for subscribing to the confirmation of // a commitment transaction, or an htlc success transaction for an incoming // HTLC on our commitment transaction.. If successful, the provided preschool @@ -1390,6 +1701,17 @@ func (u *utxoNursery) waitForPreschoolConf(kid *kidOutput, outputType, err) return } + + // Register spend notification for this now confirmed kid output. We + // don't handle the case where it is spend before the commit tx is + // properly confirmed. This should not happen as channel_arbitrator only + // hands off to nursery after confirmation of the commit tx. + err = u.registerKinderSpend(kid, kid.ConfHeight()) + if err != nil { + utxnLog.Errorf("Unable to register spend notification for %v "+ + "kid output: %v", outputType, err) + return + } } // contractMaturityReport is a report that details the maturity progress of a @@ -1448,7 +1770,9 @@ type htlcMaturityReport struct { // stage indicates whether the htlc is in the CLTV-timeout stage (1) or // the CSV-delay stage (2). A stage 1 htlc's maturity height will be set // to its expiry height, while a stage 2 htlc's maturity height will be - // set to its confirmation height plus the maturity requirement. + // set to its confirmation height plus the maturity requirement. Stage 0 + // indicates that the htlc was swept by the remote party by either a + // revocation key or a payment pre-image. stage uint32 } @@ -1479,6 +1803,17 @@ func (c *contractMaturityReport) AddRecoveredCommitment(kid *kidOutput) { c.maturityHeight = kid.BlocksToMaturity() + kid.ConfHeight() } +// AddLostHtlc adds a lost commitment output to maturity +// report's htlcs. +func (c *contractMaturityReport) AddLostHtlc(kid *kidOutput) { + c.htlcs = append(c.htlcs, htlcMaturityReport{ + // TODO: check fields + outpoint: *kid.OutPoint(), + amount: kid.Amount(), + stage: 0, + }) +} + // AddLimboStage1TimeoutHtlc adds an htlc crib output to the maturity report's // htlcs, and contributes its amount to the limbo balance. func (c *contractMaturityReport) AddLimboStage1TimeoutHtlc(baby *babyOutput) { @@ -1496,7 +1831,7 @@ func (c *contractMaturityReport) AddLimboStage1TimeoutHtlc(baby *babyOutput) { // AddLimboDirectHtlc adds a direct HTLC on the commitment transaction of the // remote party to the maturity report. This a CLTV time-locked output that -// hasn't yet expired. +// has or hasn't expired yet. func (c *contractMaturityReport) AddLimboDirectHtlc(kid *kidOutput) { c.limboBalance += kid.Amount() @@ -1560,6 +1895,7 @@ func (c *contractMaturityReport) AddRecoveredHtlc(kid *kidOutput) { confHeight: kid.ConfHeight(), maturityRequirement: kid.BlocksToMaturity(), maturityHeight: kid.ConfHeight() + kid.BlocksToMaturity(), + // TODO(joostjager): set stage? }) } diff --git a/utxonursery_test.go b/utxonursery_test.go index adff6d3e3b0..da482e40588 100644 --- a/utxonursery_test.go +++ b/utxonursery_test.go @@ -5,14 +5,21 @@ package main import ( "bytes" "fmt" + "github.com/lightningnetwork/lnd/channeldb" + "io/ioutil" + "math" + "os" "reflect" "testing" + "time" "github.com/btcsuite/btcd/btcec" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/txscript" "github.com/btcsuite/btcd/wire" + "github.com/btcsuite/btclog" "github.com/btcsuite/btcutil" + "github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/lnwallet" ) @@ -379,3 +386,927 @@ func TestBabyOutputSerialization(t *testing.T) { } } + +var testChanPoint = wire.OutPoint{} + +type testWriter struct { +} + +func (w testWriter) Write(p []byte) (n int, err error) { + os.Stdout.Write(p) + + return len(p), nil +} + +func init() { + backendLog := btclog.NewBackend(testWriter{}) + utxnLog = backendLog.Logger("TEST") + utxnLog.SetLevel(btclog.LevelTrace) +} + +type nurseryTestContext struct { + nursery *utxoNursery + notifier *nurseryMockNotifier + publishChan chan wire.MsgTx + store *nurseryStoreInterceptor + restart func() bool + receiveTx func() wire.MsgTx + t *testing.T +} + +func createNurseryTestContext(t *testing.T, + checkStartStop func(func()) bool) *nurseryTestContext { + + // Create a temporary database and connect nurseryStore to it. The + // alternative, mocking nurseryStore, is not chosen because there is + // still considerable logic in the store. + + tempDirName, err := ioutil.TempDir("", "channeldb") + if err != nil { + t.Fatalf("unable to create temp dir: %v", err) + } + + cdb, err := channeldb.Open(tempDirName) + if err != nil { + t.Fatalf("unable to open channeldb: %v", err) + } + + store, err := newNurseryStore(&chainhash.Hash{}, cdb) + if err != nil { + t.Fatal(err) + } + + // Wrap the store in an inceptor to be able to wait for events in this + // test. + storeIntercepter := newNurseryStoreInterceptor(store) + + notifier := newNurseryMockNotifier() + + cfg := NurseryConfig{ + Notifier: notifier, + DB: cdb, + Store: storeIntercepter, + ChainIO: &mockChainIO{}, + GenSweepScript: func() ([]byte, error) { + return []byte{}, nil + }, + Estimator: &mockFeeEstimator{}, + Signer: &nurseryMockSigner{}, + } + + publishChan := make(chan wire.MsgTx, 1) + cfg.PublishTransaction = func(tx *wire.MsgTx) error { + utxnLog.Tracef("Publishing tx %v", tx.TxHash()) + publishChan <- *tx + return nil + } + + nursery := newUtxoNursery(&cfg) + nursery.Start() + + ctx := &nurseryTestContext{ + nursery: nursery, + notifier: notifier, + store: storeIntercepter, + publishChan: publishChan, + t: t, + } + + ctx.restart = func() bool { + return checkStartStop(func() { + ctx.nursery.Stop() + // Simulate lnd restart. + ctx.nursery = newUtxoNursery(ctx.nursery.cfg) + ctx.nursery.Start() + }) + } + + ctx.receiveTx = func() wire.MsgTx { + var tx wire.MsgTx + select { + case tx = <-ctx.publishChan: + return tx + case <-time.After(5 * time.Second): + t.Fatalf("tx not published") + } + return tx + } + + // Start with testing an immediate restart. + ctx.restart() + + return ctx +} + +func (ctx *nurseryTestContext) finish() { + // Add a final restart point in this state + ctx.restart() + + // We assume that when finish is called, nursery has finished all its + // goroutines. This implies that the waitgroup is empty. + signalChan := make(chan struct{}) + go func() { + ctx.nursery.wg.Wait() + close(signalChan) + }() + + // The only goroutine that is still expected to be running is + // incubator(). Simulate exit of this goroutine. + ctx.nursery.wg.Done() + + // We now expect the Wait to succeed. + select { + case <-signalChan: + case <-time.After(time.Second): + ctx.t.Fatalf("lingering goroutines detected after test is finished") + } + + // Restore waitgroup state to what it was before. + ctx.nursery.wg.Add(1) + + ctx.nursery.Stop() + + // We should have consumed and asserted all published transactions in + // our unit tests. + select { + case <-ctx.publishChan: + ctx.t.Fatalf("unexpected transactions published") + default: + } + + // Assert that the database is empty. All channels removed and height + // index cleared. + nurseryChannels, err := ctx.nursery.cfg.Store.ListChannels() + if err != nil { + ctx.t.Fatal(err) + } + if len(nurseryChannels) > 0 { + ctx.t.Fatalf("Expected all channels to be removed from store") + } + + activeHeights, err := ctx.nursery.cfg.Store.HeightsBelowOrEqual( + math.MaxUint32) + if err != nil { + ctx.t.Fatal(err) + } + if len(activeHeights) > 0 { + ctx.t.Fatalf("Expected height index to be empty") + } +} + +func createOutgoingRes(onLocalCommitment bool) *lnwallet.OutgoingHtlcResolution { + // Set up an outgoing htlc resolution to hand off to nursery. + closeTx := &wire.MsgTx{} + + htlcOp := wire.OutPoint{ + Hash: closeTx.TxHash(), + Index: 0, + } + + outgoingRes := lnwallet.OutgoingHtlcResolution{ + Expiry: 125, + SweepSignDesc: lnwallet.SignDescriptor{ + Output: &wire.TxOut{ + Value: 10000, + }, + }, + CsvDelay: 2, + } + + if onLocalCommitment { + timeoutTx := &wire.MsgTx{ + TxIn: []*wire.TxIn{ + { + PreviousOutPoint: htlcOp, + Witness: [][]byte{{}}, + }, + }, + TxOut: []*wire.TxOut{ + {}, + }, + } + + outgoingRes.SignedTimeoutTx = timeoutTx + } else { + outgoingRes.ClaimOutpoint = htlcOp + } + + return &outgoingRes +} + +func createCommitmentRes() *lnwallet.CommitOutputResolution { + // Set up a commitment output resolution to hand off to nursery. + commitRes := lnwallet.CommitOutputResolution{ + SelfOutPoint: wire.OutPoint{}, + SelfOutputSignDesc: lnwallet.SignDescriptor{ + Output: &wire.TxOut{ + Value: 10000, + }, + }, + MaturityDelay: 2, + } + + return &commitRes +} + +func incubateTestOutput(t *testing.T, nursery *utxoNursery, + onLocalCommitment bool) *lnwallet.OutgoingHtlcResolution { + + outgoingRes := createOutgoingRes(onLocalCommitment) + + // Hand off to nursery. + err := nursery.IncubateOutgoingHtlcOutput( + testChanPoint, + *outgoingRes, + ) + if err != nil { + t.Fatal(err) + } + + // IncubateOutputs is executing synchronously and we expect the output + // to immediately show up in the report. + expectedStage := uint32(2) + if onLocalCommitment { + expectedStage = 1 + } + assertNurseryReport(t, nursery, 1, expectedStage, 10000) + + return outgoingRes +} + +func assertNurseryReport(t *testing.T, nursery *utxoNursery, + expectedNofHtlcs int, expectedStage uint32, + expectedLimboBalance btcutil.Amount) { + report, err := nursery.NurseryReport(&testChanPoint) + if err != nil { + t.Fatal(err) + } + + if len(report.htlcs) != expectedNofHtlcs { + t.Fatalf("expected %v outputs to be reported, but report "+ + "contains %v", expectedNofHtlcs, len(report.htlcs)) + } + + if expectedNofHtlcs != 0 { + htlcReport := report.htlcs[0] + if htlcReport.stage != expectedStage { + t.Fatalf("expected htlc be advanced to stage %v, but it is "+ + "reported in stage %v", expectedStage, htlcReport.stage) + } + } + + if report.limboBalance != expectedLimboBalance { + t.Fatalf("expected limbo balance to be %v, but it is %v instead", + expectedLimboBalance, report.limboBalance) + } +} + +func assertNurseryReportUnavailable(t *testing.T, nursery *utxoNursery) { + _, err := nursery.NurseryReport(&testChanPoint) + if err != ErrContractNotFound { + t.Fatal("expected report to be unavailable") + } +} + +// testRestartLoop runs the specified test multiple times and in every run it +// will attempt to execute a restart action in a different location. This is to +// assert that the unit under test is recovering correctly from restarts. +func testRestartLoop(t *testing.T, test func(*testing.T, + func(func()) bool)) { + + // Start with running the test without any restarts (index zero) + restartIdx := 0 + + for { + currentStartStopIdx := 0 + + // checkStartStop is called at every point in the test where a + // restart should be exercised. When this function is called as + // many times as the current value of currentStartStopIdx, it + // will execute startStopFunc. + checkStartStop := func(startStopFunc func()) bool { + currentStartStopIdx++ + if restartIdx == currentStartStopIdx { + startStopFunc() + + return true + } + return false + } + + t.Run(fmt.Sprintf("restart_%v", restartIdx), + func(t *testing.T) { + test(t, checkStartStop) + }) + + // Exit the loop when all restart points have been tested. + if currentStartStopIdx == restartIdx { + return + } + restartIdx++ + } +} + +func TestNurserySuccessLocal(t *testing.T) { + testRestartLoop(t, testNurserySuccessLocal) +} + +func testNurserySuccessLocal(t *testing.T, + checkStartStop func(func()) bool) { + + ctx := createNurseryTestContext(t, checkStartStop) + + outgoingRes := incubateTestOutput(t, ctx.nursery, true) + + ctx.restart() + + // Notify arrival of block where HTLC CLTV expires. + ctx.notifier.epochChan <- &chainntnfs.BlockEpoch{ + Height: 125, + } + + // This should trigger nursery to publish the timeout tx. + ctx.receiveTx() + + if ctx.restart() { + // Restart should retrigger broadcast of timeout tx. + ctx.receiveTx() + } + + // Spend the previous outpoint of the timeout transaction. + // Nursery is listening for it. + timeoutTxHash := outgoingRes.SignedTimeoutTx.TxHash() + ctx.notifier.spendOutpoint( + &outgoingRes.SignedTimeoutTx.TxIn[0].PreviousOutPoint, + outgoingRes.SignedTimeoutTx) + + // Confirm the timeout tx. This should promote the HTLC to KNDR state. + ctx.notifier.confirmTx(&timeoutTxHash, 126) + + // Wait for output to be promoted in store to KNDR. + select { + case <-ctx.store.cribToKinderChan: + case <-time.After(5 * time.Second): + t.Fatalf("output not promoted to KNDR") + } + + ctx.restart() + + // Notify arrival of block where second level HTLC unlocks. + ctx.notifier.epochChan <- &chainntnfs.BlockEpoch{ + Height: 128, + } + + // Check final sweep into wallet. + testSweepHtlc(t, ctx) + + ctx.finish() +} + +func TestNurserySuccessRemote(t *testing.T) { + testRestartLoop(t, testNurserySuccessRemote) +} + +func testNurserySuccessRemote(t *testing.T, + checkStartStop func(func()) bool) { + + ctx := createNurseryTestContext(t, checkStartStop) + + outgoingRes := incubateTestOutput(t, ctx.nursery, false) + + // TODO(joostjager): for this restart to work, channel db needs to be + // mocked. Waiting for merge of #1847 to completely remove reading + // closed channel summary. + + // ctx.restart() + + // Notify confirmation of the commitment tx. Is only listened to when + // resolving remote commitment tx. + // + // TODO(joostjager): This is probably not correct? + ctx.notifier.confirmTx(&outgoingRes.ClaimOutpoint.Hash, 124) + + // Wait for output to be promoted from PSCL to KNDR. + select { + case <-ctx.store.preschoolToKinderChan: + case <-time.After(5 * time.Second): + t.Fatalf("output not promoted to KNDR") + } + + ctx.restart() + + // Notify arrival of block where HTLC CLTV expires. + ctx.notifier.epochChan <- &chainntnfs.BlockEpoch{ + Height: 125, + } + + // Check final sweep into wallet. + testSweepHtlc(t, ctx) + + ctx.finish() +} + +func TestNurseryCommitmentOutput(t *testing.T) { + testRestartLoop(t, testNurseryCommitmentOutput) +} + +func testNurseryCommitmentOutput(t *testing.T, + checkStartStop func(func()) bool) { + + ctx := createNurseryTestContext(t, checkStartStop) + + commitRes := createCommitmentRes() + + // Hand off to nursery. + err := ctx.nursery.IncubateCommitOutput( + testChanPoint, + commitRes, + ) + if err != nil { + t.Fatal(err) + } + + // Verify that commitment output is showing up in nursery report as + // limbo balance. + assertNurseryReport(t, ctx.nursery, 0, 0, 10000) + + // TODO(joostjager): for this restart to work, channel db needs to be + // mocked. Waiting for merge of #1847 to completely remove reading + // closed channel summary. + + // ctx.restart() + + // Notify confirmation of the commitment tx. + ctx.notifier.confirmTx(&commitRes.SelfOutPoint.Hash, 124) + + // Wait for output to be promoted from PSCL to KNDR. + select { + case <-ctx.store.preschoolToKinderChan: + case <-time.After(5 * time.Second): + t.Fatalf("output not promoted to KNDR") + } + + ctx.restart() + + // Notify arrival of block where commit output CSV expires. + ctx.notifier.epochChan <- &chainntnfs.BlockEpoch{ + Height: 126, + } + + // Check final sweep into wallet. + testSweep(t, ctx, func() { + // Check limbo balance after sweep publication + assertNurseryReport(t, ctx.nursery, 0, 0, 10000) + }) + + ctx.finish() +} + +func TestNurseryCommitmentOutputIdempotent(t *testing.T) { + testRestartLoop(t, testNurseryCommitmentOutputIdempotent) +} + +func testNurseryCommitmentOutputIdempotent(t *testing.T, + checkStartStop func(func()) bool) { + + ctx := createNurseryTestContext(t, checkStartStop) + + commitRes := createCommitmentRes() + + // Hand off to nursery. + err := ctx.nursery.IncubateCommitOutput( + testChanPoint, + commitRes, + ) + if err != nil { + t.Fatal(err) + } + + // Notify confirmation of the commitment tx. + ctx.notifier.confirmTx(&commitRes.SelfOutPoint.Hash, 124) + + // Wait for output to be promoted from PSCL to KNDR. + select { + case <-ctx.store.preschoolToKinderChan: + case <-time.After(5 * time.Second): + t.Fatalf("output not promoted to KNDR") + } + + remoteSpendTx := wire.MsgTx{ + TxIn: []*wire.TxIn{ + { + PreviousOutPoint: commitRes.SelfOutPoint, + }, + }, + } + + ctx.notifier.spendOutpoint( + &commitRes.SelfOutPoint, + &remoteSpendTx) + + // Wait for output to be promoted in store to GRAD. + select { + case <-ctx.store.graduateKinderChan: + case <-time.After(5 * time.Second): + t.Fatalf("output not graduated") + } + + ctx.restart() + + // As there only was one output to graduate, we expect the channel to be + // closed and no report available anymore. + assertNurseryReportUnavailable(t, ctx.nursery) + + ctx.finish() +} + +func testSweepHtlc(t *testing.T, ctx *nurseryTestContext) { + testSweep(t, ctx, func() { + // Verify stage in nursery report. HTLCs should now both still be in + // stage two. + assertNurseryReport(t, ctx.nursery, 1, 2, 10000) + }) +} + +func testSweep(t *testing.T, ctx *nurseryTestContext, + afterPublishAssert func()) { + // Wait for nursery to publish the sweep tx. + sweepTx := ctx.receiveTx() + + if ctx.restart() { + // Restart will trigger rebroadcast of sweep tx. + sweepTx = ctx.receiveTx() + } + + afterPublishAssert() + + sweepTxHash := sweepTx.TxHash() + + ctx.notifier.spendOutpoint( + &sweepTx.TxIn[0].PreviousOutPoint, + &sweepTx) + + // Confirm the sweep tx. + ctx.notifier.confirmTx(&sweepTxHash, 129) + + // Wait for output to be promoted in store to GRAD. + select { + case <-ctx.store.graduateKinderChan: + case <-time.After(5 * time.Second): + t.Fatalf("output not graduated") + } + + ctx.restart() + + // As there only was one output to graduate, we expect the channel to be + // closed and no report available anymore. + assertNurseryReportUnavailable(t, ctx.nursery) +} + +func TestNurseryRemoteSpendOnLocal(t *testing.T) { + testRestartLoop(t, testNurseryRemoteSpendOnLocal) +} + +func testNurseryRemoteSpendOnLocal(t *testing.T, + checkStartStop func(func()) bool) { + + ctx := createNurseryTestContext(t, checkStartStop) + + outgoingRes := incubateTestOutput(t, ctx.nursery, true) + + ctx.restart() + + // Have the remote party spend the commit tx outpoint. + remoteSpendTxPreviousOutpoint := + outgoingRes.SignedTimeoutTx.TxIn[0].PreviousOutPoint + + remoteSpendTx := wire.MsgTx{ + TxIn: []*wire.TxIn{ + { + PreviousOutPoint: remoteSpendTxPreviousOutpoint, + }, + }, + } + ctx.notifier.spendOutpoint( + &remoteSpendTxPreviousOutpoint, + &remoteSpendTx) + + // Wait for output to be marked as SPND. + select { + case <-ctx.store.cribToRemoteSpendChan: + case <-time.After(5 * time.Second): + t.Fatalf("not marked as SPND") + } + + // Notify arrival of block where HTLC expires. + ctx.notifier.epochChan <- &chainntnfs.BlockEpoch{ + Height: 125, + } + + // This should not trigger the nursery to publish the timeout tx. + select { + case <-ctx.publishChan: + t.Fatalf("tx should not be published anymore") + default: + } + + // Verify stage in nursery report. + assertNurseryReport(t, ctx.nursery, 1, 0, 0) + + ctx.nursery.Stop() +} + +func TestNurseryRemoteSpendOnRemote(t *testing.T) { + testRestartLoop(t, testNurseryRemoteSpendOnRemote) +} + +func testNurseryRemoteSpendOnRemote(t *testing.T, + checkStartStop func(func()) bool) { + + ctx := createNurseryTestContext(t, checkStartStop) + + outgoingRes := incubateTestOutput(t, ctx.nursery, false) + + ctx.notifier.confirmTx(&outgoingRes.ClaimOutpoint.Hash, 124) + + // Wait for output to be promoted from PSCL to KNDR. + select { + case <-ctx.store.preschoolToKinderChan: + case <-time.After(5 * time.Second): + t.Fatalf("output not promoted to KNDR") + } + + ctx.restart() + + // Remote spend the htlc using a remote success tx. Set witness script + // length to 5 so nursery recognizes it properly. + remoteSpendTxPreviousOutpoint := outgoingRes.ClaimOutpoint + + remoteSpendTx := wire.MsgTx{ + TxIn: []*wire.TxIn{ + { + PreviousOutPoint: remoteSpendTxPreviousOutpoint, + Witness: [][]byte{{}, {}, {}, {}, {}}, + }, + }, + } + ctx.notifier.spendOutpoint( + &remoteSpendTxPreviousOutpoint, + &remoteSpendTx) + + // Wait for output to be marked as SPND. + select { + case <-ctx.store.kidToRemoteSpendChan: + case <-time.After(5 * time.Second): + t.Fatalf("not marked as SPND") + } + + // Notify arrival of block where HTLC expires. + ctx.notifier.epochChan <- &chainntnfs.BlockEpoch{ + Height: 125, + } + + // This should not trigger the nursery to publish the timeout tx. + select { + case <-ctx.publishChan: + t.Fatalf("tx should not be published anymore") + default: + } + + // Verify stage in nursery report. + assertNurseryReport(t, ctx.nursery, 1, 0, 0) + + ctx.nursery.Stop() +} + +type nurseryStoreInterceptor struct { + ns NurseryStore + + // TODO(joostjager): put more useful info through these channels. + cribToKinderChan chan struct{} + cribToRemoteSpendChan chan struct{} + graduateKinderChan chan struct{} + preschoolToKinderChan chan struct{} + kidToRemoteSpendChan chan struct{} +} + +func newNurseryStoreInterceptor(ns NurseryStore) *nurseryStoreInterceptor { + return &nurseryStoreInterceptor{ + ns: ns, + cribToKinderChan: make(chan struct{}), + cribToRemoteSpendChan: make(chan struct{}), + graduateKinderChan: make(chan struct{}), + preschoolToKinderChan: make(chan struct{}), + kidToRemoteSpendChan: make(chan struct{}), + } +} + +func (i *nurseryStoreInterceptor) Incubate(kidOutputs []kidOutput, + babyOutputs []babyOutput) error { + + return i.ns.Incubate(kidOutputs, babyOutputs) +} + +func (i *nurseryStoreInterceptor) CribToKinder(babyOutput *babyOutput) error { + err := i.ns.CribToKinder(babyOutput) + + i.cribToKinderChan <- struct{}{} + + return err +} + +func (i *nurseryStoreInterceptor) PreschoolToKinder(kidOutput *kidOutput) error { + err := i.ns.PreschoolToKinder(kidOutput) + + i.preschoolToKinderChan <- struct{}{} + + return err +} + +func (i *nurseryStoreInterceptor) CribToRemoteSpend(baby *babyOutput) error { + err := i.ns.CribToRemoteSpend(baby) + + i.cribToRemoteSpendChan <- struct{}{} + + return err +} + +func (i *nurseryStoreInterceptor) KidToRemoteSpend(kid *kidOutput) error { + err := i.ns.KidToRemoteSpend(kid) + + i.kidToRemoteSpendChan <- struct{}{} + + return err +} + +func (i *nurseryStoreInterceptor) GraduateKinder(height uint32) error { + err := i.ns.GraduateKinder(height) + + i.graduateKinderChan <- struct{}{} + + return err +} + +func (i *nurseryStoreInterceptor) FetchPreschools() ([]kidOutput, error) { + return i.ns.FetchPreschools() +} + +func (i *nurseryStoreInterceptor) FetchKinder() ([]kidOutput, error) { + return i.ns.FetchKinder() +} + +func (i *nurseryStoreInterceptor) FetchCribs() ([]babyOutput, error) { + return i.ns.FetchCribs() +} + +func (i *nurseryStoreInterceptor) FetchClass(height uint32) (*wire.MsgTx, []kidOutput, []babyOutput, error) { + return i.ns.FetchClass(height) +} + +func (i *nurseryStoreInterceptor) FinalizeKinder(height uint32, tx *wire.MsgTx) error { + return i.ns.FinalizeKinder(height, tx) +} + +func (i *nurseryStoreInterceptor) LastFinalizedHeight() (uint32, error) { + return i.ns.LastFinalizedHeight() +} + +func (i *nurseryStoreInterceptor) GraduateHeight(height uint32) error { + return i.ns.GraduateHeight(height) +} + +func (i *nurseryStoreInterceptor) LastGraduatedHeight() (uint32, error) { + return i.ns.LastGraduatedHeight() +} + +func (i *nurseryStoreInterceptor) HeightsBelowOrEqual(height uint32) ([]uint32, error) { + return i.ns.HeightsBelowOrEqual(height) +} + +func (i *nurseryStoreInterceptor) ForChanOutputs(chanPoint *wire.OutPoint, + callback func([]byte, []byte) error) error { + + return i.ns.ForChanOutputs(chanPoint, callback) +} + +func (i *nurseryStoreInterceptor) ListChannels() ([]wire.OutPoint, error) { + return i.ns.ListChannels() +} + +func (i *nurseryStoreInterceptor) IsMatureChannel(chanPoint *wire.OutPoint) (bool, error) { + return i.ns.IsMatureChannel(chanPoint) +} + +func (i *nurseryStoreInterceptor) RemoveChannel(chanPoint *wire.OutPoint) error { + return i.ns.RemoveChannel(chanPoint) +} + +type mockFeeEstimator struct{} + +func (m *mockFeeEstimator) EstimateFeePerKW( + numBlocks uint32) (lnwallet.SatPerKWeight, error) { + + return lnwallet.SatPerKWeight(10000), nil +} + +func (m *mockFeeEstimator) Start() error { + return nil +} +func (m *mockFeeEstimator) Stop() error { + return nil +} + +type nurseryMockSigner struct { +} + +func (m *nurseryMockSigner) SignOutputRaw(tx *wire.MsgTx, + signDesc *lnwallet.SignDescriptor) ([]byte, error) { + + return []byte{}, nil +} + +func (m *nurseryMockSigner) ComputeInputScript(tx *wire.MsgTx, + signDesc *lnwallet.SignDescriptor) (*lnwallet.InputScript, error) { + + return &lnwallet.InputScript{}, nil +} + +type nurseryMockNotifier struct { + confChannel map[chainhash.Hash]chan *chainntnfs.TxConfirmation + epochChan chan *chainntnfs.BlockEpoch + spendChan map[wire.OutPoint]chan *chainntnfs.SpendDetail +} + +func newNurseryMockNotifier() *nurseryMockNotifier { + return &nurseryMockNotifier{ + confChannel: make(map[chainhash.Hash]chan *chainntnfs.TxConfirmation), + epochChan: make(chan *chainntnfs.BlockEpoch), + spendChan: make(map[wire.OutPoint]chan *chainntnfs.SpendDetail), + } +} + +func (m *nurseryMockNotifier) confirmTx(txid *chainhash.Hash, height uint32) { + m.getConfChannel(txid) <- &chainntnfs.TxConfirmation{BlockHeight: height} +} + +func (m *nurseryMockNotifier) spendOutpoint(outpoint *wire.OutPoint, + spendingTx *wire.MsgTx) { + + spenderTxHash := spendingTx.TxHash() + m.getSpendChannel(outpoint) <- &chainntnfs.SpendDetail{ + SpenderTxHash: &spenderTxHash, + SpendingTx: spendingTx, + } +} + +func (m *nurseryMockNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash, + _ []byte, numConfs, heightHint uint32) (*chainntnfs.ConfirmationEvent, error) { + + return &chainntnfs.ConfirmationEvent{ + Confirmed: m.getConfChannel(txid), + }, nil +} + +func (m *nurseryMockNotifier) getConfChannel(txid *chainhash.Hash) chan *chainntnfs.TxConfirmation { + channel, ok := m.confChannel[*txid] + if ok { + return channel + } + channel = make(chan *chainntnfs.TxConfirmation) + m.confChannel[*txid] = channel + return channel +} + +func (m *nurseryMockNotifier) RegisterBlockEpochNtfn( + bestBlock *chainntnfs.BlockEpoch) (*chainntnfs.BlockEpochEvent, error) { + return &chainntnfs.BlockEpochEvent{ + Epochs: m.epochChan, + Cancel: func() {}, + }, nil +} + +func (m *nurseryMockNotifier) Start() error { + return nil +} + +func (m *nurseryMockNotifier) Stop() error { + return nil +} + +func (m *nurseryMockNotifier) getSpendChannel(outpoint *wire.OutPoint) chan *chainntnfs.SpendDetail { + channel, ok := m.spendChan[*outpoint] + if ok { + return channel + } + channel = make(chan *chainntnfs.SpendDetail, 1) + m.spendChan[*outpoint] = channel + return channel +} + +func (m *nurseryMockNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, _ []byte, + heightHint uint32) (*chainntnfs.SpendEvent, error) { + return &chainntnfs.SpendEvent{ + Spend: m.getSpendChannel(outpoint), + Cancel: func() {}, + }, nil +}