htlcswitch+invoices: allow settling invoices via multi-path payments

channeldb/invoices.go

@@ -117,6 +117,7 @@ const (
 	resolveTimeType  tlv.Type = 11
 	expiryHeightType tlv.Type = 13
 	htlcStateType    tlv.Type = 15
+	mppTotalAmtType  tlv.Type = 17
 
 	// A set of tlv type definitions used to serialize invoice bodiees.
 	//
@@ -289,6 +290,10 @@ type InvoiceHTLC struct {
 	// Amt is the amount that is carried by this htlc.
 	Amt lnwire.MilliSatoshi
 
+	// MppTotalAmt is a field for mpp that indicates the expected total
+	// amount.
+	MppTotalAmt lnwire.MilliSatoshi
+
 	// AcceptHeight is the block height at which the invoice registry
 	// decided to accept this htlc as a payment to the invoice. At this
 	// height, the invoice cltv delay must have been met.
@@ -323,6 +328,10 @@ type HtlcAcceptDesc struct {
 	// Amt is the amount that is carried by this htlc.
 	Amt lnwire.MilliSatoshi
 
+	// MppTotalAmt is a field for mpp that indicates the expected total
+	// amount.
+	MppTotalAmt lnwire.MilliSatoshi
+
 	// Expiry is the expiry height of this htlc.
 	Expiry uint32
 
@@ -1018,6 +1027,7 @@ func serializeHtlcs(w io.Writer, htlcs map[CircuitKey]*InvoiceHTLC) error {
 		// Encode the htlc in a tlv stream.
 		chanID := key.ChanID.ToUint64()
 		amt := uint64(htlc.Amt)
+		mppTotalAmt := uint64(htlc.MppTotalAmt)
 		acceptTime := uint64(htlc.AcceptTime.UnixNano())
 		resolveTime := uint64(htlc.ResolveTime.UnixNano())
 		state := uint8(htlc.State)
@@ -1034,6 +1044,7 @@ func serializeHtlcs(w io.Writer, htlcs map[CircuitKey]*InvoiceHTLC) error {
 			tlv.MakePrimitiveRecord(resolveTimeType, &resolveTime),
 			tlv.MakePrimitiveRecord(expiryHeightType, &htlc.Expiry),
 			tlv.MakePrimitiveRecord(htlcStateType, &state),
+			tlv.MakePrimitiveRecord(mppTotalAmtType, &mppTotalAmt),
 		)
 
 		// Convert the custom records to tlv.Record types that are ready
@@ -1193,7 +1204,7 @@ func deserializeHtlcs(r io.Reader) (map[CircuitKey]*InvoiceHTLC, error) {
 			chanID                  uint64
 			state                   uint8
 			acceptTime, resolveTime uint64
-			amt                     uint64
+			amt, mppTotalAmt        uint64
 		)
 		tlvStream, err := tlv.NewStream(
 			tlv.MakePrimitiveRecord(chanIDType, &chanID),
@@ -1206,6 +1217,7 @@ func deserializeHtlcs(r io.Reader) (map[CircuitKey]*InvoiceHTLC, error) {
 			tlv.MakePrimitiveRecord(resolveTimeType, &resolveTime),
 			tlv.MakePrimitiveRecord(expiryHeightType, &htlc.Expiry),
 			tlv.MakePrimitiveRecord(htlcStateType, &state),
+			tlv.MakePrimitiveRecord(mppTotalAmtType, &mppTotalAmt),
 		)
 		if err != nil {
 			return nil, err
@@ -1221,6 +1233,7 @@ func deserializeHtlcs(r io.Reader) (map[CircuitKey]*InvoiceHTLC, error) {
 		htlc.ResolveTime = time.Unix(0, int64(resolveTime))
 		htlc.State = HtlcState(state)
 		htlc.Amt = lnwire.MilliSatoshi(amt)
+		htlc.MppTotalAmt = lnwire.MilliSatoshi(mppTotalAmt)
 
 		// Reconstruct the custom records fields from the parsed types
 		// map return from the tlv parser.
@@ -1324,6 +1337,7 @@ func (d *DB) updateInvoice(hash lntypes.Hash, invoices, settleIndex *bbolt.Bucke
 
 		htlc := &InvoiceHTLC{
 			Amt:           htlcUpdate.Amt,
+			MppTotalAmt:   htlcUpdate.MppTotalAmt,
 			Expiry:        htlcUpdate.Expiry,
 			AcceptHeight:  uint32(htlcUpdate.AcceptHeight),
 			AcceptTime:    now,

htlcswitch/mock.go

@@ -790,9 +790,12 @@ func newMockRegistry(minDelta uint32) *mockInvoiceRegistry {
 		panic(err)
 	}
 
-	finalCltvRejectDelta := int32(5)
-
-	registry := invoices.NewRegistry(cdb, finalCltvRejectDelta)
+	registry := invoices.NewRegistry(
+		cdb,
+		&invoices.RegistryConfig{
+			FinalCltvRejectDelta: 5,
+		},
+	)
 	registry.Start()
 
 	return &mockInvoiceRegistry{

invoices/clock_test.go

@@ -0,0 +1,77 @@
+package invoices
+
+import (
+	"sync"
+	"time"
+)
+
+// testClock can be used in tests to mock time.
+type testClock struct {
+	currentTime time.Time
+	timeChanMap map[time.Time][]chan time.Time
+	timeLock    sync.Mutex
+}
+
+// newTestClock returns a new test clock.
+func newTestClock(startTime time.Time) *testClock {
+	return &testClock{
+		currentTime: startTime,
+		timeChanMap: make(map[time.Time][]chan time.Time),
+	}
+}
+
+// now returns the current (test) time.
+func (c *testClock) now() time.Time {
+	c.timeLock.Lock()
+	defer c.timeLock.Unlock()
+
+	return c.currentTime
+}
+
+// tickAfter returns a channel that will receive a tick at the specified time.
+func (c *testClock) tickAfter(duration time.Duration) <-chan time.Time {
+	c.timeLock.Lock()
+	defer c.timeLock.Unlock()
+
+	triggerTime := c.currentTime.Add(duration)
+	log.Debugf("tickAfter called: duration=%v, trigger_time=%v",
+		duration, triggerTime)
+
+	ch := make(chan time.Time, 1)
+
+	// If already expired, tick immediately.
+	if !triggerTime.After(c.currentTime) {
+		ch <- c.currentTime
+		return ch
+	}
+
+	// Otherwise store the channel until the trigger time is there.
+	chans := c.timeChanMap[triggerTime]
+	chans = append(chans, ch)
+	c.timeChanMap[triggerTime] = chans
+
+	return ch
+}
+
+// setTime sets the (test) time and triggers tick channels when they expire.
+func (c *testClock) setTime(now time.Time) {
+	c.timeLock.Lock()
+	defer c.timeLock.Unlock()
+
+	c.currentTime = now
+	remainingChans := make(map[time.Time][]chan time.Time)
+	for triggerTime, chans := range c.timeChanMap {
+		// If the trigger time is still in the future, keep this channel
+		// in the channel map for later.
+		if triggerTime.After(now) {
+			remainingChans[triggerTime] = chans
+			continue
+		}
+
+		for _, c := range chans {
+			c <- now
+		}
+	}
+
+	c.timeChanMap = remainingChans
+}