Merge pull request #9897 from ellemouton/inboundFeeTLV

multi: explicitly define InboundFees in ChannelUpdate and ChannelEdgePolicy

Author: Roasbeef

discovery/gossiper.go

@@ -3257,6 +3257,7 @@ func (d *AuthenticatedGossiper) handleChanUpdate(ctx context.Context,
 		MaxHTLC:                   upd.HtlcMaximumMsat,
 		FeeBaseMSat:               lnwire.MilliSatoshi(upd.BaseFee),
 		FeeProportionalMillionths: lnwire.MilliSatoshi(upd.FeeRate),
+		InboundFee:                upd.InboundFee.ValOpt(),
 		ExtraOpaqueData:           upd.ExtraOpaqueData,
 	}
 

docs/release-notes/release-notes-0.20.0.md

@@ -88,6 +88,11 @@ circuit. The indices are only available for forwarding events saved after v0.20.
 # Technical and Architectural Updates
 ## BOLT Spec Updates
 
+* Explicitly define the [inbound fee TLV 
+  record](https://github.com/lightningnetwork/lnd/pull/9897) on the 
+  `channel_update` message and handle it explicitly throughout the code base 
+  instead of extracting it from the TLV stream at various call-sites.
+
 ## Testing
 
 ## Database
@@ -99,4 +104,5 @@ circuit. The indices are only available for forwarding events saved after v0.20.
 # Contributors (Alphabetical Order)
 
 * Abdulkbk
-Pins
+* Elle Mouton
+* Pins

go.mod

@@ -203,6 +203,9 @@ require (
 // store have been included in a tagged sqldb version.
 replace github.com/lightningnetwork/lnd/sqldb => ./sqldb
 
+// TODO(elle): replace once the updated tlv package has been tagged.
+replace github.com/lightningnetwork/lnd/tlv => ./tlv
+
 // This replace is for https://github.com/advisories/GHSA-25xm-hr59-7c27
 replace github.com/ulikunitz/xz => github.com/ulikunitz/xz v0.5.11
 

go.sum

@@ -377,8 +377,6 @@ github.com/lightningnetwork/lnd/queue v1.1.1 h1:99ovBlpM9B0FRCGYJo6RSFDlt8/vOkQQ
 github.com/lightningnetwork/lnd/queue v1.1.1/go.mod h1:7A6nC1Qrm32FHuhx/mi1cieAiBZo5O6l8IBIoQxvkz4=
 github.com/lightningnetwork/lnd/ticker v1.1.1 h1:J/b6N2hibFtC7JLV77ULQp++QLtCwT6ijJlbdiZFbSM=
 github.com/lightningnetwork/lnd/ticker v1.1.1/go.mod h1:waPTRAAcwtu7Ji3+3k+u/xH5GHovTsCoSVpho0KDvdA=
-github.com/lightningnetwork/lnd/tlv v1.3.1 h1:o7CZg06y+rJZfUMAo0WzBLr0pgBWCzrt0f9gpujYUzk=
-github.com/lightningnetwork/lnd/tlv v1.3.1/go.mod h1:pJuiBj1ecr1WWLOtcZ+2+hu9Ey25aJWFIsjmAoPPnmc=
 github.com/lightningnetwork/lnd/tor v1.1.6 h1:WHUumk7WgU6BUFsqHuqszI9P6nfhMeIG+rjJBlVE6OE=
 github.com/lightningnetwork/lnd/tor v1.1.6/go.mod h1:qSRB8llhAK+a6kaTPWOLLXSZc6Hg8ZC0mq1sUQ/8JfI=
 github.com/ltcsuite/ltcd v0.0.0-20190101042124-f37f8bf35796 h1:sjOGyegMIhvgfq5oaue6Td+hxZuf3tDC8lAPrFldqFw=

graph/builder.go

@@ -943,7 +943,7 @@ func (b *Builder) ApplyChannelUpdate(msg *lnwire.ChannelUpdate1) bool {
 		return false
 	}
 
-	err = b.UpdateEdge(&models.ChannelEdgePolicy{
+	update := &models.ChannelEdgePolicy{
 		SigBytes:                  msg.Signature.ToSignatureBytes(),
 		ChannelID:                 msg.ShortChannelID.ToUint64(),
 		LastUpdate:                time.Unix(int64(msg.Timestamp), 0),
@@ -954,8 +954,11 @@ func (b *Builder) ApplyChannelUpdate(msg *lnwire.ChannelUpdate1) bool {
 		MaxHTLC:                   msg.HtlcMaximumMsat,
 		FeeBaseMSat:               lnwire.MilliSatoshi(msg.BaseFee),
 		FeeProportionalMillionths: lnwire.MilliSatoshi(msg.FeeRate),
+		InboundFee:                msg.InboundFee.ValOpt(),
 		ExtraOpaqueData:           msg.ExtraOpaqueData,
-	})
+	}
+
+	err = b.UpdateEdge(update)
 	if err != nil && !IsError(err, ErrIgnored, ErrOutdated) {
 		log.Errorf("Unable to apply channel update: %v", err)
 		return false

graph/db/errors.go

@@ -12,6 +12,10 @@ var (
 	ErrEdgePolicyOptionalFieldNotFound = fmt.Errorf("optional field not " +
 		"present")
 
+	// ErrParsingExtraTLVBytes is returned when we attempt to parse
+	// extra opaque bytes as a TLV stream, but the parsing fails.
+	ErrParsingExtraTLVBytes = fmt.Errorf("error parsing extra TLV bytes")
+
 	// ErrGraphNotFound is returned when at least one of the components of
 	// graph doesn't exist.
 	ErrGraphNotFound = fmt.Errorf("graph bucket not initialized")

graph/db/graph_cache.go

@@ -180,17 +180,6 @@ func (c *GraphCache) updateOrAddEdge(node route.Vertex, edge *DirectedChannel) {
 func (c *GraphCache) UpdatePolicy(policy *models.ChannelEdgePolicy, fromNode,
 	toNode route.Vertex, edge1 bool) {
 
-	// Extract inbound fee if possible and available. If there is a decoding
-	// error, ignore this policy.
-	var inboundFee lnwire.Fee
-	_, err := policy.ExtraOpaqueData.ExtractRecords(&inboundFee)
-	if err != nil {
-		log.Errorf("Failed to extract records from edge policy %v: %v",
-			policy.ChannelID, err)
-
-		return
-	}
-
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 
@@ -216,13 +205,17 @@ func (c *GraphCache) UpdatePolicy(policy *models.ChannelEdgePolicy, fromNode,
 		// policy for node 1.
 		case channel.IsNode1 && edge1:
 			channel.OutPolicySet = true
-			channel.InboundFee = inboundFee
+			policy.InboundFee.WhenSome(func(fee lnwire.Fee) {
+				channel.InboundFee = fee
+			})
 
 		// This is node 2, and it is edge 2, so this is the outgoing
 		// policy for node 2.
 		case !channel.IsNode1 && !edge1:
 			channel.OutPolicySet = true
-			channel.InboundFee = inboundFee
+			policy.InboundFee.WhenSome(func(fee lnwire.Fee) {
+				channel.InboundFee = fee
+			})
 
 		// The other two cases left mean it's the inbound policy for the
 		// node.

graph/db/graph_cache_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/hex"
 	"testing"
 
+	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/graph/db/models"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/routing/route"
@@ -37,11 +38,17 @@ func TestGraphCacheAddNode(t *testing.T) {
 			channelFlagA, channelFlagB = 1, 0
 		}
 
+		inboundFee := lnwire.Fee{
+			BaseFee: 10,
+			FeeRate: 20,
+		}
+
 		outPolicy1 := &models.ChannelEdgePolicy{
 			ChannelID:    1000,
 			ChannelFlags: lnwire.ChanUpdateChanFlags(channelFlagA),
 			ToNode:       nodeB,
 			// Define an inbound fee.
+			InboundFee: fn.Some(inboundFee),
 			ExtraOpaqueData: []byte{
 				253, 217, 3, 8, 0, 0, 0, 10, 0, 0, 0, 20,
 			},

graph/db/graph_test.go

@@ -4195,19 +4195,25 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) {
 
 	directedChan := getSingleChannel()
 	require.NotNil(t, directedChan)
-	require.Equal(t, directedChan.InboundFee, lnwire.Fee{
+	expectedInbound := lnwire.Fee{
 		BaseFee: 10,
 		FeeRate: 20,
-	})
+	}
+	require.Equal(t, expectedInbound, directedChan.InboundFee)
 
-	// Set an invalid inbound fee and check that the edge is no longer
-	// returned.
+	// Set an invalid inbound fee and check that persistence fails.
 	edge1.ExtraOpaqueData = []byte{
 		253, 217, 3, 8, 0,
 	}
-	require.NoError(t, graph.UpdateEdgePolicy(edge1))
+	require.ErrorIs(
+		t, graph.UpdateEdgePolicy(edge1), ErrParsingExtraTLVBytes,
+	)
 
-	require.Nil(t, getSingleChannel())
+	// Since persistence of the last update failed, we should still bet
+	// the previous result when we query the channel again.
+	directedChan = getSingleChannel()
+	require.NotNil(t, directedChan)
+	require.Equal(t, expectedInbound, directedChan.InboundFee)
 }
 
 // TestGraphLoading asserts that the cache is properly reconstructed after a

graph/db/kv_store.go

@@ -22,6 +22,7 @@ import (
 	"github.com/btcsuite/btcwallet/walletdb"
 	"github.com/lightningnetwork/lnd/aliasmgr"
 	"github.com/lightningnetwork/lnd/batch"
+	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/graph/db/models"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/kvdb"
@@ -284,6 +285,13 @@ func (c *KVStore) getChannelMap(edges kvdb.RBucket) (
 		case errors.Is(err, ErrEdgePolicyOptionalFieldNotFound):
 			return nil
 
+		// We don't want a single policy with bad TLV data to stop us
+		// from loading the rest of the data, so we just skip this
+		// policy. This is for backwards compatibility since we did not
+		// use to validate TLV data in the past before persisting it.
+		case errors.Is(err, ErrParsingExtraTLVBytes):
+			return nil
+
 		case err != nil:
 			return err
 		}
@@ -474,24 +482,19 @@ func (c *KVStore) forEachNodeDirectedChannel(tx kvdb.RTx,
 			cachedInPolicy.ToNodeFeatures = toNodeFeatures
 		}
 
-		var inboundFee lnwire.Fee
-		if p1 != nil {
-			// Extract inbound fee. If there is a decoding error,
-			// skip this edge.
-			_, err := p1.ExtraOpaqueData.ExtractRecords(&inboundFee)
-			if err != nil {
-				return nil
-			}
-		}
-
 		directedChannel := &DirectedChannel{
 			ChannelID:    e.ChannelID,
 			IsNode1:      node == e.NodeKey1Bytes,
 			OtherNode:    e.NodeKey2Bytes,
 			Capacity:     e.Capacity,
 			OutPolicySet: p1 != nil,
 			InPolicy:     cachedInPolicy,
-			InboundFee:   inboundFee,
+		}
+
+		if p1 != nil {
+			p1.InboundFee.WhenSome(func(fee lnwire.Fee) {
+				directedChannel.InboundFee = fee
+			})
 		}
 
 		if node == e.NodeKey2Bytes {
@@ -2342,7 +2345,7 @@ func (c *KVStore) FilterChannelRange(startHeight,
 				edge, err := deserializeChanEdgePolicyRaw(r)
 				if err != nil && !errors.Is(
 					err, ErrEdgePolicyOptionalFieldNotFound,
-				) {
+				) && !errors.Is(err, ErrParsingExtraTLVBytes) {
 
 					return err
 				}
@@ -2357,7 +2360,7 @@ func (c *KVStore) FilterChannelRange(startHeight,
 				edge, err := deserializeChanEdgePolicyRaw(r)
 				if err != nil && !errors.Is(
 					err, ErrEdgePolicyOptionalFieldNotFound,
-				) {
+				) && !errors.Is(err, ErrParsingExtraTLVBytes) {
 
 					return err
 				}
@@ -4334,15 +4337,20 @@ func putChanEdgePolicy(edges kvdb.RwBucket, edge *models.ChannelEdgePolicy,
 		// need to deserialize the existing policy within the database
 		// (now outdated by the new one), and delete its corresponding
 		// entry within the update index. We'll ignore any
-		// ErrEdgePolicyOptionalFieldNotFound error, as we only need
-		// the channel ID and update time to delete the entry.
+		// ErrEdgePolicyOptionalFieldNotFound or ErrParsingExtraTLVBytes
+		// errors, as we only need the channel ID and update time to
+		// delete the entry.
+		//
 		// TODO(halseth): get rid of these invalid policies in a
 		// migration.
+		// TODO(elle): complete the above TODO in migration from kvdb
+		// to SQL.
 		oldEdgePolicy, err := deserializeChanEdgePolicy(
 			bytes.NewReader(edgeBytes),
 		)
 		if err != nil &&
-			!errors.Is(err, ErrEdgePolicyOptionalFieldNotFound) {
+			!errors.Is(err, ErrEdgePolicyOptionalFieldNotFound) &&
+			!errors.Is(err, ErrParsingExtraTLVBytes) {
 
 			return err
 		}
@@ -4449,6 +4457,11 @@ func fetchChanEdgePolicy(edges kvdb.RBucket, chanID []byte,
 	case errors.Is(err, ErrEdgePolicyOptionalFieldNotFound):
 		return nil, nil
 
+	// If the policy contains invalid TLV bytes, we return nil as if
+	// the policy was unknown.
+	case errors.Is(err, ErrParsingExtraTLVBytes):
+		return nil, nil
+
 	case err != nil:
 		return nil, err
 	}
@@ -4546,6 +4559,12 @@ func serializeChanEdgePolicy(w io.Writer, edge *models.ChannelEdgePolicy,
 		}
 	}
 
+	// Validate that the ExtraOpaqueData is in fact a valid TLV stream.
+	err = edge.ExtraOpaqueData.ValidateTLV()
+	if err != nil {
+		return fmt.Errorf("%w: %w", ErrParsingExtraTLVBytes, err)
+	}
+
 	if len(edge.ExtraOpaqueData) > MaxAllowedExtraOpaqueBytes {
 		return ErrTooManyExtraOpaqueBytes(len(edge.ExtraOpaqueData))
 	}
@@ -4562,15 +4581,18 @@ func serializeChanEdgePolicy(w io.Writer, edge *models.ChannelEdgePolicy,
 
 func deserializeChanEdgePolicy(r io.Reader) (*models.ChannelEdgePolicy, error) {
 	// Deserialize the policy. Note that in case an optional field is not
-	// found, both an error and a populated policy object are returned.
-	edge, deserializeErr := deserializeChanEdgePolicyRaw(r)
-	if deserializeErr != nil &&
-		!errors.Is(deserializeErr, ErrEdgePolicyOptionalFieldNotFound) {
+	// found or if the edge has invalid TLV data, then both an error and a
+	// populated policy object are returned so that the caller can decide
+	// if it still wants to use the edge or not.
+	edge, err := deserializeChanEdgePolicyRaw(r)
+	if err != nil &&
+		!errors.Is(err, ErrEdgePolicyOptionalFieldNotFound) &&
+		!errors.Is(err, ErrParsingExtraTLVBytes) {
 
-		return nil, deserializeErr
+		return nil, err
 	}
 
-	return edge, deserializeErr
+	return edge, err
 }
 
 func deserializeChanEdgePolicyRaw(r io.Reader) (*models.ChannelEdgePolicy,
@@ -4657,6 +4679,22 @@ func deserializeChanEdgePolicyRaw(r io.Reader) (*models.ChannelEdgePolicy,
 		edge.ExtraOpaqueData = opq[8:]
 	}
 
+	// Attempt to extract the inbound fee from the opaque data. If we fail
+	// to parse the TLV here, we return an error we also return the edge
+	// so that the caller can still use it. This is for backwards
+	// compatibility in case we have already persisted some policies that
+	// have invalid TLV data.
+	var inboundFee lnwire.Fee
+	typeMap, err := edge.ExtraOpaqueData.ExtractRecords(&inboundFee)
+	if err != nil {
+		return edge, fmt.Errorf("%w: %w", ErrParsingExtraTLVBytes, err)
+	}
+
+	val, ok := typeMap[lnwire.FeeRecordType]
+	if ok && val == nil {
+		edge.InboundFee = fn.Some(inboundFee)
+	}
+
 	return edge, nil
 }