diff --git a/discovery/chan_series.go b/discovery/chan_series.go index 9ba9607e9..ed64453a4 100644 --- a/discovery/chan_series.go +++ b/discovery/chan_series.go @@ -73,12 +73,12 @@ type ChannelGraphTimeSeries interface { // in-protocol channel range queries to quickly and efficiently synchronize our // channel state with all peers. type ChanSeries struct { - graph *graphdb.ChannelGraph + graph *graphdb.VersionedGraph } // NewChanSeries constructs a new ChanSeries backed by a channeldb.ChannelGraph. // The returned ChanSeries implements the ChannelGraphTimeSeries interface. -func NewChanSeries(graph *graphdb.ChannelGraph) *ChanSeries { +func NewChanSeries(graph *graphdb.VersionedGraph) *ChanSeries { return &ChanSeries{ graph: graph, } diff --git a/docs/release-notes/release-notes-0.21.0.md b/docs/release-notes/release-notes-0.21.0.md index f81ba63dc..d6b3742f6 100644 --- a/docs/release-notes/release-notes-0.21.0.md +++ b/docs/release-notes/release-notes-0.21.0.md @@ -180,7 +180,8 @@ nodes and channels [1](https://github.com/lightningnetwork/lnd/pull/10339) [2](https://github.com/lightningnetwork/lnd/pull/10379) [3](https://github.com/lightningnetwork/lnd/pull/10380) - [4](https://github.com/lightningnetwork/lnd/pull/10542). + [4](https://github.com/lightningnetwork/lnd/pull/10542), + [5](https://github.com/lightningnetwork/lnd/pull/10572). ## Code Health diff --git a/graph/builder.go b/graph/builder.go index 9d3baebfe..5100066a3 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -573,13 +573,15 @@ func (b *Builder) pruneZombieChans() error { // both edges. If they're both disabled, then we can interpret this as // the channel being closed and can prune it from our graph. if b.cfg.AssumeChannelValid { - disabledChanIDs, err := b.cfg.Graph.DisabledChannelIDs() + disabledChanIDs, err := b.cfg.Graph.DisabledChannelIDs( + lnwire.GossipVersion1, + ) if err != nil { return fmt.Errorf("unable to get disabled channels "+ "ids chans: %v", err) } - disabledEdges, err := b.cfg.Graph.FetchChanInfos( + disabledEdges, err := b.v1Graph.FetchChanInfos( disabledChanIDs, ) if err != nil { diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 00cfeaf33..2887b7d22 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -669,7 +669,8 @@ func BenchmarkGraphReadMethods(b *testing.B) { name: "ForEachNodeCacheable", fn: func(b testing.TB, store Store) { err := store.ForEachNodeCacheable( - ctx, func(_ route.Vertex, + ctx, lnwire.GossipVersion1, + func(_ route.Vertex, _ *lnwire.FeatureVector) error { // Increment the counter to diff --git a/graph/db/channel_cache.go b/graph/db/channel_cache.go index b50bbf498..b65a5ab5e 100644 --- a/graph/db/channel_cache.go +++ b/graph/db/channel_cache.go @@ -1,11 +1,20 @@ package graphdb +import "github.com/lightningnetwork/lnd/lnwire" + +// channelCacheKey uniquely identifies a channel entry in the channel cache by +// gossip version and channel ID. +type channelCacheKey struct { + version lnwire.GossipVersion + chanID uint64 +} + // channelCache is an in-memory cache used to improve the performance of // ChanUpdatesInHorizon. It caches the chan info and edge policies for a // particular channel. type channelCache struct { n int - channels map[uint64]ChannelEdge + channels map[channelCacheKey]ChannelEdge } // newChannelCache creates a new channelCache with maximum capacity of n @@ -13,13 +22,18 @@ type channelCache struct { func newChannelCache(n int) *channelCache { return &channelCache{ n: n, - channels: make(map[uint64]ChannelEdge), + channels: make(map[channelCacheKey]ChannelEdge), } } // get returns the channel from the cache, if it exists. -func (c *channelCache) get(chanid uint64) (ChannelEdge, bool) { - channel, ok := c.channels[chanid] +func (c *channelCache) get(version lnwire.GossipVersion, + chanid uint64) (ChannelEdge, bool) { + + channel, ok := c.channels[channelCacheKey{ + version: version, + chanID: chanid, + }] return channel, ok } @@ -27,10 +41,17 @@ func (c *channelCache) get(chanid uint64) (ChannelEdge, bool) { // exists, it will be replaced with the new entry. If the entry doesn't exist, // it will be inserted to the cache, performing a random eviction if the cache // is at capacity. -func (c *channelCache) insert(chanid uint64, channel ChannelEdge) { +func (c *channelCache) insert(version lnwire.GossipVersion, chanid uint64, + channel ChannelEdge) { + + key := channelCacheKey{ + version: version, + chanID: chanid, + } + // If entry exists, replace it. - if _, ok := c.channels[chanid]; ok { - c.channels[chanid] = channel + if _, ok := c.channels[key]; ok { + c.channels[key] = channel return } @@ -41,10 +62,13 @@ func (c *channelCache) insert(chanid uint64, channel ChannelEdge) { break } } - c.channels[chanid] = channel + c.channels[key] = channel } // remove deletes an edge for chanid from the cache, if it exists. -func (c *channelCache) remove(chanid uint64) { - delete(c.channels, chanid) +func (c *channelCache) remove(version lnwire.GossipVersion, chanid uint64) { + delete(c.channels, channelCacheKey{ + version: version, + chanID: chanid, + }) } diff --git a/graph/db/channel_cache_test.go b/graph/db/channel_cache_test.go index 27ff65432..6ecbed8f3 100644 --- a/graph/db/channel_cache_test.go +++ b/graph/db/channel_cache_test.go @@ -6,6 +6,7 @@ import ( "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/lightningnetwork/lnd/graph/db/models" + "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing/route" ) @@ -14,37 +15,39 @@ import ( func TestChannelCache(t *testing.T) { const cacheSize = 100 + v := lnwire.GossipVersion1 + // Create a new channel cache with the configured max size. c := newChannelCache(cacheSize) // As a sanity check, assert that querying the empty cache does not // return an entry. - _, ok := c.get(0) + _, ok := c.get(v, 0) if ok { t.Fatalf("channel cache should be empty") } // Now, fill up the cache entirely. for i := uint64(0); i < cacheSize; i++ { - c.insert(i, channelForInt(i)) + c.insert(v, i, channelForInt(i)) } // Assert that the cache has all of the entries just inserted, since no // eviction should occur until we try to surpass the max size. - assertHasChanEntries(t, c, 0, cacheSize) + assertHasChanEntries(t, c, v, 0, cacheSize) // Now, insert a new element that causes the cache to evict an element. - c.insert(cacheSize, channelForInt(cacheSize)) + c.insert(v, cacheSize, channelForInt(cacheSize)) // Assert that the cache has this last entry, as the cache should evict // some prior element and not the newly inserted one. - assertHasChanEntries(t, c, cacheSize, cacheSize) + assertHasChanEntries(t, c, v, cacheSize, cacheSize) // Iterate over all inserted elements and construct a set of the evicted // elements. evicted := make(map[uint64]struct{}) for i := uint64(0); i < cacheSize+1; i++ { - _, ok := c.get(i) + _, ok := c.get(v, i) if !ok { evicted[i] = struct{}{} } @@ -58,35 +61,37 @@ func TestChannelCache(t *testing.T) { // Remove the highest item which initially caused the eviction and // reinsert the element that was evicted prior. - c.remove(cacheSize) + c.remove(v, cacheSize) for i := range evicted { - c.insert(i, channelForInt(i)) + c.insert(v, i, channelForInt(i)) } // Since the removal created an extra slot, the last insertion should // not have caused an eviction and the entries for all channels in the // original set that filled the cache should be present. - assertHasChanEntries(t, c, 0, cacheSize) + assertHasChanEntries(t, c, v, 0, cacheSize) // Finally, reinsert the existing set back into the cache and test that // the cache still has all the entries. If the randomized eviction were // happening on inserts for existing cache items, we expect this to fail // with high probability. for i := uint64(0); i < cacheSize; i++ { - c.insert(i, channelForInt(i)) + c.insert(v, i, channelForInt(i)) } - assertHasChanEntries(t, c, 0, cacheSize) + assertHasChanEntries(t, c, v, 0, cacheSize) } // assertHasEntries queries the edge cache for all channels in the range [start, // end), asserting that they exist and their value matches the entry produced by // entryForInt. -func assertHasChanEntries(t *testing.T, c *channelCache, start, end uint64) { +func assertHasChanEntries(t *testing.T, c *channelCache, + v lnwire.GossipVersion, start, end uint64) { + t.Helper() for i := start; i < end; i++ { - entry, ok := c.get(i) + entry, ok := c.get(v, i) if !ok { t.Fatalf("channel cache should contain chan %d", i) } diff --git a/graph/db/graph.go b/graph/db/graph.go index baee3eccb..283f7d536 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -163,28 +163,42 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error { log.Info("Populating in-memory channel graph, this might take a " + "while...") - err := c.db.ForEachNodeCacheable(ctx, func(node route.Vertex, - features *lnwire.FeatureVector) error { + for _, v := range []lnwire.GossipVersion{ + gossipV1, gossipV2, + } { + // TODO(elle): If we have both v1 and v2 entries for the same + // node/channel, prefer v2 when merging. + err := c.db.ForEachNodeCacheable(ctx, v, + func(node route.Vertex, + features *lnwire.FeatureVector) error { - c.graphCache.AddNodeFeatures(node, features) + c.graphCache.AddNodeFeatures(node, features) - return nil - }, func() {}) - if err != nil { - return err - } + return nil + }, func() {}, + ) + if err != nil && !errors.Is( + err, ErrVersionNotSupportedForKVDB, + ) { - err = c.db.ForEachChannelCacheable( - func(info *models.CachedEdgeInfo, - policy1, policy2 *models.CachedEdgePolicy) error { + return err + } - c.graphCache.AddChannel(info, policy1, policy2) + err = c.db.ForEachChannelCacheable( + v, func(info *models.CachedEdgeInfo, + policy1, + policy2 *models.CachedEdgePolicy) error { - return nil - }, func() {}, - ) - if err != nil { - return err + c.graphCache.AddChannel(info, policy1, policy2) + + return nil + }, func() {}, + ) + if err != nil && + !errors.Is(err, ErrVersionNotSupportedForKVDB) { + + return err + } } log.Infof("Finished populating in-memory channel graph (took %v, %s)", @@ -210,7 +224,10 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex, return c.graphCache.ForEachChannel(node, cb) } - return c.db.ForEachNodeDirectedChannel(node, cb, reset) + // TODO(elle): once the no-cache path needs to support + // pathfinding across gossip versions, this should iterate + // across all versions rather than defaulting to v1. + return c.db.ForEachNodeDirectedChannel(gossipV1, node, cb, reset) } // FetchNodeFeatures returns the features of the given node. If no features are @@ -332,7 +349,9 @@ func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error { if c.graphCache != nil { // We need to add the channel back into our graph cache, // otherwise we won't use it for path finding. - infos, err := c.db.FetchChanInfos([]uint64{chanID}) + infos, err := c.db.FetchChanInfos( + lnwire.GossipVersion1, []uint64{chanID}, + ) if err != nil { return err } @@ -587,14 +606,6 @@ func (c *ChannelGraph) UpdateEdgePolicy(ctx context.Context, return nil } -// ForEachSourceNodeChannel iterates through all channels of the source node. -func (c *ChannelGraph) ForEachSourceNodeChannel(ctx context.Context, - cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.Node) error, reset func()) error { - - return c.db.ForEachSourceNodeChannel(ctx, cb, reset) -} - // ForEachNodeChannel iterates through all channels of the given node. func (c *ChannelGraph) ForEachNodeChannel(ctx context.Context, v lnwire.GossipVersion, nodePub route.Vertex, @@ -614,10 +625,10 @@ func (c *ChannelGraph) ForEachNode(ctx context.Context, // ForEachNodeCacheable iterates through all stored vertices/nodes in the graph. func (c *ChannelGraph) ForEachNodeCacheable(ctx context.Context, - cb func(route.Vertex, *lnwire.FeatureVector) error, - reset func()) error { + v lnwire.GossipVersion, cb func(route.Vertex, + *lnwire.FeatureVector) error, reset func()) error { - return c.db.ForEachNodeCacheable(ctx, cb, reset) + return c.db.ForEachNodeCacheable(ctx, v, cb, reset) } // NodeUpdatesInHorizon returns all known lightning nodes with updates in the @@ -650,17 +661,11 @@ func (c *ChannelGraph) ForEachChannel(ctx context.Context, return c.db.ForEachChannel(ctx, v, cb, reset) } -// ForEachChannelCacheable iterates through all channel edges for the cache. -func (c *ChannelGraph) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, - reset func()) error { - - return c.db.ForEachChannelCacheable(cb, reset) -} - // DisabledChannelIDs returns the channel ids of disabled channels. -func (c *ChannelGraph) DisabledChannelIDs() ([]uint64, error) { - return c.db.DisabledChannelIDs() +func (c *ChannelGraph) DisabledChannelIDs(v lnwire.GossipVersion) ( + []uint64, error) { + + return c.db.DisabledChannelIDs(v) } // HasV1ChannelEdge returns true if the database knows of a channel edge. @@ -684,14 +689,11 @@ func (c *ChannelGraph) AddEdgeProof(chanID lnwire.ShortChannelID, return c.db.AddEdgeProof(chanID, proof) } -// ChannelID attempts to lookup the 8-byte compact channel ID. -func (c *ChannelGraph) ChannelID(chanPoint *wire.OutPoint) (uint64, error) { - return c.db.ChannelID(chanPoint) -} - // HighestChanID returns the "highest" known channel ID in the channel graph. -func (c *ChannelGraph) HighestChanID(ctx context.Context) (uint64, error) { - return c.db.HighestChanID(ctx) +func (c *ChannelGraph) HighestChanID(ctx context.Context, + v lnwire.GossipVersion) (uint64, error) { + + return c.db.HighestChanID(ctx, v) } // ChanUpdatesInHorizon returns all known channel edges with updates in the @@ -710,8 +712,10 @@ func (c *ChannelGraph) FilterChannelRange(startHeight, endHeight uint32, } // FetchChanInfos returns the set of channel edges for the passed channel IDs. -func (c *ChannelGraph) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) { - return c.db.FetchChanInfos(chanIDs) +func (c *ChannelGraph) FetchChanInfos(v lnwire.GossipVersion, + chanIDs []uint64) ([]ChannelEdge, error) { + + return c.db.FetchChanInfos(v, chanIDs) } // FetchChannelEdgesByOutpoint attempts to lookup directed edges by funding @@ -906,6 +910,14 @@ func (c *VersionedGraph) HasChannelEdge(chanID uint64) (bool, bool, error) { return c.db.HasChannelEdge(c.v, chanID) } +// ForEachSourceNodeChannel iterates through all channels of the source node. +func (c *VersionedGraph) ForEachSourceNodeChannel(ctx context.Context, + cb func(chanPoint wire.OutPoint, havePolicy bool, + otherNode *models.Node) error, reset func()) error { + + return c.db.ForEachSourceNodeChannel(ctx, c.v, cb, reset) +} + // ForEachNodeChannel iterates through all channels of the given node. func (c *VersionedGraph) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, @@ -923,6 +935,44 @@ func (c *VersionedGraph) ForEachChannel(ctx context.Context, return c.db.ForEachChannel(ctx, c.v, cb, reset) } +// ForEachNodeCacheable iterates through all stored vertices/nodes in the graph. +func (c *VersionedGraph) ForEachNodeCacheable(ctx context.Context, + cb func(route.Vertex, *lnwire.FeatureVector) error, + reset func()) error { + + return c.db.ForEachNodeCacheable(ctx, c.v, cb, reset) +} + +// ForEachChannelCacheable iterates through all channel edges for the cache. +func (c *VersionedGraph) ForEachChannelCacheable( + cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error, reset func()) error { + + return c.db.ForEachChannelCacheable(c.v, cb, reset) +} + +// DisabledChannelIDs returns the channel ids of disabled channels. +func (c *VersionedGraph) DisabledChannelIDs() ([]uint64, error) { + return c.db.DisabledChannelIDs(c.v) +} + +// FetchChanInfos returns the set of channel edges for the passed channel IDs. +func (c *VersionedGraph) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, + error) { + + return c.db.FetchChanInfos(c.v, chanIDs) +} + +// HighestChanID returns the "highest" known channel ID in the channel graph. +func (c *VersionedGraph) HighestChanID(ctx context.Context) (uint64, error) { + return c.db.HighestChanID(ctx, c.v) +} + +// ChannelID attempts to lookup the 8-byte compact channel ID. +func (c *VersionedGraph) ChannelID(chanPoint *wire.OutPoint) (uint64, error) { + return c.db.ChannelID(c.v, chanPoint) +} + // IsPublicNode determines whether the node is seen as public in the graph. func (c *VersionedGraph) IsPublicNode(pubKey [33]byte) (bool, error) { return c.db.IsPublicNode(c.v, pubKey) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 2edfc064a..0398031a7 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -154,6 +154,18 @@ var versionedTests = []versionedTest{ name: "incomplete channel policies", test: testIncompleteChannelPolicies, }, + { + name: "add channel edge shell nodes", + test: testAddChannelEdgeShellNodes, + }, + { + name: "for each source node channel", + test: testForEachSourceNodeChannel, + }, + { + name: "graph traversal cacheable", + test: testGraphTraversalCacheable, + }, { name: "partial node", test: testPartialNode, @@ -166,6 +178,34 @@ var versionedTests = []versionedTest{ name: "node is public empty channel signature", test: testIsPublicNodeEmptyChannelSignature, }, + { + name: "edge info updates", + test: testEdgeInfoUpdates, + }, + { + name: "batched update edge policy", + test: testBatchedUpdateEdgePolicy, + }, + { + name: "disabled channel ids", + test: testDisabledChannelIDs, + }, + { + name: "batched add channel edge", + test: testBatchedAddChannelEdge, + }, + { + name: "graph cache for each node channel", + test: testGraphCacheForEachNodeChannel, + }, + { + name: "highest chan id", + test: testHighestChanID, + }, + { + name: "fetch chan infos", + test: testFetchChanInfos, + }, } // TestVersionedDBs runs various tests against both v1 and v2 versioned @@ -1003,7 +1043,8 @@ func assertEdgeInfoEqual(t *testing.T, e1 *models.ChannelEdgeInfo, } } -func createChannelEdge(node1, node2 *models.Node) (*models.ChannelEdgeInfo, +func createChannelEdge(node1, node2 *models.Node, + v lnwire.GossipVersion) (*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) { var ( @@ -1038,82 +1079,149 @@ func createChannelEdge(node1, node2 *models.Node) (*models.ChannelEdgeInfo, 3, 3, 3, 3, 3, } - proof := models.NewV1ChannelAuthProof( - testSig.Serialize(), - testSig.Serialize(), - testSig.Serialize(), - testSig.Serialize(), + var ( + edgeInfo *models.ChannelEdgeInfo + edge1 *models.ChannelEdgePolicy + edge2 *models.ChannelEdgePolicy ) - edgeInfo, _ := models.NewV1Channel( - chanID, *chaincfg.MainNetParams.GenesisHash, node1Key, node2Key, - &models.ChannelV1Fields{ - BitcoinKey1Bytes: node1Key, - BitcoinKey2Bytes: node2Key, - ExtraOpaqueData: extraData, - }, - models.WithChanProof(proof), - models.WithChannelPoint(outpoint), - models.WithCapacity(1000), - ) + switch v { + case gossipV1: + proof := models.NewV1ChannelAuthProof( + testSig.Serialize(), + testSig.Serialize(), + testSig.Serialize(), + testSig.Serialize(), + ) - edge1 := &models.ChannelEdgePolicy{ - Version: lnwire.GossipVersion1, - SigBytes: testSig.Serialize(), - ChannelID: chanID, - LastUpdate: nextUpdateTime(), - SecondPeer: false, - MessageFlags: 1, - ChannelFlags: 0, - TimeLockDelta: 99, - MinHTLC: 2342135, - MaxHTLC: 13928598, - FeeBaseMSat: 4352345, - FeeProportionalMillionths: 3452352, - ToNode: secondNode, - ExtraOpaqueData: []byte{1, 0}, - } - edge2 := &models.ChannelEdgePolicy{ - Version: lnwire.GossipVersion1, - SigBytes: testSig.Serialize(), - ChannelID: chanID, - SecondPeer: true, - LastUpdate: nextUpdateTime(), - MessageFlags: 1, - ChannelFlags: 1, - TimeLockDelta: 99, - MinHTLC: 2342135, - MaxHTLC: 13928598, - FeeBaseMSat: 4352345, - FeeProportionalMillionths: 90392423, - ToNode: firstNode, - ExtraOpaqueData: []byte{1, 0}, + edgeInfo, _ = models.NewV1Channel( + chanID, *chaincfg.MainNetParams.GenesisHash, + node1Key, node2Key, &models.ChannelV1Fields{ + BitcoinKey1Bytes: node1Key, + BitcoinKey2Bytes: node2Key, + ExtraOpaqueData: extraData, + }, + models.WithChanProof(proof), + models.WithChannelPoint(outpoint), + models.WithCapacity(1000), + ) + + edge1 = &models.ChannelEdgePolicy{ + Version: lnwire.GossipVersion1, + SigBytes: testSig.Serialize(), + ChannelID: chanID, + LastUpdate: nextUpdateTime(), + MessageFlags: 1, + ChannelFlags: 0, + TimeLockDelta: 99, + MinHTLC: 2342135, + MaxHTLC: 13928598, + FeeBaseMSat: 4352345, + FeeProportionalMillionths: 3452352, + ToNode: secondNode, + ExtraOpaqueData: []byte{1, 0}, + } + edge2 = &models.ChannelEdgePolicy{ + Version: lnwire.GossipVersion1, + SigBytes: testSig.Serialize(), + ChannelID: chanID, + LastUpdate: nextUpdateTime(), + MessageFlags: 1, + ChannelFlags: 1, + TimeLockDelta: 99, + MinHTLC: 2342135, + MaxHTLC: 13928598, + FeeBaseMSat: 4352345, + FeeProportionalMillionths: 90392423, + ToNode: firstNode, + ExtraOpaqueData: []byte{1, 0}, + } + + case gossipV2: + var merkleRoot chainhash.Hash + copy(merkleRoot[:], bytes.Repeat([]byte{0xaa}, 32)) + + fundingScript := []byte{0x00, 0x20} + fundingScript = append( + fundingScript, bytes.Repeat([]byte{0xbb}, 32)..., + ) + + proof := models.NewV2ChannelAuthProof(testSig.Serialize()) + + edgeInfo, _ = models.NewV2Channel( + chanID, *chaincfg.MainNetParams.GenesisHash, + node1Key, node2Key, &models.ChannelV2Fields{ + BitcoinKey1Bytes: fn.Some(node1Key), + BitcoinKey2Bytes: fn.Some(node2Key), + MerkleRootHash: fn.Some(merkleRoot), + FundingScript: fn.Some(fundingScript), + ExtraSignedFields: make(map[uint64][]byte), + }, + models.WithChanProof(proof), + models.WithChannelPoint(outpoint), + models.WithCapacity(1000), + ) + + edge1 = &models.ChannelEdgePolicy{ + Version: lnwire.GossipVersion2, + SigBytes: testSig.Serialize(), + ChannelID: chanID, + LastBlockHeight: nextBlockHeight(), + SecondPeer: false, + DisableFlags: 0, + TimeLockDelta: 99, + MinHTLC: 2342135, + MaxHTLC: 13928598, + FeeBaseMSat: 4352345, + FeeProportionalMillionths: 3452352, + ToNode: secondNode, + ExtraSignedFields: map[uint64][]byte{ + 100: {0x1, 0x2}, + }, + } + edge2 = &models.ChannelEdgePolicy{ + Version: lnwire.GossipVersion2, + SigBytes: testSig.Serialize(), + ChannelID: chanID, + LastBlockHeight: nextBlockHeight(), + SecondPeer: true, + DisableFlags: 0, + TimeLockDelta: 99, + MinHTLC: 2342135, + MaxHTLC: 13928598, + FeeBaseMSat: 4352345, + FeeProportionalMillionths: 90392423, + ToNode: firstNode, + ExtraSignedFields: map[uint64][]byte{ + 101: {0x3, 0x4}, + }, + } } return edgeInfo, edge1, edge2 } -func TestEdgeInfoUpdates(t *testing.T) { +func testEdgeInfoUpdates(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // We'd like to test the update of edges inserted into the database, so // we create two vertexes to connect. - node1 := createTestVertex(t, lnwire.GossipVersion1) + node1 := createTestVertex(t, v) if err := graph.AddNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } - assertNodeInCache(t, graph, node1, testFeatures) - node2 := createTestVertex(t, lnwire.GossipVersion1) + assertNodeInCache(t, graph.ChannelGraph, node1, testFeatures) + node2 := createTestVertex(t, v) if err := graph.AddNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } - assertNodeInCache(t, graph, node2, testFeatures) + assertNodeInCache(t, graph.ChannelGraph, node2, testFeatures) // Create an edge and add it to the db. - edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) + edgeInfo, edge1, edge2 := createChannelEdge(node1, node2, v) // Make sure inserting the policy at this point, before the edge info // is added, will fail. @@ -1125,7 +1233,7 @@ func TestEdgeInfoUpdates(t *testing.T) { if err := graph.AddChannelEdge(ctx, edgeInfo); err != nil { t.Fatalf("unable to create channel edge: %v", err) } - assertEdgeWithNoPoliciesInCache(t, graph, edgeInfo) + assertEdgeWithNoPoliciesInCache(t, graph.ChannelGraph, edgeInfo) chanID := edgeInfo.ChannelID outpoint := edgeInfo.ChannelPoint @@ -1135,17 +1243,19 @@ func TestEdgeInfoUpdates(t *testing.T) { if err := graph.UpdateEdgePolicy(ctx, edge1); err != nil { t.Fatalf("unable to update edge: %v", err) } - assertEdgeWithPolicyInCache(t, graph, edgeInfo, edge1, true) + assertEdgeWithPolicyInCache( + t, graph.ChannelGraph, edgeInfo, edge1, true, + ) if err := graph.UpdateEdgePolicy(ctx, edge2); err != nil { t.Fatalf("unable to update edge: %v", err) } - assertEdgeWithPolicyInCache(t, graph, edgeInfo, edge2, false) + assertEdgeWithPolicyInCache( + t, graph.ChannelGraph, edgeInfo, edge2, false, + ) // Check for existence of the edge within the database, it should be // found. - found, isZombie, err := graph.HasChannelEdge( - lnwire.GossipVersion1, chanID, - ) + found, isZombie, err := graph.HasChannelEdge(chanID) require.NoError(t, err, "unable to query for edge") if !found { t.Fatalf("graph should have of inserted edge") @@ -1605,16 +1715,16 @@ func testAddEdgeProof(t *testing.T, v lnwire.GossipVersion) { require.NotNil(t, dbEdge2.AuthProof) } -// TestForEachSourceNodeChannel tests that the ForEachSourceNodeChannel +// testForEachSourceNodeChannel tests that the ForEachSourceNodeChannel // correctly iterates through the channels of the set source node. -func TestForEachSourceNodeChannel(t *testing.T) { +func testForEachSourceNodeChannel(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // Create a source node (A) and set it as such in the DB. - nodeA := createTestVertex(t, lnwire.GossipVersion1) + nodeA := createTestVertex(t, v) require.NoError(t, graph.SetSourceNode(ctx, nodeA)) // Now, create a few more nodes (B, C, D) along with some channels @@ -1630,32 +1740,40 @@ func TestForEachSourceNodeChannel(t *testing.T) { // outgoing policy but for the A-C channel, we will set only an incoming // policy. - nodeB := createTestVertex(t, lnwire.GossipVersion1) - nodeC := createTestVertex(t, lnwire.GossipVersion1) - nodeD := createTestVertex(t, lnwire.GossipVersion1) + nodeB := createTestVertex(t, v) + nodeC := createTestVertex(t, v) + nodeD := createTestVertex(t, v) - abEdge, abPolicy1, abPolicy2 := createChannelEdge(nodeA, nodeB) + abEdge, _ := createEdge(v, 100, 0, 0, 0, nodeA, nodeB) require.NoError(t, graph.AddChannelEdge(ctx, abEdge)) - acEdge, acPolicy1, acPolicy2 := createChannelEdge(nodeA, nodeC) + acEdge, _ := createEdge(v, 200, 0, 0, 1, nodeA, nodeC) require.NoError(t, graph.AddChannelEdge(ctx, acEdge)) - bdEdge, _, _ := createChannelEdge(nodeB, nodeD) + bdEdge, _ := createEdge(v, 300, 0, 0, 2, nodeB, nodeD) require.NoError(t, graph.AddChannelEdge(ctx, bdEdge)) - // Figure out which of the policies returned above are node A's so that - // we know which to persist. - // - // First, set the outgoing policy for the A-B channel. - abPolicyAOutgoing := abPolicy1 - if !bytes.Equal(abPolicy1.ToNode[:], nodeB.PubKeyBytes[:]) { - abPolicyAOutgoing = abPolicy2 + newPolicy := func(edge *models.ChannelEdgeInfo, fromNode, + toNode route.Vertex) *models.ChannelEdgePolicy { + + isNode1 := bytes.Equal(fromNode[:], edge.NodeKey1Bytes[:]) + policy := newEdgePolicy( + v, edge.ChannelID, nextUpdateTime().Unix(), isNode1, + ) + policy.ToNode = toNode + policy.SigBytes = testSig.Serialize() + + return policy } + + // First, set the outgoing policy for the A-B channel. + abPolicyAOutgoing := newPolicy( + abEdge, nodeA.PubKeyBytes, nodeB.PubKeyBytes, + ) require.NoError(t, graph.UpdateEdgePolicy(ctx, abPolicyAOutgoing)) // Now, set the incoming policy for the A-C channel. - acPolicyAIncoming := acPolicy1 - if !bytes.Equal(acPolicy1.ToNode[:], nodeA.PubKeyBytes[:]) { - acPolicyAIncoming = acPolicy2 - } + acPolicyAIncoming := newPolicy( + acEdge, nodeC.PubKeyBytes, nodeA.PubKeyBytes, + ) require.NoError(t, graph.UpdateEdgePolicy(ctx, acPolicyAIncoming)) type sourceNodeChan struct { @@ -1677,21 +1795,24 @@ func TestForEachSourceNodeChannel(t *testing.T) { // Now, we'll use the ForEachSourceNodeChannel and assert that it // returns the expected data in the call-back. - err := graph.ForEachSourceNodeChannel(ctx, func(chanPoint wire.OutPoint, - havePolicy bool, otherNode *models.Node) error { + err := graph.ForEachSourceNodeChannel( + ctx, func(chanPoint wire.OutPoint, havePolicy bool, + otherNode *models.Node) error { - require.Contains(t, expectedSrcChans, chanPoint) - expected := expectedSrcChans[chanPoint] + require.Contains(t, expectedSrcChans, chanPoint) + expected := expectedSrcChans[chanPoint] - require.Equal( - t, expected.otherNode[:], otherNode.PubKeyBytes[:], - ) - require.Equal(t, expected.havePolicy, havePolicy) + require.Equal( + t, expected.otherNode[:], + otherNode.PubKeyBytes[:], + ) + require.Equal(t, expected.havePolicy, havePolicy) - delete(expectedSrcChans, chanPoint) + delete(expectedSrcChans, chanPoint) - return nil - }, func() {}) + return nil + }, func() {}, + ) require.NoError(t, err) require.Empty(t, expectedSrcChans) } @@ -1707,7 +1828,9 @@ func TestGraphTraversal(t *testing.T) { // graph. And we'll create 5 channels between each node pair. const numNodes = 20 const numChannels = 5 - chanIndex, nodeList := fillTestGraph(t, graph, numNodes, numChannels) + chanIndex, nodeList := fillTestGraph( + t, graph, numNodes, numChannels, lnwire.GossipVersion1, + ) // Make an index of the node list for easy look up below. nodeIndex := make(map[route.Vertex]struct{}) @@ -1794,46 +1917,43 @@ func TestGraphTraversal(t *testing.T) { require.Equal(t, numChannels, numNodeChans) } -// TestGraphTraversalCacheable tests that the memory optimized node traversal is +// testGraphTraversalCacheable tests that the memory optimized node traversal is // working correctly. -func TestGraphTraversalCacheable(t *testing.T) { +func testGraphTraversalCacheable(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // We'd like to test some of the graph traversal capabilities within // the DB, so we'll create a series of fake nodes to insert into the // graph. And we'll create 5 channels between the first two nodes. const numNodes = 20 const numChannels = 5 - chanIndex, _ := fillTestGraph(t, graph, numNodes, numChannels) + chanIndex, nodeList := fillTestGraph( + t, graph.ChannelGraph, numNodes, numChannels, v, + ) - // Create a map of all nodes with the iteration we know works (because - // it is tested in another test). + // Create a map of all nodes with the nodes we just inserted. nodeMap := make(map[route.Vertex]struct{}) - err := graph.ForEachNode(ctx, func(n *models.Node) error { - nodeMap[n.PubKeyBytes] = struct{}{} - - return nil - }, func() {}) - require.NoError(t, err) + for _, node := range nodeList { + nodeMap[node.PubKeyBytes] = struct{}{} + } require.Len(t, nodeMap, numNodes) // Iterate through all the known channels within the graph DB by // iterating over each node, once again if the map is empty that // indicates that all edges have properly been reached. var nodes []route.Vertex - err = graph.ForEachNodeCacheable(ctx, func(node route.Vertex, - features *lnwire.FeatureVector) error { + err := graph.ForEachNodeCacheable(ctx, + func(node route.Vertex, features *lnwire.FeatureVector) error { + delete(nodeMap, node) + nodes = append(nodes, node) - delete(nodeMap, node) - nodes = append(nodes, node) - - return nil - }, func() { - nodes = nil - }) + return nil + }, func() { + nodes = nil + }) require.NoError(t, err) require.Len(t, nodeMap, 0) @@ -1848,7 +1968,7 @@ func TestGraphTraversalCacheable(t *testing.T) { for _, node := range nodes { // Query the ChannelGraph which uses the cache to iterate // through the channels for each node. - err = graph.ForEachNodeDirectedChannel( + err = graph.ChannelGraph.ForEachNodeDirectedChannel( node, func(d *DirectedChannel) error { delete(chanIndex, d.ChannelID) return nil @@ -1858,7 +1978,7 @@ func TestGraphTraversalCacheable(t *testing.T) { // Now skip the cache and query the DB directly. err = graph.db.ForEachNodeDirectedChannel( - node, func(d *DirectedChannel) error { + v, node, func(d *DirectedChannel) error { delete(chanIndex2, d.ChannelID) return nil }, func() {}, @@ -1879,7 +1999,9 @@ func TestGraphCacheTraversal(t *testing.T) { // graph. And we'll create 5 channels between each node pair. const numNodes = 20 const numChannels = 5 - chanIndex, nodeList := fillTestGraph(t, graph, numNodes, numChannels) + chanIndex, nodeList := fillTestGraph( + t, graph, numNodes, numChannels, lnwire.GossipVersion1, + ) // Iterate through all the known channels within the graph DB, once // again if the map is empty that indicates that all edges have @@ -1922,20 +2044,21 @@ func TestGraphCacheTraversal(t *testing.T) { require.Equal(t, numChannels*2*(numNodes-1), numNodeChans) } -// fillTestGraph fills the graph with a given number of nodes and create a given -// number of channels between each node. +// fillTestGraph fills the graph with nodes and channels using the requested +// gossip version. func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, - numChannels int) (map[uint64]struct{}, []*models.Node) { + numChannels int, v lnwire.GossipVersion) (map[uint64]struct{}, + []*models.Node) { ctx := t.Context() nodes := make([]*models.Node, numNodes) - nodeIndex := map[string]struct{}{} + nodeIndex := map[route.Vertex]struct{}{} for i := 0; i < numNodes; i++ { - node := createTestVertex(t, lnwire.GossipVersion1) + node := createTestVertex(t, v) nodes[i] = node - nodeIndex[node.Alias.UnwrapOr("")] = struct{}{} + nodeIndex[node.PubKeyBytes] = struct{}{} } // Add each of the nodes into the graph, they should be inserted @@ -1946,16 +2069,86 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, // Iterate over each node as returned by the graph, if all nodes are // reached, then the map created above should be empty. - err := graph.ForEachNode(ctx, func(n *models.Node) error { - delete(nodeIndex, n.Alias.UnwrapOr("")) - return nil - }, func() {}) + err := graph.ForEachNodeCacheable(ctx, v, + func(node route.Vertex, _ *lnwire.FeatureVector) error { + delete(nodeIndex, node) + + return nil + }, func() {}) require.NoError(t, err) require.Len(t, nodeIndex, 0) // Create a number of channels between each of the node pairs generated // above. This will result in numChannels*(numNodes-1) channels. chanIndex := map[uint64]struct{}{} + buildEdgeInfo := func(chanID uint64, node1Key, + node2Key route.Vertex, op wire.OutPoint, + version lnwire.GossipVersion) *models.ChannelEdgeInfo { + + switch version { + case gossipV1: + proof := models.NewV1ChannelAuthProof( + testSig.Serialize(), + testSig.Serialize(), + testSig.Serialize(), + testSig.Serialize(), + ) + + edgeInfo, err := models.NewV1Channel( + chanID, *chaincfg.MainNetParams.GenesisHash, + node1Key, node2Key, &models.ChannelV1Fields{ + BitcoinKey1Bytes: node1Key, + BitcoinKey2Bytes: node2Key, + }, + models.WithChanProof(proof), + models.WithChannelPoint(op), + models.WithCapacity(1000), + ) + require.NoError(t, err) + + return edgeInfo + + case gossipV2: + var merkleRoot chainhash.Hash + copy(merkleRoot[:], bytes.Repeat([]byte{0xaa}, 32)) + + fundingScript := []byte{0x00, 0x20} + fundingScript = append( + fundingScript, + bytes.Repeat([]byte{0xbb}, 32)..., + ) + + proof := models.NewV2ChannelAuthProof( + testSig.Serialize(), + ) + + v2Fields := &models.ChannelV2Fields{ + BitcoinKey1Bytes: fn.Some(node1Key), + BitcoinKey2Bytes: fn.Some(node2Key), + MerkleRootHash: fn.Some(merkleRoot), + FundingScript: fn.Some(fundingScript), + ExtraSignedFields: make( + map[uint64][]byte, + ), + } + + edgeInfo, err := models.NewV2Channel( + chanID, *chaincfg.MainNetParams.GenesisHash, + node1Key, node2Key, v2Fields, + models.WithChanProof(proof), + models.WithChannelPoint(op), + models.WithCapacity(1000), + ) + require.NoError(t, err) + + return edgeInfo + } + + require.Failf(t, "unknown gossip version", "%v", version) + + return nil + } + for n := 0; n < numNodes-1; n++ { node1 := nodes[n] node2 := nodes[n+1] @@ -1977,39 +2170,26 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, copy(node1Key[:], node1.PubKeyBytes[:]) copy(node2Key[:], node2.PubKeyBytes[:]) - proof := models.NewV1ChannelAuthProof( - testSig.Serialize(), - testSig.Serialize(), - testSig.Serialize(), - testSig.Serialize(), + edgeInfo := buildEdgeInfo( + chanID, node1Key, node2Key, op, v, ) - - edgeInfo, err := models.NewV1Channel( - chanID, *chaincfg.MainNetParams.GenesisHash, - node1Key, node2Key, &models.ChannelV1Fields{ - BitcoinKey1Bytes: node1Key, - BitcoinKey2Bytes: node2Key, - }, - models.WithChanProof(proof), - models.WithChannelPoint(op), - models.WithCapacity(1000), - ) - require.NoError(t, err) err = graph.AddChannelEdge(ctx, edgeInfo) require.NoError(t, err) // Create and add an edge with random data that points // from node1 -> node2. - edge := randEdgePolicy(chanID) - edge.ChannelFlags = 0 + edge := newEdgePolicy( + v, chanID, prand.Int63(), true, + ) edge.ToNode = node2.PubKeyBytes edge.SigBytes = testSig.Serialize() require.NoError(t, graph.UpdateEdgePolicy(ctx, edge)) // Create another random edge that points from // node2 -> node1 this time. - edge = randEdgePolicy(chanID) - edge.ChannelFlags = 1 + edge = newEdgePolicy( + v, chanID, prand.Int63(), false, + ) edge.ToNode = node1.PubKeyBytes edge.SigBytes = testSig.Serialize() require.NoError(t, graph.UpdateEdgePolicy(ctx, edge)) @@ -2312,41 +2492,32 @@ func TestGraphPruning(t *testing.T) { // TestHighestChanID tests that we're able to properly retrieve the highest // known channel ID in the database. -func TestHighestChanID(t *testing.T) { +func testHighestChanID(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // If we don't yet have any channels in the database, then we should // get a channel ID of zero if we ask for the highest channel ID. bestID, err := graph.HighestChanID(ctx) require.NoError(t, err, "unable to get highest ID") if bestID != 0 { - t.Fatalf("best ID w/ no chan should be zero, is instead: %v", - bestID) + require.Equal(t, uint64(0), bestID) } // Next, we'll insert two channels into the database, with each channel // connecting the same two nodes. - node1 := createTestVertex(t, lnwire.GossipVersion1) - node2 := createTestVertex(t, lnwire.GossipVersion1) + node1 := createTestVertex(t, v) + node2 := createTestVertex(t, v) // The first channel with be at height 10, while the other will be at // height 100. - edge1, _ := createEdge( - lnwire.GossipVersion1, 10, 0, 0, 0, node1, node2, - ) - edge2, chanID2 := createEdge( - lnwire.GossipVersion1, 100, 0, 0, 0, node1, node2, - ) + edge1, _ := createEdge(v, 10, 0, 0, 0, node1, node2) + edge2, chanID2 := createEdge(v, 100, 0, 0, 0, node1, node2) - if err := graph.AddChannelEdge(ctx, edge1); err != nil { - t.Fatalf("unable to create channel edge: %v", err) - } - if err := graph.AddChannelEdge(ctx, edge2); err != nil { - t.Fatalf("unable to create channel edge: %v", err) - } + require.NoError(t, graph.AddChannelEdge(ctx, edge1)) + require.NoError(t, graph.AddChannelEdge(ctx, edge2)) // Now that the edges has been inserted, we'll query for the highest // known channel ID in the database. @@ -2354,25 +2525,17 @@ func TestHighestChanID(t *testing.T) { require.NoError(t, err, "unable to get highest ID") if bestID != chanID2.ToUint64() { - t.Fatalf("expected %v got %v for best chan ID: ", - chanID2.ToUint64(), bestID) + require.Equal(t, chanID2.ToUint64(), bestID) } // If we add another edge, then the current best chan ID should be // updated as well. - edge3, chanID3 := createEdge( - lnwire.GossipVersion1, 1000, 0, 0, 0, node1, node2, - ) - if err := graph.AddChannelEdge(ctx, edge3); err != nil { - t.Fatalf("unable to create channel edge: %v", err) - } + edge3, chanID3 := createEdge(v, 1000, 0, 0, 0, node1, node2) + require.NoError(t, graph.AddChannelEdge(ctx, edge3)) bestID, err = graph.HighestChanID(ctx) require.NoError(t, err, "unable to get highest ID") - if bestID != chanID3.ToUint64() { - t.Fatalf("expected %v got %v for best chan ID: ", - chanID3.ToUint64(), bestID) - } + require.Equal(t, chanID3.ToUint64(), bestID) } // TestChanUpdatesInHorizon tests the we're able to properly retrieve all known @@ -3709,22 +3872,18 @@ func TestFilterChannelRange(t *testing.T) { // TestFetchChanInfos tests that we're able to properly retrieve the full set // of ChannelEdge structs for a given set of short channel ID's. -func TestFetchChanInfos(t *testing.T) { +func testFetchChanInfos(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // We'll first populate our graph with two nodes. All channels created // below will be made between these two nodes. - node1 := createTestVertex(t, lnwire.GossipVersion1) - if err := graph.AddNode(ctx, node1); err != nil { - t.Fatalf("unable to add node: %v", err) - } - node2 := createTestVertex(t, lnwire.GossipVersion1) - if err := graph.AddNode(ctx, node2); err != nil { - t.Fatalf("unable to add node: %v", err) - } + node1 := createTestVertex(t, v) + require.NoError(t, graph.AddNode(ctx, node1)) + node2 := createTestVertex(t, v) + require.NoError(t, graph.AddNode(ctx, node2)) // We'll make 5 test channels, ensuring we keep track of which channel // ID corresponds to a particular ChannelEdge. @@ -3735,36 +3894,35 @@ func TestFetchChanInfos(t *testing.T) { edgeQuery := make([]uint64, 0, numChans) for i := 0; i < numChans; i++ { channel, chanID := createEdge( - lnwire.GossipVersion1, uint32(i*10), 0, 0, 0, - node1, node2, + v, uint32(i*10), 0, 0, 0, node1, node2, ) - if err := graph.AddChannelEdge(ctx, channel); err != nil { - t.Fatalf("unable to create channel edge: %v", err) - } + require.NoError(t, graph.AddChannelEdge(ctx, channel)) updateTime := endTime endTime = updateTime.Add(time.Second * 10) edge1 := newEdgePolicy( - lnwire.GossipVersion1, chanID.ToUint64(), + v, chanID.ToUint64(), updateTime.Unix(), true, ) + if v == lnwire.GossipVersion1 { + edge1.ChannelFlags = 0 + } edge1.ToNode = node2.PubKeyBytes edge1.SigBytes = testSig.Serialize() - if err := graph.UpdateEdgePolicy(ctx, edge1); err != nil { - t.Fatalf("unable to update edge: %v", err) - } + require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1)) edge2 := newEdgePolicy( - lnwire.GossipVersion1, chanID.ToUint64(), + v, chanID.ToUint64(), updateTime.Unix(), false, ) + if v == lnwire.GossipVersion1 { + edge2.ChannelFlags = 1 + } edge2.ToNode = node1.PubKeyBytes edge2.SigBytes = testSig.Serialize() - if err := graph.UpdateEdgePolicy(ctx, edge2); err != nil { - t.Fatalf("unable to update edge: %v", err) - } + require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2)) edges = append(edges, ChannelEdge{ Info: channel, @@ -3782,11 +3940,9 @@ func TestFetchChanInfos(t *testing.T) { // Add an another edge to the query that has been marked as a zombie // edge. The query should also skip this channel. zombieChan, zombieChanID := createEdge( - lnwire.GossipVersion1, 666, 0, 0, 0, node1, node2, + v, 666, 0, 0, 0, node1, node2, ) - if err := graph.AddChannelEdge(ctx, zombieChan); err != nil { - t.Fatalf("unable to create channel edge: %v", err) - } + require.NoError(t, graph.AddChannelEdge(ctx, zombieChan)) err := graph.DeleteChannelEdges(false, true, zombieChan.ChannelID) require.NoError(t, err, "unable to delete and mark edge zombie") edgeQuery = append(edgeQuery, zombieChanID.ToUint64()) @@ -4106,25 +4262,25 @@ func TestPruneGraphNodes(t *testing.T) { require.NotNil(t, err) } -// TestAddChannelEdgeShellNodes tests that when we attempt to add a ChannelEdge +// testAddChannelEdgeShellNodes tests that when we attempt to add a ChannelEdge // to the graph, one or both of the nodes the edge involves aren't found in the // database, then shell edges are created for each node if needed. -func TestAddChannelEdgeShellNodes(t *testing.T) { +func testAddChannelEdgeShellNodes(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1) + graph := NewVersionedGraph(MakeTestGraph(t), v) // To start, we'll create two nodes, and only add one of them to the // channel graph. - node1 := createTestVertex(t, lnwire.GossipVersion1) + node1 := createTestVertex(t, v) require.NoError(t, graph.SetSourceNode(ctx, node1)) - node2 := createTestVertex(t, lnwire.GossipVersion1) + node2 := createTestVertex(t, v) // We'll now create an edge between the two nodes, as a result, node2 // should be inserted into the database as a shell node. edgeInfo, _ := createEdge( - lnwire.GossipVersion1, 100, 0, 0, 0, node1, node2, + v, 100, 0, 0, 0, node1, node2, ) require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo)) @@ -4398,7 +4554,9 @@ func BenchmarkIsPublicNode(b *testing.B) { // Create a graph with a reasonable number of nodes and channels. numNodes := 100 numChans := 4 - _, nodes := fillTestGraph(b, graph, numNodes, numChans) + _, nodes := fillTestGraph( + b, graph, numNodes, numChans, lnwire.GossipVersion1, + ) // Use deterministic random number generator for reproducible results. rng := prand.New(prand.NewSource(42)) @@ -4415,27 +4573,32 @@ func BenchmarkIsPublicNode(b *testing.B) { // TestDisabledChannelIDs ensures that the disabled channels within the // disabledEdgePolicyBucket are managed properly and the list returned from // DisabledChannelIDs is correct. -func TestDisabledChannelIDs(t *testing.T) { +func testDisabledChannelIDs(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // Create first node and add it to the graph. - node1 := createTestVertex(t, lnwire.GossipVersion1) + node1 := createTestVertex(t, v) if err := graph.AddNode(ctx, node1); err != nil { t.Fatalf("unable to add node: %v", err) } // Create second node and add it to the graph. - node2 := createTestVertex(t, lnwire.GossipVersion1) + node2 := createTestVertex(t, v) if err := graph.AddNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } // Adding a new channel edge to the graph. - edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) - node2.LastUpdate = nextUpdateTime() + edgeInfo, edge1, edge2 := createChannelEdge(node1, node2, v) + switch v { + case lnwire.GossipVersion1: + node2.LastUpdate = nextUpdateTime() + case lnwire.GossipVersion2: + node2.LastBlockHeight = nextBlockHeight() + } if err := graph.AddNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -4454,7 +4617,12 @@ func TestDisabledChannelIDs(t *testing.T) { // Add one disabled policy and ensure the channel is still not in the // disabled list. - edge1.ChannelFlags |= lnwire.ChanUpdateDisabled + switch v { + case lnwire.GossipVersion1: + edge1.ChannelFlags |= lnwire.ChanUpdateDisabled + case lnwire.GossipVersion2: + edge1.DisableFlags |= lnwire.ChanUpdateDisableIncoming + } if err := graph.UpdateEdgePolicy(ctx, edge1); err != nil { t.Fatalf("unable to update edge: %v", err) } @@ -4467,7 +4635,12 @@ func TestDisabledChannelIDs(t *testing.T) { // Add second disabled policy and ensure the channel is now in the // disabled list. - edge2.ChannelFlags |= lnwire.ChanUpdateDisabled + switch v { + case lnwire.GossipVersion1: + edge2.ChannelFlags |= lnwire.ChanUpdateDisabled + case lnwire.GossipVersion2: + edge2.DisableFlags |= lnwire.ChanUpdateDisableIncoming + } if err := graph.UpdateEdgePolicy(ctx, edge2); err != nil { t.Fatalf("unable to update edge: %v", err) } @@ -4520,7 +4693,9 @@ func TestEdgePolicyMissingMaxHTLC(t *testing.T) { } node2 := createTestVertex(t, lnwire.GossipVersion1) - edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) + edgeInfo, edge1, edge2 := createChannelEdge( + node1, node2, lnwire.GossipVersion1, + ) if err := graph.AddNode(ctx, node2); err != nil { t.Fatalf("unable to add node: %v", err) } @@ -4659,7 +4834,9 @@ func TestGraphZombieIndex(t *testing.T) { node1, node2 = node2, node1 } - edge, _, _ := createChannelEdge(node1, node2) + edge, _, _ := createChannelEdge( + node1, node2, lnwire.GossipVersion1, + ) require.NoError(t, graph.AddChannelEdge(ctx, edge)) // Since the edge is known the graph and it isn't a zombie, IsZombieEdge @@ -4825,19 +5002,19 @@ func TestComputeFee(t *testing.T) { // TestBatchedAddChannelEdge asserts that BatchedAddChannelEdge properly // executes multiple AddChannelEdge requests in a single txn. -func TestBatchedAddChannelEdge(t *testing.T) { +func testBatchedAddChannelEdge(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) - sourceNode := createTestVertex(t, lnwire.GossipVersion1) + sourceNode := createTestVertex(t, v) require.Nil(t, graph.SetSourceNode(ctx, sourceNode)) // We'd like to test the insertion/deletion of edges, so we create two // vertexes to connect. - node1 := createTestVertex(t, lnwire.GossipVersion1) - node2 := createTestVertex(t, lnwire.GossipVersion1) + node1 := createTestVertex(t, v) + node2 := createTestVertex(t, v) // In addition to the fake vertexes we create some fake channel // identifiers. @@ -4860,23 +5037,20 @@ func TestBatchedAddChannelEdge(t *testing.T) { // Create an edge which has its block height at 156. height := uint32(156) - edgeInfo, _ := createEdge( - lnwire.GossipVersion1, height, 0, 0, 0, node1, node2, - ) + edgeInfo, _ := createEdge(v, height, 0, 0, 0, node1, node2) // Create an edge with block height 157. We give it // maximum values for tx index and position, to make // sure our database range scan get edges from the // entire range. edgeInfo2, _ := createEdge( - lnwire.GossipVersion1, height+1, - math.MaxUint32&0x00ffffff, math.MaxUint16, 1, node1, - node2, + v, height+1, math.MaxUint32&0x00ffffff, math.MaxUint16, 1, + node1, node2, ) // Create a third edge, this with a block height of 155. edgeInfo3, _ := createEdge( - lnwire.GossipVersion1, height-1, 0, 0, 2, node1, node2, + v, height-1, 0, 0, 2, node1, node2, ) edges := []models.ChannelEdgeInfo{*edgeInfo, *edgeInfo2, *edgeInfo3} @@ -4907,21 +5081,21 @@ func TestBatchedAddChannelEdge(t *testing.T) { // TestBatchedUpdateEdgePolicy asserts that BatchedUpdateEdgePolicy properly // executes multiple UpdateEdgePolicy requests in a single txn. -func TestBatchedUpdateEdgePolicy(t *testing.T) { +func testBatchedUpdateEdgePolicy(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // We'd like to test the update of edges inserted into the database, so // we create two vertexes to connect. - node1 := createTestVertex(t, lnwire.GossipVersion1) + node1 := createTestVertex(t, v) require.NoError(t, graph.AddNode(ctx, node1)) - node2 := createTestVertex(t, lnwire.GossipVersion1) + node2 := createTestVertex(t, v) require.NoError(t, graph.AddNode(ctx, node2)) // Create an edge and add it to the db. - edgeInfo, edge1, edge2 := createChannelEdge(node1, node2) + edgeInfo, edge1, edge2 := createChannelEdge(node1, node2, v) // Make sure inserting the policy at this point, before the edge info // is added, will fail. @@ -4966,7 +5140,9 @@ func BenchmarkForEachChannel(b *testing.B) { const numNodes = 100 const numChannels = 4 - _, _ = fillTestGraph(b, graph, numNodes, numChannels) + _, _ = fillTestGraph( + b, graph, numNodes, numChannels, lnwire.GossipVersion1, + ) b.ReportAllocs() b.ResetTimer() @@ -4977,21 +5153,22 @@ func BenchmarkForEachChannel(b *testing.B) { ) var nodes []route.Vertex - err := graph.ForEachNodeCacheable(ctx, func(node route.Vertex, - vector *lnwire.FeatureVector) error { + err := graph.ForEachNodeCacheable( + ctx, lnwire.GossipVersion1, func(node route.Vertex, + vector *lnwire.FeatureVector) error { - nodes = append(nodes, node) + nodes = append(nodes, node) - return nil - }, func() { - nodes = nil - }) + return nil + }, func() { + nodes = nil + }) require.NoError(b, err) for _, n := range nodes { cb := func(info *models.ChannelEdgeInfo, policy *models.ChannelEdgePolicy, - policy2 *models.ChannelEdgePolicy) error { //nolint:ll + policy2 *models.ChannelEdgePolicy) error { // We need to do something with // the data here, otherwise the @@ -5015,23 +5192,25 @@ func BenchmarkForEachChannel(b *testing.B) { // TestGraphCacheForEachNodeChannel tests that the forEachNodeDirectedChannel // method works as expected, and is able to handle nil self edges. -func TestGraphCacheForEachNodeChannel(t *testing.T) { +func testGraphCacheForEachNodeChannel(t *testing.T, + v lnwire.GossipVersion) { + t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // Unset the channel graph cache to simulate the user running with the // option turned off. graph.graphCache = nil - node1 := createTestVertex(t, lnwire.GossipVersion1) + node1 := createTestVertex(t, v) require.NoError(t, graph.AddNode(ctx, node1)) - node2 := createTestVertex(t, lnwire.GossipVersion1) + node2 := createTestVertex(t, v) require.NoError(t, graph.AddNode(ctx, node2)) // Create an edge and add it to the db. - edgeInfo, e1, e2 := createChannelEdge(node1, node2) + edgeInfo, e1, e2 := createChannelEdge(node1, node2, v) // Because of lexigraphical sorting and the usage of random node keys in // this test, we need to determine which edge belongs to node 1 at @@ -5048,7 +5227,8 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { getSingleChannel := func() *DirectedChannel { var ch *DirectedChannel - err := graph.ForEachNodeDirectedChannel(node1.PubKeyBytes, + err := graph.db.ForEachNodeDirectedChannel( + v, node1.PubKeyBytes, func(c *DirectedChannel) error { require.Nil(t, ch) ch = c @@ -5074,6 +5254,12 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { FeeRate: 20, } edge1.InboundFee = fn.Some(inboundFee) + switch v { + case lnwire.GossipVersion1: + edge1.LastUpdate = edge1.LastUpdate.Add(time.Second) + case lnwire.GossipVersion2: + edge1.LastBlockHeight = nextBlockHeight() + } require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1)) edge1 = copyEdgePolicy(edge1) // Avoid read/write race conditions. @@ -5081,22 +5267,29 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { require.NotNil(t, directedChan) require.Equal(t, inboundFee, directedChan.InboundFee) - // Set an invalid inbound fee and check that persistence fails. - edge1.ExtraOpaqueData = []byte{ - 253, 217, 3, 8, 0, - } - // We need to update the timestamp so that we don't hit the DB conflict - // error when we try to update the edge policy. - edge1.LastUpdate = edge1.LastUpdate.Add(time.Second) - require.ErrorIs( - t, graph.UpdateEdgePolicy(ctx, edge1), ErrParsingExtraTLVBytes, - ) + // The below test only applies to v1 since in v2, we would fail TLV + // parsing at the lnwire level when parsing bytes from the wire. + if v == lnwire.GossipVersion1 { + // Set an invalid inbound fee and check that persistence fails. + edge1.ExtraOpaqueData = []byte{ + 253, 217, 3, 8, 0, + } + // We need to update the timestamp so that we don't hit + // the DB conflict error when we try to update the edge + // policy. + edge1.LastUpdate = edge1.LastUpdate.Add(time.Second) + require.ErrorIs( + t, graph.UpdateEdgePolicy(ctx, edge1), + ErrParsingExtraTLVBytes, + ) - // 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, inboundFee, directedChan.InboundFee) + // 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, inboundFee, directedChan.InboundFee) + } } // TestGraphLoading asserts that the cache is properly reconstructed after a @@ -5117,7 +5310,9 @@ func TestGraphLoading(t *testing.T) { // Populate the graph with test data. const numNodes = 100 const numChannels = 4 - _, _ = fillTestGraph(t, graph, numNodes, numChannels) + _, _ = fillTestGraph( + t, graph, numNodes, numChannels, lnwire.GossipVersion1, + ) // Recreate the graph. This should cause the graph cache to be // populated. diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 0936386be..446644077 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -33,7 +33,7 @@ type NodeTraverser interface { type Store interface { //nolint:interfacebloat // ForEachNodeDirectedChannel calls the callback for every channel of // the given node. - ForEachNodeDirectedChannel(nodePub route.Vertex, + ForEachNodeDirectedChannel(v lnwire.GossipVersion, nodePub route.Vertex, cb func(channel *DirectedChannel) error, reset func()) error // FetchNodeFeatures returns the features of the given node. @@ -59,7 +59,7 @@ type Store interface { //nolint:interfacebloat // node, executing the passed callback on each. The call-back is // provided with the channel's outpoint, whether we have a policy for // the channel and the channel peer's node information. - ForEachSourceNodeChannel(ctx context.Context, + ForEachSourceNodeChannel(ctx context.Context, v lnwire.GossipVersion, cb func(chanPoint wire.OutPoint, havePolicy bool, otherNode *models.Node) error, reset func()) error @@ -103,8 +103,9 @@ type Store interface { //nolint:interfacebloat // in the graph, executing the passed callback with each node // encountered. If the callback returns an error, then the transaction // is aborted and the iteration stops early. - ForEachNodeCacheable(ctx context.Context, cb func(route.Vertex, - *lnwire.FeatureVector) error, reset func()) error + ForEachNodeCacheable(ctx context.Context, v lnwire.GossipVersion, + cb func(route.Vertex, *lnwire.FeatureVector) error, + reset func()) error // LookupAlias attempts to return the alias as advertised by the target // node. @@ -182,14 +183,14 @@ type Store interface { //nolint:interfacebloat // // NOTE: this method is like ForEachChannel but fetches only the data // required for the graph cache. - ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, - reset func()) error + ForEachChannelCacheable(v lnwire.GossipVersion, + cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error, reset func()) error // DisabledChannelIDs returns the channel ids of disabled channels. // A channel is disabled when two of the associated ChanelEdgePolicies // have their disabled bit on. - DisabledChannelIDs() ([]uint64, error) + DisabledChannelIDs(v lnwire.GossipVersion) ([]uint64, error) // AddChannelEdge adds a new (undirected, blank) edge to the graph // database. An undirected edge from the two target nodes are created. @@ -238,13 +239,15 @@ type Store interface { //nolint:interfacebloat // ChannelID attempt to lookup the 8-byte compact channel ID which maps // to the passed channel point (outpoint). If the passed channel doesn't // exist within the database, then ErrEdgeNotFound is returned. - ChannelID(chanPoint *wire.OutPoint) (uint64, error) + ChannelID(v lnwire.GossipVersion, + chanPoint *wire.OutPoint) (uint64, error) // HighestChanID returns the "highest" known channel ID in the channel // graph. This represents the "newest" channel from the PoV of the // chain. This method can be used by peers to quickly determine if // they're graphs are in sync. - HighestChanID(ctx context.Context) (uint64, error) + HighestChanID(ctx context.Context, v lnwire.GossipVersion) ( + uint64, error) // ChanUpdatesInHorizon returns all the known channel edges which have // at least one edge that has an update timestamp within the specified @@ -279,7 +282,8 @@ type Store interface { //nolint:interfacebloat // edges that exist at the time of the query. This can be used to // respond to peer queries that are seeking to fill in gaps in their // view of the channel graph. - FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) + FetchChanInfos(v lnwire.GossipVersion, + chanIDs []uint64) ([]ChannelEdge, error) // FetchChannelEdgesByOutpoint attempts to lookup the two directed edges // for the channel identified by the funding outpoint. If the channel diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 22c87f0ee..de9ff401e 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -496,9 +496,13 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo, // // NOTE: this method is like ForEachChannel but fetches only the data required // for the graph cache. -func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, - reset func()) error { +func (c *KVStore) ForEachChannelCacheable(v lnwire.GossipVersion, + cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error, reset func()) error { + + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } return c.db.View(func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) @@ -665,8 +669,13 @@ func (c *KVStore) fetchNodeFeatures(tx kvdb.RTx, // Unknown policies are passed into the callback as nil values. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (c *KVStore) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error, reset func()) error { +func (c *KVStore) ForEachNodeDirectedChannel(v lnwire.GossipVersion, + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + reset func()) error { + + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } return c.forEachNodeDirectedChannel(nil, nodePub, cb, reset) } @@ -763,7 +772,13 @@ func (c *KVStore) ForEachNodeCached(ctx context.Context, withAddrs bool, // DisabledChannelIDs returns the channel ids of disabled channels. // A channel is disabled when two of the associated ChanelEdgePolicies // have their disabled bit on. -func (c *KVStore) DisabledChannelIDs() ([]uint64, error) { +func (c *KVStore) DisabledChannelIDs( + v lnwire.GossipVersion) ([]uint64, error) { + + if v != lnwire.GossipVersion1 { + return nil, ErrVersionNotSupportedForKVDB + } + var disabledChanIDs []uint64 var chanEdgeFound map[uint64]struct{} @@ -874,8 +889,12 @@ func forEachNode(db kvdb.Backend, // callback returns an error, then the transaction is aborted and the iteration // stops early. func (c *KVStore) ForEachNodeCacheable(_ context.Context, - cb func(route.Vertex, *lnwire.FeatureVector) error, - reset func()) error { + v lnwire.GossipVersion, cb func(route.Vertex, + *lnwire.FeatureVector) error, reset func()) error { + + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } traversal := func(tx kvdb.RTx) error { // First grab the nodes bucket which stores the mapping from @@ -1186,7 +1205,9 @@ func (c *KVStore) AddChannelEdge(ctx context.Context, c.rejectCache.remove( lnwire.GossipVersion1, edge.ChannelID, ) - c.chanCache.remove(edge.ChannelID) + c.chanCache.remove( + lnwire.GossipVersion1, edge.ChannelID, + ) return nil } @@ -1588,7 +1609,7 @@ func (c *KVStore) PruneGraph(spentOutputs []*wire.OutPoint, for _, channel := range chansClosed { c.rejectCache.remove(lnwire.GossipVersion1, channel.ChannelID) - c.chanCache.remove(channel.ChannelID) + c.chanCache.remove(lnwire.GossipVersion1, channel.ChannelID) } return chansClosed, prunedNodes, nil @@ -1855,7 +1876,7 @@ func (c *KVStore) DisconnectBlockAtHeight(height uint32) ( for _, channel := range removedChans { c.rejectCache.remove(lnwire.GossipVersion1, channel.ChannelID) - c.chanCache.remove(channel.ChannelID) + c.chanCache.remove(lnwire.GossipVersion1, channel.ChannelID) } return removedChans, nil @@ -1974,7 +1995,7 @@ func (c *KVStore) DeleteChannelEdges(v lnwire.GossipVersion, for _, chanID := range chanIDs { c.rejectCache.remove(lnwire.GossipVersion1, chanID) - c.chanCache.remove(chanID) + c.chanCache.remove(lnwire.GossipVersion1, chanID) } return infos, nil @@ -1983,7 +2004,13 @@ func (c *KVStore) DeleteChannelEdges(v lnwire.GossipVersion, // ChannelID attempt to lookup the 8-byte compact channel ID which maps to the // passed channel point (outpoint). If the passed channel doesn't exist within // the database, then ErrEdgeNotFound is returned. -func (c *KVStore) ChannelID(chanPoint *wire.OutPoint) (uint64, error) { +func (c *KVStore) ChannelID(v lnwire.GossipVersion, + chanPoint *wire.OutPoint) (uint64, error) { + + if v != lnwire.GossipVersion1 { + return 0, ErrVersionNotSupportedForKVDB + } + var chanID uint64 if err := kvdb.View(c.db, func(tx kvdb.RTx) error { var err error @@ -2029,7 +2056,13 @@ func getChanID(tx kvdb.RTx, chanPoint *wire.OutPoint) (uint64, error) { // HighestChanID returns the "highest" known channel ID in the channel graph. // This represents the "newest" channel from the PoV of the chain. This method // can be used by peers to quickly determine if they're graphs are in sync. -func (c *KVStore) HighestChanID(_ context.Context) (uint64, error) { +func (c *KVStore) HighestChanID(_ context.Context, + v lnwire.GossipVersion) (uint64, error) { + + if v != lnwire.GossipVersion1 { + return 0, ErrVersionNotSupportedForKVDB + } + var cid uint64 err := kvdb.View(c.db, func(tx kvdb.RTx) error { @@ -2104,7 +2137,7 @@ func (c *KVStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) { defer c.cacheMu.Unlock() for cid, edge := range edgesToCache { - c.chanCache.insert(cid, edge) + c.chanCache.insert(lnwire.GossipVersion1, cid, edge) } } @@ -2256,7 +2289,10 @@ func (c *KVStore) fetchNextChanUpdateBatch( } // Check cache (we already hold shared read lock). - if channel, ok := c.chanCache.get(chanIDInt); ok { + channel, ok := c.chanCache.get( + lnwire.GossipVersion1, chanIDInt, + ) + if ok { state.edgesSeen[chanIDInt] = struct{}{} batch = append(batch, channel) @@ -2300,7 +2336,7 @@ func (c *KVStore) fetchNextChanUpdateBatch( // Now we have all the information we need to build the // channel edge. - channel := ChannelEdge{ + channel = ChannelEdge{ Info: edgeInfo, Policy1: edge1, Policy2: edge2, @@ -2924,7 +2960,13 @@ func (c *KVStore) FilterChannelRange(startHeight, // skipped and the result will contain only those edges that exist at the time // of the query. This can be used to respond to peer queries that are seeking to // fill in gaps in their view of the channel graph. -func (c *KVStore) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) { +func (c *KVStore) FetchChanInfos(v lnwire.GossipVersion, + chanIDs []uint64) ([]ChannelEdge, error) { + + if v != lnwire.GossipVersion1 { + return nil, ErrVersionNotSupportedForKVDB + } + return c.fetchChanInfos(nil, chanIDs) } @@ -3302,13 +3344,16 @@ func (c *KVStore) updateEdgeCache(e *models.ChannelEdgePolicy, // the entry with the updated policy for the direction that was just // written. If the edge doesn't exist, we'll defer loading the info and // policies and lazily read from disk during the next query. - if channel, ok := c.chanCache.get(e.ChannelID); ok { + channel, ok := c.chanCache.get( + lnwire.GossipVersion1, e.ChannelID, + ) + if ok { if isUpdate1 { channel.Policy1 = e } else { channel.Policy2 = e } - c.chanCache.insert(e.ChannelID, channel) + c.chanCache.insert(lnwire.GossipVersion1, e.ChannelID, channel) } } @@ -3714,8 +3759,13 @@ func (c *KVStore) ForEachNodeChannel(_ context.Context, // channel's outpoint, whether we have a policy for the channel and the channel // peer's node information. func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, - cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.Node) error, reset func()) error { + v lnwire.GossipVersion, cb func(chanPoint wire.OutPoint, + havePolicy bool, otherNode *models.Node) error, + reset func()) error { + + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } return kvdb.View(c.db, func(tx kvdb.RTx) error { nodes := tx.ReadBucket(nodeBucket) @@ -4215,7 +4265,7 @@ func (c *KVStore) MarkEdgeZombie(chanID uint64, } c.rejectCache.remove(lnwire.GossipVersion1, chanID) - c.chanCache.remove(chanID) + c.chanCache.remove(lnwire.GossipVersion1, chanID) return nil } @@ -4284,7 +4334,7 @@ func (c *KVStore) markEdgeLiveUnsafe(tx kvdb.RwTx, chanID uint64) error { } c.rejectCache.remove(lnwire.GossipVersion1, chanID) - c.chanCache.remove(chanID) + c.chanCache.remove(lnwire.GossipVersion1, chanID) return nil } @@ -4450,8 +4500,9 @@ type nodeTraverserSession struct { // node. // // NOTE: Part of the NodeTraverser interface. -func (c *nodeTraverserSession) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error, _ func()) error { +func (c *nodeTraverserSession) ForEachNodeDirectedChannel( + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + _ func()) error { return c.db.forEachNodeDirectedChannel(c.tx, nodePub, cb, func() {}) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 44e2fd75b..a04a70aba 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -33,6 +33,11 @@ import ( "github.com/lightningnetwork/lnd/tor" ) +const ( + gossipV1 = lnwire.GossipVersion1 + gossipV2 = lnwire.GossipVersion2 +) + // SQLQueries is a subset of the sqlc.Querier interface that can be used to // execute queries against the SQL graph tables. // @@ -71,6 +76,8 @@ type SQLQueries interface { GetNodeFeaturesBatch(ctx context.Context, ids []int64) ([]sqlc.GraphNodeFeature, error) GetNodeFeaturesByPubKey(ctx context.Context, arg sqlc.GetNodeFeaturesByPubKeyParams) ([]int32, error) DeleteNodeFeature(ctx context.Context, arg sqlc.DeleteNodeFeatureParams) error + GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error) + GetV2DisabledSCIDs(ctx context.Context) ([][]byte, error) /* Source node queries. @@ -114,7 +121,6 @@ type SQLQueries interface { */ UpsertEdgePolicy(ctx context.Context, arg sqlc.UpsertEdgePolicyParams) (int64, error) GetChannelPolicyByChannelAndNode(ctx context.Context, arg sqlc.GetChannelPolicyByChannelAndNodeParams) (sqlc.GraphChannelPolicy, error) - GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error) UpsertChanPolicyExtraType(ctx context.Context, arg sqlc.UpsertChanPolicyExtraTypeParams) error GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]sqlc.GetChannelPolicyExtraTypesBatchRow, error) @@ -305,7 +311,7 @@ func (s *SQLStore) HasV1Node(ctx context.Context, err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { dbNode, err := db.GetNodeByPubKey( ctx, sqlc.GetNodeByPubKeyParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(gossipV1), PubKey: pubKey[:], }, ) @@ -454,13 +460,26 @@ func (s *SQLStore) FetchNodeFeatures(v lnwire.GossipVersion, // have their disabled bit on. // // NOTE: part of the Store interface. -func (s *SQLStore) DisabledChannelIDs() ([]uint64, error) { +func (s *SQLStore) DisabledChannelIDs( + v lnwire.GossipVersion) ([]uint64, error) { + var ( ctx = context.TODO() chanIDs []uint64 ) err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { - dbChanIDs, err := db.GetV1DisabledSCIDs(ctx) + var ( + dbChanIDs [][]byte + err error + ) + switch v { + case gossipV1: + dbChanIDs, err = db.GetV1DisabledSCIDs(ctx) + case gossipV2: + dbChanIDs, err = db.GetV2DisabledSCIDs(ctx) + default: + return fmt.Errorf("unsupported gossip version: %d", v) + } if err != nil { return fmt.Errorf("unable to fetch disabled "+ "channels: %w", err) @@ -750,7 +769,9 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context, s.rejectCache.remove( edge.Version, edge.ChannelID, ) - s.chanCache.remove(edge.ChannelID) + s.chanCache.remove( + edge.Version, edge.ChannelID, + ) return nil } @@ -765,10 +786,16 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context, // can be used by peers to quickly determine if their graphs are in sync. // // NOTE: This is part of the Store interface. -func (s *SQLStore) HighestChanID(ctx context.Context) (uint64, error) { +func (s *SQLStore) HighestChanID(ctx context.Context, + v lnwire.GossipVersion) (uint64, error) { + var highestChanID uint64 err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { - chanID, err := db.HighestSCID(ctx, int16(lnwire.GossipVersion1)) + if !isKnownGossipVersion(v) { + return fmt.Errorf("unsupported gossip version: %d", v) + } + + chanID, err := db.HighestSCID(ctx, int16(v)) if errors.Is(err, sql.ErrNoRows) { return nil } else if err != nil { @@ -867,11 +894,11 @@ func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy, // during the next query for this edge. if entry, ok := s.rejectCache.get(e.Version, e.ChannelID); ok { switch e.Version { - case lnwire.GossipVersion1: + case gossipV1: updateRejectCacheEntryV1( &entry, isUpdate1, e.LastUpdate, ) - case lnwire.GossipVersion2: + case gossipV2: updateRejectCacheEntryV2( &entry, isUpdate1, e.LastBlockHeight, ) @@ -883,13 +910,13 @@ func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy, // the entry with the updated policy for the direction that was just // written. If the edge doesn't exist, we'll defer loading the info and // policies and lazily read from disk during the next query. - if channel, ok := s.chanCache.get(e.ChannelID); ok { + if channel, ok := s.chanCache.get(e.Version, e.ChannelID); ok { if isUpdate1 { channel.Policy1 = e } else { channel.Policy2 = e } - s.chanCache.insert(e.ChannelID, channel) + s.chanCache.insert(e.Version, e.ChannelID, channel) } } @@ -900,20 +927,19 @@ func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy, // // NOTE: part of the Store interface. func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, - cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.Node) error, reset func()) error { + v lnwire.GossipVersion, cb func(chanPoint wire.OutPoint, + havePolicy bool, otherNode *models.Node) error, + reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { - nodeID, nodePub, err := s.getSourceNode( - ctx, db, lnwire.GossipVersion1, - ) + nodeID, nodePub, err := s.getSourceNode(ctx, db, v) if err != nil { return fmt.Errorf("unable to fetch source node: %w", err) } return forEachNodeChannel( - ctx, db, s.cfg, lnwire.GossipVersion1, nodeID, + ctx, db, s.cfg, v, nodeID, func(info *models.ChannelEdgeInfo, outPolicy *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error { @@ -935,8 +961,8 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, } _, otherNode, err := getNodeByPubKey( - ctx, s.cfg.QueryCfg, db, - lnwire.GossipVersion1, otherNodePub, + ctx, s.cfg.QueryCfg, db, v, + otherNodePub, ) if err != nil { return fmt.Errorf("unable to fetch "+ @@ -982,13 +1008,14 @@ func (s *SQLStore) ForEachNode(ctx context.Context, // Unknown policies are passed into the callback as nil values. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error, reset func()) error { +func (s *SQLStore) ForEachNodeDirectedChannel(v lnwire.GossipVersion, + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + reset func()) error { var ctx = context.TODO() return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { - return forEachNodeDirectedChannel(ctx, db, nodePub, cb) + return forEachNodeDirectedChannel(ctx, db, v, nodePub, cb) }, reset) } @@ -997,12 +1024,12 @@ func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, // callback returns an error, then the transaction is aborted and the iteration // stops early. func (s *SQLStore) ForEachNodeCacheable(ctx context.Context, - cb func(route.Vertex, *lnwire.FeatureVector) error, - reset func()) error { + v lnwire.GossipVersion, cb func(route.Vertex, + *lnwire.FeatureVector) error, reset func()) error { err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeCacheable( - ctx, s.cfg.QueryCfg, db, + ctx, s.cfg.QueryCfg, db, v, func(_ int64, nodePub route.Vertex, features *lnwire.FeatureVector) error { @@ -1120,7 +1147,9 @@ func (s *SQLStore) buildChannelFromRow(ctx context.Context, db SQLQueries, // updateChanCacheBatch updates the channel cache with multiple edges at once. // This method acquires the cache lock only once for the entire batch. -func (s *SQLStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) { +func (s *SQLStore) updateChanCacheBatch(v lnwire.GossipVersion, + edgesToCache map[uint64]ChannelEdge) { + if len(edgesToCache) == 0 { return } @@ -1129,7 +1158,7 @@ func (s *SQLStore) updateChanCacheBatch(edgesToCache map[uint64]ChannelEdge) { defer s.cacheMu.Unlock() for chanID, edge := range edgesToCache { - s.chanCache.insert(chanID, edge) + s.chanCache.insert(v, chanID, edge) } } @@ -1232,6 +1261,7 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time, // Check cache (we already hold // shared read lock). channel, ok := s.chanCache.get( + lnwire.GossipVersion1, chanIDInt, ) if ok { @@ -1287,7 +1317,9 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time, // Update cache after successful batch yield, setting // the cache lock only once for the entire batch. - s.updateChanCacheBatch(edgesToCache) + s.updateChanCacheBatch( + lnwire.GossipVersion1, edgesToCache, + ) edgesToCache = make(map[uint64]ChannelEdge) // If the batch didn't yield anything, then we're done. @@ -1536,12 +1568,16 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, withAddrs bool, // // NOTE: this method is like ForEachChannel but fetches only the data // required for the graph cache. -func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, - reset func()) error { +func (s *SQLStore) ForEachChannelCacheable(v lnwire.GossipVersion, + cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error, reset func()) error { ctx := context.TODO() + if !isKnownGossipVersion(v) { + return fmt.Errorf("unsupported gossip version: %d", v) + } + handleChannel := func(_ context.Context, row sqlc.ListChannelsWithPoliciesForCachePaginatedRow) error { @@ -1585,7 +1621,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, return db.ListChannelsWithPoliciesForCachePaginated( ctx, sqlc.ListChannelsWithPoliciesForCachePaginatedParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), ID: lastID, Limit: limit, }, @@ -1779,7 +1815,7 @@ func (s *SQLStore) MarkEdgeZombie(chanID uint64, } s.rejectCache.remove(lnwire.GossipVersion1, chanID) - s.chanCache.remove(chanID) + s.chanCache.remove(lnwire.GossipVersion1, chanID) return nil } @@ -1828,7 +1864,7 @@ func (s *SQLStore) MarkEdgeLive(chanID uint64) error { } s.rejectCache.remove(lnwire.GossipVersion1, chanID) - s.chanCache.remove(chanID) + s.chanCache.remove(lnwire.GossipVersion1, chanID) return err } @@ -2010,7 +2046,7 @@ func (s *SQLStore) DeleteChannelEdges(v lnwire.GossipVersion, for _, chanID := range chanIDs { s.rejectCache.remove(v, chanID) - s.chanCache.remove(chanID) + s.chanCache.remove(v, chanID) } return edges, nil @@ -2236,7 +2272,7 @@ func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, // We'll query the cache with the shared lock held to allow multiple // readers to access values in the cache concurrently if they exist. s.cacheMu.RLock() - if entry, ok := s.rejectCache.get(lnwire.GossipVersion1, chanID); ok { + if entry, ok := s.rejectCache.get(gossipV1, chanID); ok { s.cacheMu.RUnlock() node1LastUpdate = time.Unix(entry.upd1Time, 0) node2LastUpdate = time.Unix(entry.upd2Time, 0) @@ -2252,7 +2288,7 @@ func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, // The item was not found with the shared lock, so we'll acquire the // exclusive lock and check the cache again in case another method added // the entry to the cache while no lock was held. - if entry, ok := s.rejectCache.get(lnwire.GossipVersion1, chanID); ok { + if entry, ok := s.rejectCache.get(gossipV1, chanID); ok { node1LastUpdate = time.Unix(entry.upd1Time, 0) node2LastUpdate = time.Unix(entry.upd2Time, 0) exists, isZombie = entry.flags.unpack() @@ -2265,7 +2301,7 @@ func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, channel, err := db.GetChannelBySCID( ctx, sqlc.GetChannelBySCIDParams{ Scid: chanIDB, - Version: int16(lnwire.GossipVersion1), + Version: int16(gossipV1), }, ) if errors.Is(err, sql.ErrNoRows) { @@ -2273,7 +2309,7 @@ func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, isZombie, err = db.IsZombieChannel( ctx, sqlc.IsZombieChannelParams{ Scid: chanIDB, - Version: int16(lnwire.GossipVersion1), + Version: int16(gossipV1), }, ) if err != nil { @@ -2290,7 +2326,7 @@ func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, policy1, err := db.GetChannelPolicyByChannelAndNode( ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(gossipV1), ChannelID: channel.ID, NodeID: channel.NodeID1, }, @@ -2304,7 +2340,7 @@ func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, policy2, err := db.GetChannelPolicyByChannelAndNode( ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(gossipV1), ChannelID: channel.ID, NodeID: channel.NodeID2, }, @@ -2324,7 +2360,7 @@ func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, } s.rejectCache.insert( - lnwire.GossipVersion1, chanID, + gossipV1, chanID, newRejectCacheEntryV1( node1LastUpdate, node2LastUpdate, exists, isZombie, @@ -2421,13 +2457,13 @@ func (s *SQLStore) HasChannelEdge(v lnwire.GossipVersion, err) } else if err == nil { switch v { - case lnwire.GossipVersion1: + case gossipV1: if policy1.LastUpdate.Valid { node1LastUpdate = time.Unix( policy1.LastUpdate.Int64, 0, ) } - case lnwire.GossipVersion2: + case gossipV2: if policy1.BlockHeight.Valid { node1Block = uint32( policy1.BlockHeight.Int64, @@ -2448,13 +2484,13 @@ func (s *SQLStore) HasChannelEdge(v lnwire.GossipVersion, err) } else if err == nil { switch v { - case lnwire.GossipVersion1: + case gossipV1: if policy2.LastUpdate.Valid { node2LastUpdate = time.Unix( policy2.LastUpdate.Int64, 0, ) } - case lnwire.GossipVersion2: + case gossipV2: if policy2.BlockHeight.Valid { node2Block = uint32( policy2.BlockHeight.Int64, @@ -2472,11 +2508,11 @@ func (s *SQLStore) HasChannelEdge(v lnwire.GossipVersion, var entry rejectCacheEntry switch v { - case lnwire.GossipVersion1: + case gossipV1: entry = newRejectCacheEntryV1( node1LastUpdate, node2LastUpdate, exists, isZombie, ) - case lnwire.GossipVersion2: + case gossipV2: entry = newRejectCacheEntryV2( node1Block, node2Block, exists, isZombie, ) @@ -2491,7 +2527,9 @@ func (s *SQLStore) HasChannelEdge(v lnwire.GossipVersion, // the database, then ErrEdgeNotFound is returned. // // NOTE: part of the Store interface. -func (s *SQLStore) ChannelID(chanPoint *wire.OutPoint) (uint64, error) { +func (s *SQLStore) ChannelID(v lnwire.GossipVersion, + chanPoint *wire.OutPoint) (uint64, error) { + var ( ctx = context.TODO() channelID uint64 @@ -2500,7 +2538,7 @@ func (s *SQLStore) ChannelID(chanPoint *wire.OutPoint) (uint64, error) { chanID, err := db.GetSCIDByOutpoint( ctx, sqlc.GetSCIDByOutpointParams{ Outpoint: chanPoint.String(), - Version: int16(lnwire.GossipVersion1), + Version: int16(v), }, ) if errors.Is(err, sql.ErrNoRows) { @@ -2538,9 +2576,9 @@ func (s *SQLStore) IsPublicNode(v lnwire.GossipVersion, pubKey [33]byte) (bool, err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { var err error switch v { - case lnwire.GossipVersion1: + case gossipV1: isPublic, err = db.IsPublicV1Node(ctx, pubKey[:]) - case lnwire.GossipVersion2: + case gossipV2: isPublic, err = db.IsPublicV2Node(ctx, pubKey[:]) } @@ -2561,12 +2599,18 @@ func (s *SQLStore) IsPublicNode(v lnwire.GossipVersion, pubKey [33]byte) (bool, // fill in gaps in their view of the channel graph. // // NOTE: part of the Store interface. -func (s *SQLStore) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) { +func (s *SQLStore) FetchChanInfos(v lnwire.GossipVersion, + chanIDs []uint64) ([]ChannelEdge, error) { + var ( ctx = context.TODO() edges = make(map[uint64]ChannelEdge) ) err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + if !isKnownGossipVersion(v) { + return fmt.Errorf("unsupported gossip version: %d", v) + } + // First, collect all channel rows. var channelRows []sqlc.GetChannelsBySCIDWithPoliciesRow chanCallBack := func(ctx context.Context, @@ -2577,7 +2621,7 @@ func (s *SQLStore) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) { } err := s.forEachChanWithPoliciesInSCIDList( - ctx, db, lnwire.GossipVersion1, chanCallBack, chanIDs, + ctx, db, v, chanCallBack, chanIDs, ) if err != nil { return err @@ -2901,7 +2945,7 @@ func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint, for _, channel := range closedChans { s.rejectCache.remove(channel.Version, channel.ChannelID) - s.chanCache.remove(channel.ChannelID) + s.chanCache.remove(channel.Version, channel.ChannelID) } return closedChans, prunedNodes, nil @@ -3170,7 +3214,7 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) ( s.cacheMu.Lock() for _, channel := range removedChans { s.rejectCache.remove(channel.Version, channel.ChannelID) - s.chanCache.remove(channel.ChannelID) + s.chanCache.remove(channel.Version, channel.ChannelID) } s.cacheMu.Unlock() @@ -3199,7 +3243,7 @@ func (s *SQLStore) AddEdgeProof(scid lnwire.ShortChannelID, err error ) switch proof.Version { - case lnwire.GossipVersion1: + case gossipV1: res, err = db.AddV1ChannelProof( ctx, sqlc.AddV1ChannelProofParams{ Scid: scidBytes, @@ -3210,7 +3254,7 @@ func (s *SQLStore) AddEdgeProof(scid lnwire.ShortChannelID, }, ) - case lnwire.GossipVersion2: + case gossipV2: res, err = db.AddV2ChannelProof( ctx, sqlc.AddV2ChannelProofParams{ Scid: scidBytes, @@ -3332,12 +3376,15 @@ func newSQLNodeTraverser(db SQLQueries, // node. // // NOTE: Part of the NodeTraverser interface. -func (s *sqlNodeTraverser) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error, _ func()) error { +func (s *sqlNodeTraverser) ForEachNodeDirectedChannel( + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + _ func()) error { ctx := context.TODO() - return forEachNodeDirectedChannel(ctx, s.db, nodePub, cb) + return forEachNodeDirectedChannel( + ctx, s.db, lnwire.GossipVersion1, nodePub, cb, + ) } // FetchNodeFeatures returns the features of the given node. If the node is @@ -3357,7 +3404,8 @@ func (s *sqlNodeTraverser) FetchNodeFeatures(nodePub route.Vertex) ( // channel and its incoming policy. If the node is not found, no error is // returned. func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, - nodePub route.Vertex, cb func(channel *DirectedChannel) error) error { + v lnwire.GossipVersion, nodePub route.Vertex, + cb func(channel *DirectedChannel) error) error { toNodeCallback := func() route.Vertex { return nodePub @@ -3365,7 +3413,7 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, dbID, err := db.GetNodeIDByPubKey( ctx, sqlc.GetNodeIDByPubKeyParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), PubKey: nodePub[:], }, ) @@ -3377,7 +3425,7 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, rows, err := db.ListChannelsByNodeID( ctx, sqlc.ListChannelsByNodeIDParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), NodeID1: dbID, }, ) @@ -3464,11 +3512,12 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, return nil } -// forEachNodeCacheable fetches all V1 node IDs and pub keys from the database, +// forEachNodeCacheable fetches all node IDs and pub keys from the database, // and executes the provided callback for each node. It does so via pagination // along with batch loading of the node feature bits. func forEachNodeCacheable(ctx context.Context, cfg *sqldb.QueryConfig, - db SQLQueries, processNode func(nodeID int64, nodePub route.Vertex, + db SQLQueries, v lnwire.GossipVersion, + processNode func(nodeID int64, nodePub route.Vertex, features *lnwire.FeatureVector) error) error { handleNode := func(_ context.Context, @@ -3493,7 +3542,7 @@ func forEachNodeCacheable(ctx context.Context, cfg *sqldb.QueryConfig, return db.ListNodeIDsAndPubKeys( ctx, sqlc.ListNodeIDsAndPubKeysParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), ID: lastID, Limit: limit, }, @@ -3685,7 +3734,8 @@ func updateChanEdgePolicy(ctx context.Context, tx SQLQueries, Signature: edge.SigBytes, } - if version == lnwire.GossipVersion1 { + switch version { + case gossipV1: params.LastUpdate = sqldb.SQLInt64(edge.LastUpdate.Unix()) params.Disabled = sql.NullBool{ Valid: true, @@ -3695,7 +3745,7 @@ func updateChanEdgePolicy(ctx context.Context, tx SQLQueries, Valid: edge.MessageFlags.HasMaxHtlc(), Int64: int64(edge.MaxHTLC), } - } else { + case gossipV2: params.BlockHeight = sqldb.SQLInt64( int64(edge.LastBlockHeight), ) @@ -3712,7 +3762,7 @@ func updateChanEdgePolicy(ctx context.Context, tx SQLQueries, // Convert the flat extra opaque data into a map of TLV types to // values. extra := edge.ExtraSignedFields - if version == lnwire.GossipVersion1 { + if version == gossipV1 { extra, err = marshalExtraOpaqueData(edge.ExtraOpaqueData) if err != nil { return node1Pub, node2Pub, false, fmt.Errorf( @@ -3788,9 +3838,9 @@ func buildNode(ctx context.Context, cfg *sqldb.QueryConfig, db SQLQueries, // and supported. func isKnownGossipVersion(v lnwire.GossipVersion) bool { switch v { - case lnwire.GossipVersion1: + case gossipV1: return true - case lnwire.GossipVersion2: + case gossipV2: return true default: return false @@ -3864,7 +3914,7 @@ func buildNodeWithBatchData(dbNode sqlc.GraphNode, // Use preloaded extra fields. if extraFields, exists := batchData.extraFields[dbNode.ID]; exists { - if v == lnwire.GossipVersion1 { + if v == gossipV1 { records := lnwire.CustomRecords(extraFields) recs, err := records.Serialize() if err != nil { @@ -3957,7 +4007,7 @@ func upsertNodeAncillaryData(ctx context.Context, db SQLQueries, // Convert the flat extra opaque data into a map of TLV types to // values. extra := node.ExtraSignedFields - if node.Version == lnwire.GossipVersion1 { + if node.Version == gossipV1 { extra, err = marshalExtraOpaqueData(node.ExtraOpaqueData) if err != nil { return fmt.Errorf("unable to marshal extra opaque "+ @@ -3996,10 +4046,10 @@ func populateNodeParams(node *models.Node, }) switch node.Version { - case lnwire.GossipVersion1: + case gossipV1: lastUpdate = sqldb.SQLInt64(node.LastUpdate.Unix()) - case lnwire.GossipVersion2: + case gossipV2: lastBlockHeight = sqldb.SQLInt64(int64(node.LastBlockHeight)) default: @@ -4576,7 +4626,7 @@ func insertChannel(ctx context.Context, db SQLQueries, // Finally, insert any extra TLV fields in the channel announcement. extra := edge.ExtraSignedFields - if v == lnwire.GossipVersion1 { + if v == gossipV1 { extra, err = marshalExtraOpaqueData(edge.ExtraOpaqueData) if err != nil { return fmt.Errorf("unable to marshal extra opaque "+ @@ -4722,7 +4772,7 @@ func buildEdgeInfoWithBatchData(chain chainhash.Hash, // Build the appropriate channel based on version. var channel *models.ChannelEdgeInfo switch v { - case lnwire.GossipVersion1: + case gossipV1: // For v1, serialize extras into ExtraOpaqueData. recs, err := lnwire.CustomRecords(extras).Serialize() if err != nil { @@ -4771,7 +4821,7 @@ func buildEdgeInfoWithBatchData(chain chainhash.Hash, ) } - case lnwire.GossipVersion2: + case gossipV2: v2Fields := &models.ChannelV2Fields{ ExtraSignedFields: extras, } @@ -4871,16 +4921,15 @@ func getAndBuildChanPolicies(ctx context.Context, cfg *sqldb.QueryConfig, return nil, nil, nil } - // TODO(elle): update to support v2 policies. if dbPol1 != nil && - lnwire.GossipVersion(dbPol1.Version) != lnwire.GossipVersion1 { + !isKnownGossipVersion(lnwire.GossipVersion(dbPol1.Version)) { return nil, nil, fmt.Errorf("unsupported policy1 version: %d", dbPol1.Version) } if dbPol2 != nil && - lnwire.GossipVersion(dbPol2.Version) != lnwire.GossipVersion1 { + !isKnownGossipVersion(lnwire.GossipVersion(dbPol2.Version)) { return nil, nil, fmt.Errorf("unsupported policy2 version: %d", dbPol2.Version) @@ -4985,7 +5034,7 @@ func buildChanPolicy(isNode1 bool, dbPolicy sqlc.GraphChannelPolicy, InboundFee: inboundFee, } - if p.Version == lnwire.GossipVersion1 { + if p.Version != gossipV2 { recs, err := lnwire.CustomRecords(extras).Serialize() if err != nil { return nil, fmt.Errorf("unable to serialize extra "+ @@ -5033,7 +5082,7 @@ func extractChannelPolicies(row any) (*sqlc.GraphChannelPolicy, case sqlc.ListChannelsWithPoliciesForCachePaginatedRow: if r.Policy1Timelock.Valid { policy1 = &sqlc.GraphChannelPolicy{ - Version: int16(lnwire.GossipVersion1), + Version: r.Policy1Version.Int16, Timelock: r.Policy1Timelock.Int32, FeePpm: r.Policy1FeePpm.Int64, BaseFeeMsat: r.Policy1BaseFeeMsat.Int64, @@ -5050,7 +5099,7 @@ func extractChannelPolicies(row any) (*sqlc.GraphChannelPolicy, } if r.Policy2Timelock.Valid { policy2 = &sqlc.GraphChannelPolicy{ - Version: int16(lnwire.GossipVersion1), + Version: r.Policy2Version.Int16, Timelock: r.Policy2Timelock.Int32, FeePpm: r.Policy2FeePpm.Int64, BaseFeeMsat: r.Policy2BaseFeeMsat.Int64, @@ -6352,7 +6401,7 @@ func handleZombieMarking(ctx context.Context, db SQLQueries, if strictZombiePruning { // TODO(elle): update for V2 last update times. - if v != lnwire.GossipVersion1 { + if v != gossipV1 { return fmt.Errorf("strict zombie pruning only "+ "supported for gossip v1, got %v", v) } diff --git a/routing/pathfind.go b/routing/pathfind.go index fab8015dd..ddaade60c 100644 --- a/routing/pathfind.go +++ b/routing/pathfind.go @@ -1461,8 +1461,7 @@ func processNodeForBlindedPath(g Graph, node route.Vertex, // Now, iterate over the node's channels in search for paths to this // node that can be used for blinded paths err = g.ForEachNodeDirectedChannel( - node, - func(channel *graphdb.DirectedChannel) error { + node, func(channel *graphdb.DirectedChannel) error { // Keep track of how many incoming channels this node // has. We only use a node as an introduction node if it // has channels other than the one that lead us to it. diff --git a/server.go b/server.go index ad97261ed..91b662463 100644 --- a/server.go +++ b/server.go @@ -1042,7 +1042,9 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, return nil, fmt.Errorf("can't create router: %w", err) } - chanSeries := discovery.NewChanSeries(s.graphDB) + chanSeries := discovery.NewChanSeries( + graphdb.NewVersionedGraph(s.graphDB, lnwire.GossipVersion1), + ) gossipMessageStore, err := discovery.NewMessageStore(dbs.ChanStateDB) if err != nil { return nil, err @@ -3576,7 +3578,10 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { graphAddrs[pubStr] = n return nil } - err = s.graphDB.ForEachSourceNodeChannel( + + // TODO(elle): for now, we only fetch our V1 channels. This should be + // updated to fetch channels across all versions. + err = s.v1Graph.ForEachSourceNodeChannel( ctx, forEachSrcNodeChan, func() { clear(graphAddrs) }, diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index aa2a04671..6293ef249 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -2410,6 +2410,41 @@ func (q *Queries) GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error) { return items, nil } +const getV2DisabledSCIDs = `-- name: GetV2DisabledSCIDs :many +SELECT c.scid +FROM graph_channels c + JOIN graph_channel_policies cp ON cp.channel_id = c.id +WHERE COALESCE(cp.disable_flags, 0) != 0 +AND c.version = 2 +GROUP BY c.scid +HAVING COUNT(*) > 1 +` + +// NOTE: this is V2 specific since V2 uses a disable flag +// bit vector instead of a single boolean. +func (q *Queries) GetV2DisabledSCIDs(ctx context.Context) ([][]byte, error) { + rows, err := q.db.QueryContext(ctx, getV2DisabledSCIDs) + if err != nil { + return nil, err + } + defer rows.Close() + var items [][]byte + for rows.Next() { + var scid []byte + if err := rows.Scan(&scid); err != nil { + return nil, err + } + items = append(items, scid) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getZombieChannel = `-- name: GetZombieChannel :one SELECT scid, version, node_key_1, node_key_2 FROM graph_zombie_channels @@ -3299,6 +3334,7 @@ SELECT n2.pub_key AS node2_pubkey, -- Node 1 policy + cp1.version AS policy1_version, cp1.timelock AS policy_1_timelock, cp1.fee_ppm AS policy_1_fee_ppm, cp1.base_fee_msat AS policy_1_base_fee_msat, @@ -3313,6 +3349,7 @@ SELECT cp1.disable_flags AS policy1_disable_flags, -- Node 2 policy + cp2.version AS policy2_version, cp2.timelock AS policy_2_timelock, cp2.fee_ppm AS policy_2_fee_ppm, cp2.base_fee_msat AS policy_2_base_fee_msat, @@ -3350,6 +3387,7 @@ type ListChannelsWithPoliciesForCachePaginatedRow struct { Capacity sql.NullInt64 Node1Pubkey []byte Node2Pubkey []byte + Policy1Version sql.NullInt16 Policy1Timelock sql.NullInt32 Policy1FeePpm sql.NullInt64 Policy1BaseFeeMsat sql.NullInt64 @@ -3362,6 +3400,7 @@ type ListChannelsWithPoliciesForCachePaginatedRow struct { Policy1ChannelFlags sql.NullInt16 Policy1BlockHeight sql.NullInt64 Policy1DisableFlags sql.NullInt16 + Policy2Version sql.NullInt16 Policy2Timelock sql.NullInt32 Policy2FeePpm sql.NullInt64 Policy2BaseFeeMsat sql.NullInt64 @@ -3391,6 +3430,7 @@ func (q *Queries) ListChannelsWithPoliciesForCachePaginated(ctx context.Context, &i.Capacity, &i.Node1Pubkey, &i.Node2Pubkey, + &i.Policy1Version, &i.Policy1Timelock, &i.Policy1FeePpm, &i.Policy1BaseFeeMsat, @@ -3403,6 +3443,7 @@ func (q *Queries) ListChannelsWithPoliciesForCachePaginated(ctx context.Context, &i.Policy1ChannelFlags, &i.Policy1BlockHeight, &i.Policy1DisableFlags, + &i.Policy2Version, &i.Policy2Timelock, &i.Policy2FeePpm, &i.Policy2BaseFeeMsat, diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index d26f845d0..8948955b7 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -83,6 +83,9 @@ type Querier interface { // structure will have a more complex disabled bit vector // and so the query for V2 may differ. GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error) + // NOTE: this is V2 specific since V2 uses a disable flag + // bit vector instead of a single boolean. + GetV2DisabledSCIDs(ctx context.Context) ([][]byte, error) GetZombieChannel(ctx context.Context, arg GetZombieChannelParams) (GraphZombieChannel, error) GetZombieChannelsSCIDs(ctx context.Context, arg GetZombieChannelsSCIDsParams) ([]GraphZombieChannel, error) HighestSCID(ctx context.Context, version int16) ([]byte, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index a4a42a1e3..ec44bc015 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -834,6 +834,7 @@ SELECT n2.pub_key AS node2_pubkey, -- Node 1 policy + cp1.version AS policy1_version, cp1.timelock AS policy_1_timelock, cp1.fee_ppm AS policy_1_fee_ppm, cp1.base_fee_msat AS policy_1_base_fee_msat, @@ -848,6 +849,7 @@ SELECT cp1.disable_flags AS policy1_disable_flags, -- Node 2 policy + cp2.version AS policy2_version, cp2.timelock AS policy_2_timelock, cp2.fee_ppm AS policy_2_fee_ppm, cp2.base_fee_msat AS policy_2_base_fee_msat, @@ -1066,6 +1068,17 @@ AND c.version = 1 GROUP BY c.scid HAVING COUNT(*) > 1; +-- name: GetV2DisabledSCIDs :many +SELECT c.scid +FROM graph_channels c + JOIN graph_channel_policies cp ON cp.channel_id = c.id +-- NOTE: this is V2 specific since V2 uses a disable flag +-- bit vector instead of a single boolean. +WHERE COALESCE(cp.disable_flags, 0) != 0 +AND c.version = 2 +GROUP BY c.scid +HAVING COUNT(*) > 1; + -- name: DeleteChannelPolicyExtraTypes :exec DELETE FROM graph_channel_policy_extra_types WHERE channel_policy_id = $1;