Repository: lightningnetwork/lnd
Pull request #3957 (2020-01-25 04:59:09)

record+routing/route: add AMP record
This PR adds the currently proposed AMP record from https://github.com/lightningnetwork/lightning-rfc/pull/658, and
implements the sanity checks when encoding a particular `route.Hop`. For now, this could will
remain dead until a number of other components are in place, but this serves as a starting point to
begin implementing the remainder of the proposal that depends on these primitive definitions
before ultimately enabling it externally.

Related to #3939. 


Patch:
diff --git a/record/amp.go b/record/amp.go
new file mode 100644
index 000000000..72b4cbf42
--- /dev/null
+++ b/record/amp.go
@@ -0,0 +1,107 @@
+package record
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/lightningnetwork/lnd/tlv"
+)
+
+// AMPOnionType is the type used in the onion to reference the AMP fields:
+// root_share, set_id, and child_index.
+const AMPOnionType tlv.Type = 10
+
+// AMP is a record that encodes the fields necessary for atomic multi-path
+// payments.
+type AMP struct {
+	rootShare  [32]byte
+	setID      [32]byte
+	childIndex uint16
+}
+
+// NewAMP generate a new AMP record with the given root_share, set_id, and
+// child_index.
+func NewAMP(rootShare, setID [32]byte, childIndex uint16) *AMP {
+	return &AMP{
+		rootShare:  rootShare,
+		setID:      setID,
+		childIndex: childIndex,
+	}
+}
+
+// RootShare returns the root share contained in the AMP record.
+func (a *AMP) RootShare() [32]byte {
+	return a.rootShare
+}
+
+// SetID returns the set id contained in the AMP record.
+func (a *AMP) SetID() [32]byte {
+	return a.setID
+}
+
+// ChildIndex returns the child index contained in the AMP record.
+func (a *AMP) ChildIndex() uint16 {
+	return a.childIndex
+}
+
+// AMPEncoder writes the AMP record to the provided io.Writer.
+func AMPEncoder(w io.Writer, val interface{}, buf *[8]byte) error {
+	if v, ok := val.(*AMP); ok {
+		if err := tlv.EBytes32(w, &v.rootShare, buf); err != nil {
+			return err
+		}
+
+		if err := tlv.EBytes32(w, &v.setID, buf); err != nil {
+			return err
+		}
+
+		return tlv.ETUint16T(w, v.childIndex, buf)
+	}
+	return tlv.NewTypeForEncodingErr(val, "AMP")
+}
+
+const (
+	// minAMPLength is the minimum length of a serialized AMP TLV record,
+	// which occurs when the truncated encoding of child_index takes 0
+	// bytes, leaving only the root_share and set_id.
+	minAMPLength = 64
+
+	// maxAMPLength is the maximum legnth of a serialized AMP TLV record,
+	// which occurs when the truncated endoing of a child_index takes 2
+	// bytes.
+	maxAMPLength = 66
+)
+
+// AMPDecoder reads the AMP record from the provided io.Reader.
+func AMPDecoder(r io.Reader, val interface{}, buf *[8]byte, l uint64) error {
+	if v, ok := val.(*AMP); ok && minAMPLength <= l && l <= maxAMPLength {
+		if err := tlv.DBytes32(r, &v.rootShare, buf, 32); err != nil {
+			return err
+		}
+
+		if err := tlv.DBytes32(r, &v.setID, buf, 32); err != nil {
+			return err
+		}
+
+		return tlv.DTUint16(r, &v.childIndex, buf, l-64)
+	}
+	return tlv.NewTypeForDecodingErr(val, "AMP", l, maxAMPLength)
+}
+
+// Record returns a tlv.Record that can be used to encode or decode this record.
+func (a *AMP) Record() tlv.Record {
+	return tlv.MakeDynamicRecord(
+		AMPOnionType, a, a.PayloadSize, AMPEncoder, AMPDecoder,
+	)
+}
+
+// PayloadSize returns the size this record takes up in encoded form.
+func (a *AMP) PayloadSize() uint64 {
+	return 32 + 32 + tlv.SizeTUint16(a.childIndex)
+}
+
+// String returns a human-readble description of the amp payload fields.
+func (a *AMP) String() string {
+	return fmt.Sprintf("root_share=%x set_id=%x child_index=%d",
+		a.rootShare, a.setID, a.childIndex)
+}
diff --git a/routing/missioncontrol.go b/routing/missioncontrol.go
index f52c6fce9..c360953af 100644
--- a/routing/missioncontrol.go
+++ b/routing/missioncontrol.go
@@ -333,18 +333,18 @@ func (m *MissionControl) setLastPairResult(fromNode, toNode route.Vertex,
 	nodePairs[toNode] = current
 }
 
-// setAllFail stores a fail result for all known connections to and from the
-// given node.
-func (m *MissionControl) setAllFail(node route.Vertex,
+// setAllFail stores a fail result for all known connection of the given node.
+func (m *MissionControl) setAllFail(fromNode route.Vertex,
 	timestamp time.Time) {
 
-	for fromNode, nodePairs := range m.lastPairResult {
-		for toNode := range nodePairs {
-			if fromNode == node || toNode == node {
-				nodePairs[toNode] = TimedPairResult{
-					FailTime: timestamp,
-				}
-			}
+	nodePairs, ok := m.lastPairResult[fromNode]
+	if !ok {
+		return
+	}
+
+	for connection := range nodePairs {
+		nodePairs[connection] = TimedPairResult{
+			FailTime: timestamp,
 		}
 	}
 }
diff --git a/routing/result_interpretation.go b/routing/result_interpretation.go
index 660ce69cc..30d65cc0b 100644
--- a/routing/result_interpretation.go
+++ b/routing/result_interpretation.go
@@ -424,13 +424,11 @@ func (i *interpretedResult) failNode(rt *route.Route, idx int) {
 	// Mark the incoming connection as failed for the node. We intent to
 	// penalize as much as we can for a node level failure, including future
 	// outgoing traffic for this connection. The pair as it is returned by
-	// getPair is penalized in the original and the reversed direction. Note
-	// that this will also affect the score of the failing node's peers.
-	// This is necessary to prevent future routes from keep going into the
-	// same node again.
+	// getPair is directed towards the failed node. Therefore we first
+	// reverse the pair. We don't want to affect the score of the node
+	// sending towards the failing node.
 	incomingChannelIdx := idx - 1
 	inPair, _ := getPair(rt, incomingChannelIdx)
-	i.pairResults[inPair] = failPairResult(0)
 	i.pairResults[inPair.Reverse()] = failPairResult(0)
 
 	// If not the ultimate node, mark the outgoing connection as failed for
@@ -439,7 +437,6 @@ func (i *interpretedResult) failNode(rt *route.Route, idx int) {
 		outgoingChannelIdx := idx
 		outPair, _ := getPair(rt, outgoingChannelIdx)
 		i.pairResults[outPair] = failPairResult(0)
-		i.pairResults[outPair.Reverse()] = failPairResult(0)
 	}
 }
 
diff --git a/routing/route/route.go b/routing/route/route.go
index 90a407e38..f4de728fc 100644
--- a/routing/route/route.go
+++ b/routing/route/route.go
@@ -34,6 +34,10 @@ var (
 	// record to an intermediate hop, only final hops can receive MPP
 	// records.
 	ErrIntermediateMPPHop = errors.New("cannot send MPP to intermediate")
+
+	// ErrAMPMissingMPP is returned when the caller tries to attach an AMP
+	// record but no MPP record is presented for the final hop.
+	ErrAMPMissingMPP = errors.New("cannot send AMP without MPP record")
 )
 
 // Vertex is a simple alias for the serialization of a compressed Bitcoin
@@ -111,6 +115,10 @@ type Hop struct {
 	// only be set for the final hop.
 	MPP *record.MPP
 
+	// AMP encapsulates the data required for option_amp. This field should
+	// only be set for the final hop.
+	AMP *record.AMP
+
 	// CustomRecords if non-nil are a set of additional TLV records that
 	// should be included in the forwarding instructions for this node.
 	CustomRecords record.CustomSet
@@ -168,6 +176,18 @@ func (h *Hop) PackHopPayload(w io.Writer, nextChanID uint64) error {
 		}
 	}
 
+	// If an AMP record is destined for this hop, ensure that we only ever
+	// attach it if we also have an MPP record. We can infer that this is
+	// already a final hop if MPP is non-nil otherwise we would have exited
+	// above.
+	if h.AMP != nil {
+		if h.MPP != nil {
+			records = append(records, h.AMP.Record())
+		} else {
+			return ErrAMPMissingMPP
+		}
+	}
+
 	// Append any custom types destined for this hop.
 	tlvRecords := tlv.MapToRecords(h.CustomRecords)
 	records = append(records, tlvRecords...)
@@ -217,6 +237,11 @@ func (h *Hop) PayloadSize(nextChanID uint64) uint64 {
 		addRecord(record.MPPOnionType, h.MPP.PayloadSize())
 	}
 
+	// Add amp if present.
+	if h.AMP != nil {
+		addRecord(record.AMPOnionType, h.AMP.PayloadSize())
+	}
+
 	// Add custom records.
 	for k, v := range h.CustomRecords {
 		addRecord(tlv.Type(k), uint64(len(v)))
diff --git a/rpcserver.go b/rpcserver.go
index f4e13cd65..16e0677d2 100644
--- a/rpcserver.go
+++ b/rpcserver.go
@@ -750,17 +750,6 @@ func (r *rpcServer) Start() error {
 		}
 	}
 
-	// The default JSON marshaler of the REST proxy only sets OrigName to
-	// true, which instructs it to use the same field names as specified in
-	// the proto file and not switch to camel case. What we also want is
-	// that the marshaler prints all values, even if they are falsey.
-	customMarshalerOption := proxy.WithMarshalerOption(
-		proxy.MIMEWildcard, &proxy.JSONPb{
-			OrigName:     true,
-			EmitDefaults: true,
-		},
-	)
-
 	// Finally, start the REST proxy for our gRPC server above. We'll ensure
 	// we direct LND to connect to its loopback address rather than a
 	// wildcard to prevent certificate issues when accessing the proxy
@@ -768,7 +757,7 @@ func (r *rpcServer) Start() error {
 	//
 	// TODO(roasbeef): eventually also allow the sub-servers to themselves
 	// have a REST proxy.
-	mux := proxy.NewServeMux(customMarshalerOption)
+	mux := proxy.NewServeMux()
 
 	err := lnrpc.RegisterLightningHandlerFromEndpoint(
 		context.Background(), mux, r.restProxyDest,


Test patch:
diff --git a/record/record_test.go b/record/record_test.go
index 052e2f1f0..8c39790ed 100644
--- a/record/record_test.go
+++ b/record/record_test.go
@@ -17,8 +17,11 @@ type recordEncDecTest struct {
 }
 
 var (
-	testTotal = lnwire.MilliSatoshi(45)
-	testAddr  = [32]byte{0x01, 0x02}
+	testTotal      = lnwire.MilliSatoshi(45)
+	testAddr       = [32]byte{0x01, 0x02}
+	testShare      = [32]byte{0x03, 0x04}
+	testSetID      = [32]byte{0x05, 0x06}
+	testChildIndex = uint16(17)
 )
 
 var recordEncDecTests = []recordEncDecTest{
@@ -40,6 +43,29 @@ var recordEncDecTests = []recordEncDecTest{
 			}
 		},
 	},
+	{
+		name: "amp",
+		encRecord: func() tlv.RecordProducer {
+			return record.NewAMP(
+				testShare, testSetID, testChildIndex,
+			)
+		},
+		decRecord: func() tlv.RecordProducer {
+			return new(record.AMP)
+		},
+		assert: func(t *testing.T, r interface{}) {
+			amp := r.(*record.AMP)
+			if amp.RootShare() != testShare {
+				t.Fatal("incorrect root share")
+			}
+			if amp.SetID() != testSetID {
+				t.Fatal("incorrect set id")
+			}
+			if amp.ChildIndex() != testChildIndex {
+				t.Fatal("incorrect child index")
+			}
+		},
+	},
 }
 
 // TestRecordEncodeDecode is a generic test framework for custom TLV records. It
diff --git a/routing/missioncontrol_test.go b/routing/missioncontrol_test.go
index 98cf95302..f921b71de 100644
--- a/routing/missioncontrol_test.go
+++ b/routing/missioncontrol_test.go
@@ -186,8 +186,8 @@ func TestMissionControl(t *testing.T) {
 	// Check whether history snapshot looks sane.
 	history := ctx.mc.GetHistorySnapshot()
 
-	if len(history.Pairs) != 4 {
-		t.Fatalf("expected 4 pairs, but got %v", len(history.Pairs))
+	if len(history.Pairs) != 3 {
+		t.Fatalf("expected 3 pairs, but got %v", len(history.Pairs))
 	}
 
 	// Test reporting a success.
diff --git a/routing/result_interpretation_test.go b/routing/result_interpretation_test.go
index 537869f0b..2204139e3 100644
--- a/routing/result_interpretation_test.go
+++ b/routing/result_interpretation_test.go
@@ -165,8 +165,6 @@ var resultTestCases = []resultTestCase{
 			pairResults: map[DirectedNodePair]pairResult{
 				getTestPair(1, 0): failPairResult(0),
 				getTestPair(1, 2): failPairResult(0),
-				getTestPair(0, 1): failPairResult(0),
-				getTestPair(2, 1): failPairResult(0),
 			},
 		},
 	},
@@ -184,7 +182,6 @@ var resultTestCases = []resultTestCase{
 			nodeFailure:        &hops[1],
 			pairResults: map[DirectedNodePair]pairResult{
 				getTestPair(1, 0): failPairResult(0),
-				getTestPair(0, 1): failPairResult(0),
 			},
 		},
 	},
@@ -236,7 +233,6 @@ var resultTestCases = []resultTestCase{
 					amt:     97,
 				},
 				getTestPair(4, 3): {},
-				getTestPair(3, 4): {},
 			},
 			finalFailureReason: &reasonError,
 			nodeFailure:        &hops[4],
@@ -261,7 +257,6 @@ var resultTestCases = []resultTestCase{
 					amt:     99,
 				},
 				getTestPair(3, 2): {},
-				getTestPair(2, 3): {},
 			},
 			finalFailureReason: &reasonError,
 			nodeFailure:        &hops[3],
@@ -289,8 +284,6 @@ var resultTestCases = []resultTestCase{
 				},
 				getTestPair(3, 2): {},
 				getTestPair(3, 4): {},
-				getTestPair(2, 3): {},
-				getTestPair(4, 3): {},
 			},
 			nodeFailure: &hops[3],
 		},
@@ -308,7 +301,6 @@ var resultTestCases = []resultTestCase{
 		expectedResult: &interpretedResult{
 			pairResults: map[DirectedNodePair]pairResult{
 				getTestPair(1, 0): {},
-				getTestPair(0, 1): {},
 			},
 			finalFailureReason: &reasonError,
 			nodeFailure:        &hops[1],
diff --git a/routing/route/route_test.go b/routing/route/route_test.go
index 6c32d8be3..2095430b8 100644
--- a/routing/route/route_test.go
+++ b/routing/route/route_test.go
@@ -71,8 +71,8 @@ var (
 	testAddr = [32]byte{0x01, 0x02}
 )
 
-// TestMPPHop asserts that a Hop will encode a non-nil to final nodes, and fail
-// when trying to send to intermediaries.
+// TestMPPHop asserts that a Hop will encode a non-nil MPP to final nodes, and
+// fail when trying to send to intermediaries.
 func TestMPPHop(t *testing.T) {
 	t.Parallel()
 
@@ -101,6 +101,47 @@ func TestMPPHop(t *testing.T) {
 	}
 }
 
+// TestAMPHop asserts that a Hop will encode a non-nil AMP to final nodes of an
+// MPP record is also present, and fail otherwise.
+func TestAMPHop(t *testing.T) {
+	t.Parallel()
+
+	hop := Hop{
+		ChannelID:        1,
+		OutgoingTimeLock: 44,
+		AmtToForward:     testAmt,
+		LegacyPayload:    false,
+		AMP:              record.NewAMP([32]byte{}, [32]byte{}, 3),
+	}
+
+	// Encoding an AMP record to an intermediate hop w/o an MPP record
+	// should result in a failure.
+	var b bytes.Buffer
+	err := hop.PackHopPayload(&b, 2)
+	if err != ErrAMPMissingMPP {
+		t.Fatalf("expected err: %v, got: %v",
+			ErrAMPMissingMPP, err)
+	}
+
+	// Encoding an AMP record to a final hop w/o an MPP record should result
+	// in a failure.
+	b.Reset()
+	err = hop.PackHopPayload(&b, 0)
+	if err != ErrAMPMissingMPP {
+		t.Fatalf("expected err: %v, got: %v",
+			ErrAMPMissingMPP, err)
+	}
+
+	// Encoding an AMP record to a final hop w/ an MPP record should be
+	// successful.
+	hop.MPP = record.NewMPP(testAmt, testAddr)
+	b.Reset()
+	err = hop.PackHopPayload(&b, 0)
+	if err != nil {
+		t.Fatalf("expected err: %v, got: %v", nil, err)
+	}
+}
+
 // TestPayloadSize tests the payload size calculation that is provided by Hop
 // structs.
 func TestPayloadSize(t *testing.T) {
@@ -123,6 +164,7 @@ func TestPayloadSize(t *testing.T) {
 			AmtToForward:     1200,
 			OutgoingTimeLock: 700000,
 			MPP:              record.NewMPP(500, [32]byte{}),
+			AMP:              record.NewAMP([32]byte{}, [32]byte{}, 8),
 			CustomRecords: map[uint64][]byte{
 				100000:  {1, 2, 3},
 				1000000: {4, 5},