multi: add height-based invoice expiry

This commit adds height-based invoice expiry for hodl invoices
that have active htlcs. This allows us to cancel our intentionally
held htlcs before channels are force closed. We only add this for
hodl invoices because we expect regular invoices to automatically
be resolved.

We still keep hodl invoices in the time-based expiry queue,
because we want to expire open invoices that reach their timeout
before any htlcs are added. Since htlcs are added after the
invoice is created, we add new htlcs as they arrive in the
invoice registry. In this commit, we allow adding of duplicate
entries for an invoice to be added to the expiry queue as each
htlc arrives to keep implementation simple. Our cancellation
logic can already handle the case where an entry is already
canceled, so this is ok.

invoices/invoice_expiry_watcher.go

@@ -5,6 +5,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/clock"
 	"github.com/lightningnetwork/lnd/lntypes"
@@ -34,6 +36,28 @@ func (e invoiceExpiryTs) Less(other queue.PriorityQueueItem) bool {
 	return e.Expiry.Before(other.(*invoiceExpiryTs).Expiry)
 }
 
+// Compile time assertion that invoiceExpiryHeight implements invoiceExpiry.
+var _ invoiceExpiry = (*invoiceExpiryHeight)(nil)
+
+// invoiceExpiryHeight holds information about an invoice which can be used to
+// cancel it based on its expiry height.
+type invoiceExpiryHeight struct {
+	paymentHash  lntypes.Hash
+	expiryHeight uint32
+}
+
+// Less implements PriorityQueueItem.Less such that the top item in the
+// priority queue is the lowest block height.
+func (b invoiceExpiryHeight) Less(other queue.PriorityQueueItem) bool {
+	return b.expiryHeight < other.(*invoiceExpiryHeight).expiryHeight
+}
+
+// expired returns a boolean that indicates whether this entry has expired,
+// taking our expiry delta into account.
+func (b invoiceExpiryHeight) expired(currentHeight, delta uint32) bool {
+	return currentHeight+delta >= b.expiryHeight
+}
+
 // InvoiceExpiryWatcher handles automatic invoice cancellation of expried
 // invoices. Upon start InvoiceExpiryWatcher will retrieve all pending (not yet
 // settled or canceled) invoices invoices to its watcing queue. When a new
@@ -49,6 +73,21 @@ type InvoiceExpiryWatcher struct {
 	// It is useful for testing.
 	clock clock.Clock
 
+	// notifier provides us with block height updates.
+	notifier chainntnfs.ChainNotifier
+
+	// blockExpiryDelta is the number of blocks before a htlc's expiry that
+	// we expire the invoice based on expiry height. We use a delta because
+	// we will go to some delta before our expiry, so we want to cancel
+	// before this to prevent force closes.
+	blockExpiryDelta uint32
+
+	// currentHeight is the current block height.
+	currentHeight uint32
+
+	// currentHash is the block hash for our current height.
+	currentHash *chainhash.Hash
+
 	// cancelInvoice is a template method that cancels an expired invoice.
 	cancelInvoice func(lntypes.Hash, bool) error
 
@@ -56,6 +95,15 @@ type InvoiceExpiryWatcher struct {
 	// the next invoice to expire.
 	timestampExpiryQueue queue.PriorityQueue
 
+	// blockExpiryQueue holds blockExpiry items and is used to find the
+	// next invoice to expire based on block height. Only hold invoices
+	// with active htlcs are added to this queue, because they require
+	// manual cancellation when the hltc is going to time out. Items in
+	// this queue may already be in the timestampExpiryQueue, this is ok
+	// because they will not be expired based on timestamp if they have
+	// active htlcs.
+	blockExpiryQueue queue.PriorityQueue
+
 	// newInvoices channel is used to wake up the main loop when a new
 	// invoices is added.
 	newInvoices chan []invoiceExpiry
@@ -67,11 +115,18 @@ type InvoiceExpiryWatcher struct {
 }
 
 // NewInvoiceExpiryWatcher creates a new InvoiceExpiryWatcher instance.
-func NewInvoiceExpiryWatcher(clock clock.Clock) *InvoiceExpiryWatcher {
+func NewInvoiceExpiryWatcher(clock clock.Clock,
+	expiryDelta, startHeight uint32, startHash *chainhash.Hash,
+	notifier chainntnfs.ChainNotifier) *InvoiceExpiryWatcher {
+
 	return &InvoiceExpiryWatcher{
-		clock:       clock,
-		newInvoices: make(chan []invoiceExpiry),
-		quit:        make(chan struct{}),
+		clock:            clock,
+		notifier:         notifier,
+		blockExpiryDelta: expiryDelta,
+		currentHeight:    startHeight,
+		currentHash:      startHash,
+		newInvoices:      make(chan []invoiceExpiry),
+		quit:             make(chan struct{}),
 	}
 }
 
@@ -91,8 +146,17 @@ func (ew *InvoiceExpiryWatcher) Start(
 
 	ew.started = true
 	ew.cancelInvoice = cancelInvoice
+
+	ntfn, err := ew.notifier.RegisterBlockEpochNtfn(&chainntnfs.BlockEpoch{
+		Height: int32(ew.currentHeight),
+		Hash:   ew.currentHash,
+	})
+	if err != nil {
+		return err
+	}
+
 	ew.wg.Add(1)
-	go ew.mainLoop()
+	go ew.mainLoop(ntfn)
 
 	return nil
 }
@@ -122,6 +186,32 @@ func makeInvoiceExpiry(paymentHash lntypes.Hash,
 	case channeldb.ContractOpen:
 		return makeTimestampExpiry(paymentHash, invoice)
 
+	// If an invoice has active htlcs, we want to expire it based on block
+	// height. We only do this for hodl invoices, since regular invoices
+	// should resolve themselves automatically.
+	case channeldb.ContractAccepted:
+		if !invoice.HodlInvoice {
+			log.Debugf("Invoice in accepted state not added to "+
+				"expiry watcher: %v", paymentHash)
+
+			return nil
+		}
+
+		var minHeight uint32
+		for _, htlc := range invoice.Htlcs {
+			// We only care about accepted htlcs, since they will
+			// trigger force-closes.
+			if htlc.State != channeldb.HtlcStateAccepted {
+				continue
+			}
+
+			if minHeight == 0 || htlc.Expiry < minHeight {
+				minHeight = htlc.Expiry
+			}
+		}
+
+		return makeHeightExpiry(paymentHash, minHeight)
+
 	default:
 		log.Debugf("Invoice not added to expiry watcher: %v",
 			paymentHash)
@@ -151,18 +241,36 @@ func makeTimestampExpiry(paymentHash lntypes.Hash,
 	}
 }
 
+// makeHeightExpiry creates height-based expiry for an invoice based on its
+// lowest htlc expiry height.
+func makeHeightExpiry(paymentHash lntypes.Hash,
+	minHeight uint32) *invoiceExpiryHeight {
+
+	if minHeight == 0 {
+		log.Warnf("make height expiry called with 0 height")
+		return nil
+	}
+
+	return &invoiceExpiryHeight{
+		paymentHash:  paymentHash,
+		expiryHeight: minHeight,
+	}
+}
+
 // AddInvoices adds invoices to the InvoiceExpiryWatcher.
 func (ew *InvoiceExpiryWatcher) AddInvoices(invoices ...invoiceExpiry) {
-	if len(invoices) > 0 {
-		select {
-		case ew.newInvoices <- invoices:
-			log.Debugf("Added %d invoices to the expiry watcher",
-				len(invoices))
+	if len(invoices) == 0 {
+		return
+	}
 
-		// Select on quit too so that callers won't get blocked in case
-		// of concurrent shutdown.
-		case <-ew.quit:
-		}
+	select {
+	case ew.newInvoices <- invoices:
+		log.Debugf("Added %d invoices to the expiry watcher",
+			len(invoices))
+
+	// Select on quit too so that callers won't get blocked in case
+	// of concurrent shutdown.
+	case <-ew.quit:
 	}
 }
 
@@ -178,6 +286,23 @@ func (ew *InvoiceExpiryWatcher) nextTimestampExpiry() <-chan time.Time {
 	return nil
 }
 
+// nextHeightExpiry returns a channel that will immediately be read from if
+// the top item on our queue has expired.
+func (ew *InvoiceExpiryWatcher) nextHeightExpiry() <-chan uint32 {
+	if ew.blockExpiryQueue.Empty() {
+		return nil
+	}
+
+	top := ew.blockExpiryQueue.Top().(*invoiceExpiryHeight)
+	if !top.expired(ew.currentHeight, ew.blockExpiryDelta) {
+		return nil
+	}
+
+	blockChan := make(chan uint32, 1)
+	blockChan <- top.expiryHeight
+	return blockChan
+}
+
 // cancelNextExpiredInvoice will cancel the next expired invoice and removes
 // it from the expiry queue.
 func (ew *InvoiceExpiryWatcher) cancelNextExpiredInvoice() {
@@ -198,6 +323,25 @@ func (ew *InvoiceExpiryWatcher) cancelNextExpiredInvoice() {
 	}
 }
 
+// cancelNextHeightExpiredInvoice looks at our height based queue and expires
+// the next invoice if we have reached its expiry block.
+func (ew *InvoiceExpiryWatcher) cancelNextHeightExpiredInvoice() {
+	if ew.blockExpiryQueue.Empty() {
+		return
+	}
+
+	top := ew.blockExpiryQueue.Top().(*invoiceExpiryHeight)
+	if !top.expired(ew.currentHeight, ew.blockExpiryDelta) {
+		return
+	}
+
+	// We always force-cancel block-based expiry so that we can
+	// cancel invoices that have been accepted but not yet resolved.
+	// This helps us avoid force closes.
+	ew.expireInvoice(top.paymentHash, true)
+	ew.blockExpiryQueue.Pop()
+}
+
 // expireInvoice attempts to expire an invoice and logs an error if we get an
 // unexpected error.
 func (ew *InvoiceExpiryWatcher) expireInvoice(hash lntypes.Hash, force bool) {
@@ -226,6 +370,11 @@ func (ew *InvoiceExpiryWatcher) pushInvoices(invoices []invoiceExpiry) {
 				ew.timestampExpiryQueue.Push(expiry)
 			}
 
+		case *invoiceExpiryHeight:
+			if expiry != nil {
+				ew.blockExpiryQueue.Push(expiry)
+			}
+
 		default:
 			log.Errorf("unexpected queue item: %T", inv)
 		}
@@ -234,12 +383,20 @@ func (ew *InvoiceExpiryWatcher) pushInvoices(invoices []invoiceExpiry) {
 
 // mainLoop is a goroutine that receives new invoices and handles cancellation
 // of expired invoices.
-func (ew *InvoiceExpiryWatcher) mainLoop() {
-	defer ew.wg.Done()
+func (ew *InvoiceExpiryWatcher) mainLoop(blockNtfns *chainntnfs.BlockEpochEvent) {
+	defer func() {
+		blockNtfns.Cancel()
+		ew.wg.Done()
+	}()
+
+	// We have two different queues, so we use a different cancel method
+	// depending on which expiry condition we have hit. Starting with time
+	// based expiry is an arbitrary choice to start off.
+	cancelNext := ew.cancelNextExpiredInvoice
 
 	for {
 		// Cancel any invoices that may have expired.
-		ew.cancelNextExpiredInvoice()
+		cancelNext()
 
 		select {
 
@@ -252,13 +409,29 @@ func (ew *InvoiceExpiryWatcher) mainLoop() {
 		default:
 			select {
 
+			// Wait until the next invoice expires.
 			case <-ew.nextTimestampExpiry():
-				// Wait until the next invoice expires.
+				cancelNext = ew.cancelNextExpiredInvoice
+				continue
+
+			case <-ew.nextHeightExpiry():
+				cancelNext = ew.cancelNextHeightExpiredInvoice
 				continue
 
 			case newInvoices := <-ew.newInvoices:
 				ew.pushInvoices(newInvoices)
 
+			// Consume new blocks.
+			case block, ok := <-blockNtfns.Epochs:
+				if !ok {
+					log.Debugf("block notifications " +
+						"canceled")
+					return
+				}
+
+				ew.currentHeight = uint32(block.Height)
+				ew.currentHash = block.Hash
+
 			case <-ew.quit:
 				return
 			}