GlcoinCore/lnd
1
0
Fork 0
mirror of https://github.com/lightningnetwork/lnd.git synced 2026-08-13 12:32:48 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 7
lnd/graph/db/graph_test.go
Oli 8047149c6a
multi: upgrade to btcd v2 modules 
Migrate all btcd dependencies to the new per-package v2 modules (wire/v2,
txscript/v2, chaincfg/v2, chainhash/v2, btcutil/v2, psbt/v2, btcec/v2)
introduced by btcd v0.26.0, and pin the tagged ecosystem versions:
btcwallet v0.17.0, neutrino v0.18.0 and lightning-onion v1.4.0.

The bulk of the import rewrite was produced by the scripted diff from
https://github.com/btcsuite/btcd/pull/2547 (followed by 'make rpc'). The
address symbols that moved out of btcutil into the new address package
are imported as btcaddr where a local "address" variable would otherwise
shadow them. The go.mod/go.sum updates and the remaining manual
compilation fixes are folded into this single commit so it builds on its
own (the migration was previously split into a reproducible scripted-diff
plus follow-ups, intended to be squashed on merge).
2026-06-24 10:58:36 -07:00

6772 lines
187 KiB
Go
Raw Permalink Blame History

package graphdb
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"image/color"
"math"
prand "math/rand"
"net"
"sync"
"testing"
"time"
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/btcec/v2/ecdsa"
"github.com/btcsuite/btcd/btcec/v2/schnorr"
"github.com/btcsuite/btcd/btcutil/v2"
"github.com/btcsuite/btcd/chaincfg/v2"
"github.com/btcsuite/btcd/chainhash/v2"
"github.com/btcsuite/btcd/wire/v2"
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/graph/db/models"
"github.com/lightningnetwork/lnd/input"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lntest/wait"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/routing/route"
"github.com/stretchr/testify/require"
"golang.org/x/exp/rand"
)
var (
testAddr = &net.TCPAddr{IP: (net.IP)([]byte{0xA, 0x0, 0x0, 0x1}),
Port: 9000}
anotherAddr, _ = net.ResolveTCPAddr("tcp",
"[2001:db8:85a3:0:0:8a2e:370:7334]:80")
testAddrs = []net.Addr{testAddr, anotherAddr}
testRBytes, _ = hex.DecodeString("8ce2bc69281ce27da07e6683571319d18" +
"e949ddfa2965fb6caa1bf0314f882d7")
testSBytes, _ = hex.DecodeString("299105481d63e0f4bc2a88121167221b6" +
"700d72a0ead154c03be696a292d24ae")
testRScalar = new(btcec.ModNScalar)
testSScalar = new(btcec.ModNScalar)
_ = testRScalar.SetByteSlice(testRBytes)
_ = testSScalar.SetByteSlice(testSBytes)
testSig = ecdsa.NewSignature(testRScalar, testSScalar)
testFeatures = lnwire.NewFeatureVector(
lnwire.NewRawFeatureVector(lnwire.GossipQueriesRequired),
lnwire.Features,
)
testPub = route.Vertex{2, 202, 4}
key = [chainhash.HashSize]byte{
0x81, 0xb6, 0x37, 0xd8, 0xfc, 0xd2, 0xc6, 0xda,
0x68, 0x59, 0xe6, 0x96, 0x31, 0x13, 0xa1, 0x17,
0xd, 0xe7, 0x93, 0xe4, 0xb7, 0x25, 0xb8, 0x4d,
0x1e, 0xb, 0x4c, 0xf9, 0x9e, 0xc5, 0x8c, 0xe9,
}
rev = [chainhash.HashSize]byte{
0x51, 0xb6, 0x37, 0xd8, 0xfc, 0xd2, 0xc6, 0xda,
0x48, 0x59, 0xe6, 0x96, 0x31, 0x13, 0xa1, 0x17,
0x2d, 0xe7, 0x93, 0xe4,
}
)
func createNode(t testing.TB, v lnwire.GossipVersion,
priv *btcec.PrivateKey) *models.Node {
pubKey := route.NewVertex(priv.PubKey())
switch v {
case lnwire.GossipVersion1:
return models.NewV1Node(
pubKey, &models.NodeV1Fields{
LastUpdate: nextUpdateTime(),
Color: color.RGBA{1, 2, 3, 0},
Alias: "kek" + hex.EncodeToString(
pubKey[:],
),
Addresses: testAddrs,
Features: testFeatures.RawFeatureVector,
AuthSigBytes: testSig.Serialize(),
},
)
case lnwire.GossipVersion2:
return models.NewV2Node(
pubKey, &models.NodeV2Fields{
Signature: testSig.Serialize(),
LastBlockHeight: nextBlockHeight(),
Color: fn.Some(
color.RGBA{1, 2, 3, 0},
),
Alias: fn.Some(
"kek" + hex.EncodeToString(pubKey[:]),
),
Features: testFeatures.
RawFeatureVector,
Addresses: testAddrs,
},
)
}
t.Fatalf("unknown gossip version: %v", v)
return nil
}
func createTestVertex(t testing.TB, v lnwire.GossipVersion) *models.Node {
t.Helper()
priv, err := btcec.NewPrivateKey()
require.NoError(t, err)
return createNode(t, v, priv)
}
type versionedTest struct {
name string
test func(t *testing.T, v lnwire.GossipVersion)
}
var versionedTests = []versionedTest{
{
name: "node crud",
test: testNodeInsertionAndDeletion,
},
{
name: "source node",
test: testSourceNode,
},
{
name: "alias lookup",
test: testAliasLookup,
},
{
name: "add edge proof",
test: testAddEdgeProof,
},
{
name: "edge insertion deletion",
test: testEdgeInsertionDeletion,
},
{
name: "edge policy crud",
test: testEdgePolicyCRUD,
},
{
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,
},
{
name: "node is public",
test: testNodeIsPublic,
},
{
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,
},
{
name: "channel view",
test: testChannelView,
},
{
name: "channel view taproot v1 round trip",
test: testChannelViewTaprootV1RoundTrip,
},
{
name: "node pruning update index deletion",
test: testNodePruningUpdateIndexDeletion,
},
{
name: "lightning node sig verification",
test: testLightningNodeSigVerification,
},
{
name: "graph zombie index",
test: testGraphZombieIndex,
},
{
name: "disconnect block at height",
test: testDisconnectBlockAtHeight,
},
{
name: "filter known chan ids zombie revival",
test: testFilterKnownChanIDsZombieRevival,
},
{
name: "filter known chan ids",
test: testFilterKnownChanIDs,
},
{
name: "fetch zombie edge versioning",
test: testFetchZombieEdgeVersioning,
},
}
// TestVersionedDBs runs various tests against both v1 and v2 versioned
// backends.
func TestVersionedDBs(t *testing.T) {
t.Parallel()
// Run all v1 tests.
for _, vt := range versionedTests {
t.Run(vt.name+"/v1", func(t *testing.T) {
vt.test(t, lnwire.GossipVersion1)
})
if !isSQLDB {
continue
}
t.Run(vt.name+"/v2", func(t *testing.T) {
vt.test(t, lnwire.GossipVersion2)
})
}
}
// testNodeInsertionAndDeletion tests the CRUD operations for a Node.
func testNodeInsertionAndDeletion(t *testing.T, v lnwire.GossipVersion) {
nodeWithAddrs := func(addrs []net.Addr) *models.Node {
return models.NewV1Node(
testPub, &models.NodeV1Fields{
AuthSigBytes: testSig.Serialize(),
LastUpdate: nextUpdateTime(),
Color: color.RGBA{1, 2, 3, 0},
Alias: "kek",
Features: testFeatures.RawFeatureVector,
Addresses: addrs,
ExtraOpaqueData: []byte{1, 1, 1, 2, 2, 2, 2},
},
)
}
if v == lnwire.GossipVersion2 {
nodeWithAddrs = func(addrs []net.Addr) *models.Node {
return models.NewV2Node(
testPub, &models.NodeV2Fields{
Signature: testSig.Serialize(),
LastBlockHeight: nextBlockHeight(),
Color: fn.Some(
color.RGBA{1, 2, 3, 0},
),
Alias: fn.Some("kek"),
Features: testFeatures.
RawFeatureVector,
Addresses: addrs,
ExtraSignedFields: map[uint64][]byte{
20: {0x1, 0x2, 0x3},
21: {0x4, 0x5, 0x6, 0x7},
},
},
)
}
}
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// First, insert the node into the graph DB. This should succeed
// without any errors.
node := nodeWithAddrs(testAddrs)
require.NoError(t, graph.AddNode(ctx, node))
assertNodeInCache(t, graph.ChannelGraph, node, testFeatures)
// Our AddNode implementation uses the batcher meaning that it is
// possible that two updates for the same node announcement may be
// processed in the same batch. So to avoid the conflict error (since we
// require at the DB level that the new timestamp is strictly
// greater than the previous one), we need to gracefully handle the
// case where the exact same node announcement is added twice.
require.NoError(t, graph.AddNode(ctx, node))
// Next, fetch the node from the database to ensure everything was
// serialized properly.
dbNode, err := graph.FetchNode(ctx, testPub)
require.NoError(t, err, "unable to locate node")
exists, err := graph.HasNode(ctx, dbNode.PubKeyBytes)
require.NoError(t, err)
require.True(t, exists)
// The two nodes should match exactly!
compareNodes(t, node, dbNode)
// Check that the node's features are fetched correctly. This check
// will use the graph cache to fetch the features.
features, err := graph.FetchNodeFeatures(ctx, node.PubKeyBytes)
require.NoError(t, err)
require.Equal(t, testFeatures, features)
// Check that the node's features are fetched correctly. This check
// will check the database directly.
features, err = graph.FetchNodeFeatures(ctx, node.PubKeyBytes)
require.NoError(t, err)
require.Equal(t, testFeatures, features)
// Next, delete the node from the graph, this should purge all data
// related to the node.
require.NoError(t, graph.DeleteNode(ctx, testPub))
assertNodeNotInCache(t, graph.ChannelGraph, testPub)
// Attempting to delete the node again should return an error since
// the node is no longer known.
require.ErrorIs(
t, graph.DeleteNode(ctx, testPub),
ErrGraphNodeNotFound,
)
// Finally, attempt to fetch the node again. This should fail as the
// node should have been deleted from the database.
_, err = graph.FetchNode(ctx, testPub)
require.ErrorIs(t, err, ErrGraphNodeNotFound)
// Now, we'll specifically test the updating of addresses of a node
// since the serialisation and persistence of addresses is a bit
// tricky.
pub, err := node.PubKey()
require.NoError(t, err)
// Initially, the node is unknown to the graph and there should be no
// addresses for it.
known, addrs, err := graph.AddrsForNode(ctx, pub)
require.NoError(t, err)
require.False(t, known)
require.Empty(t, addrs)
// Add the node without any addresses.
node = nodeWithAddrs(nil)
require.NoError(t, graph.AddNode(ctx, node))
// Fetch the node and assert the empty addresses.
dbNode, err = graph.FetchNode(ctx, testPub)
require.NoError(t, err)
compareNodes(t, node, dbNode)
known, addrs, err = graph.AddrsForNode(ctx, pub)
require.NoError(t, err)
require.True(t, known)
require.Empty(t, addrs)
// Now, update the node's addresses.
expAddrs := []net.Addr{
// Add 2 IPV4 addresses.
testAddr,
testIPV4Addr,
// Add 2 IPV6 addresses.
testIPV6Addr,
anotherAddr,
// Add one v2 and one v3 onion address.
testOnionV2Addr,
testOnionV3Addr,
// Add a DNS host address.
testDNSAddr,
// Make sure to also test the opaque address type.
testOpaqueAddr,
}
node = nodeWithAddrs(expAddrs)
require.NoError(t, graph.AddNode(ctx, node))
// Fetch the node and assert the updated addresses.
dbNode, err = graph.FetchNode(ctx, testPub)
require.NoError(t, err)
require.Equal(t, expAddrs, dbNode.Addresses)
known, addrs, err = graph.AddrsForNode(ctx, pub)
require.NoError(t, err)
require.True(t, known)
require.EqualValues(t, expAddrs, addrs)
// Now, change the address set a bit: change the order of the
// IPV4 addresses, remove one IPV6 address and remove both onion
// addresses.
expAddrs = []net.Addr{
testIPV4Addr,
testAddr,
testIPV6Addr,
}
node = nodeWithAddrs(expAddrs)
require.NoError(t, graph.AddNode(ctx, node))
// Fetch the node and assert the updated addresses.
dbNode, err = graph.FetchNode(ctx, testPub)
require.NoError(t, err)
require.Equal(t, expAddrs, dbNode.Addresses)
// Finally, update the set to only contain the Tor addresses.
expAddrs = []net.Addr{
testOnionV2Addr,
testOnionV3Addr,
}
node = nodeWithAddrs(expAddrs)
require.NoError(t, graph.AddNode(ctx, node))
// Fetch the node and assert the updated addresses.
dbNode, err = graph.FetchNode(ctx, testPub)
require.NoError(t, err)
require.Equal(t, expAddrs, dbNode.Addresses)
}
// testPartialNode tests that partial/shell nodes are correctly created when
// a channel edge is added referencing nodes we are not yet aware of.
func testPartialNode(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(
MakeTestGraph(t, WithSyncGraphCachePopulation()), v,
)
// To insert a partial node, we need to add a channel edge that has
// node keys for nodes we are not yet aware of.
var node1, node2 models.Node
copy(node1.PubKeyBytes[:], pubKey1Bytes)
copy(node2.PubKeyBytes[:], pubKey2Bytes)
// Create an edge attached to these nodes and add it to the graph.
edgeInfo, _ := createEdge(v, 140, 0, 0, 0, &node1, &node2)
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
// Both of the nodes should now be in both the graph (as partial/shell)
// nodes _and_ the cache should also have an awareness of both nodes.
assertNodeInCache(t, graph.ChannelGraph, &node1, nil)
assertNodeInCache(t, graph.ChannelGraph, &node2, nil)
// Next, fetch the nodes from the database to ensure everything was
// serialized properly.
dbNode1, err := graph.FetchNode(ctx, pubKey1)
require.NoError(t, err)
dbNode2, err := graph.FetchNode(ctx, pubKey2)
require.NoError(t, err)
exists, err := graph.HasNode(ctx, dbNode1.PubKeyBytes)
require.NoError(t, err)
require.True(t, exists)
// The two nodes should match exactly! (with default values for
// LastUpdate and db set to satisfy compareNodes())
expectedNode1 := models.NewShellNode(v, pubKey1)
compareNodes(t, expectedNode1, dbNode1)
exists, err = graph.HasNode(ctx, dbNode2.PubKeyBytes)
require.NoError(t, err)
require.True(t, exists)
// The two nodes should match exactly! (with default values for
// LastUpdate and db set to satisfy compareNodes())
expectedNode2 := models.NewShellNode(v, pubKey2)
compareNodes(t, expectedNode2, dbNode2)
// Next, delete the node from the graph, this should purge all data
// related to the node.
require.NoError(t, graph.DeleteNode(ctx, pubKey1))
assertNodeNotInCache(t, graph.ChannelGraph, testPub)
// Finally, attempt to fetch the node again. This should fail as the
// node should have been deleted from the database.
_, err = graph.FetchNode(ctx, testPub)
require.ErrorIs(t, err, ErrGraphNodeNotFound)
}
// testAliasLookup tests the alias lookup functionality of the graph store.
func testAliasLookup(t *testing.T, v lnwire.GossipVersion) {
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// We'd like to test the alias index within the database, so first
// create a new test node.
testNode := createTestVertex(t, v)
// Add the node to the graph's database, this should also insert an
// entry into the alias index for this node.
require.NoError(t, graph.AddNode(ctx, testNode))
// Next, attempt to lookup the alias. The alias should exactly match
// the one which the test node was assigned.
nodePub, err := testNode.PubKey()
require.NoError(t, err, "unable to generate pubkey")
dbAlias, err := graph.LookupAlias(ctx, nodePub)
require.NoError(t, err, "unable to find alias")
require.Equal(t, testNode.Alias.UnwrapOr(""), dbAlias)
// Ensure that looking up a non-existent alias results in an error.
node := createTestVertex(t, v)
nodePub, err = node.PubKey()
require.NoError(t, err, "unable to generate pubkey")
_, err = graph.LookupAlias(ctx, nodePub)
require.ErrorIs(t, err, ErrNodeAliasNotFound)
}
// testSourceNode tests the source node functionality of the graph store.
func testSourceNode(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// We'd like to test the setting/getting of the source node, so we
// first create a fake node to use within the test.
testNode := createTestVertex(t, v)
// Attempt to fetch the source node, this should return an error as the
// source node hasn't yet been set.
_, err := graph.SourceNode(ctx)
require.ErrorIs(t, err, ErrSourceNodeNotSet)
// Set the source node, this should insert the node into the
// database in a special way indicating it's the source node.
require.NoError(t, graph.SetSourceNode(ctx, testNode))
// Retrieve the source node from the database, it should exactly match
// the one we set above.
sourceNode, err := graph.SourceNode(ctx)
require.NoError(t, err, "unable to fetch source node")
compareNodes(t, testNode, sourceNode)
}
// TestSetSourceNodeSameTimestamp tests that SetSourceNode accepts updates
// with the same timestamp. This is necessary because multiple code paths
// (setSelfNode, createNewHiddenService, RPC updates) can race during startup,
// reading the same old timestamp and independently incrementing it to the same
// new value. For our own node, we want parameter changes to persist even with
// timestamp collisions (unlike network gossip where same timestamp means same
// content).
func TestSetSourceNodeSameTimestamp(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
// Create and set the initial source node.
testNode := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.SetSourceNode(ctx, testNode))
// Verify the source node was set correctly.
sourceNode, err := graph.SourceNode(ctx)
require.NoError(t, err)
compareNodes(t, testNode, sourceNode)
// Create a modified version of the node with the same timestamp but
// different parameters (e.g., different alias and color). This
// simulates the race condition where multiple goroutines read the
// same old timestamp, independently increment it, and try to update
// with different changes.
modifiedNode := models.NewV1Node(
testNode.PubKeyBytes, &models.NodeV1Fields{
// Same timestamp.
LastUpdate: testNode.LastUpdate,
// Different alias.
Alias: "different-alias",
Color: color.RGBA{R: 100, G: 200, B: 50, A: 0},
Addresses: testNode.Addresses,
Features: testNode.Features.RawFeatureVector,
AuthSigBytes: testNode.AuthSigBytes,
},
)
// Attempt to set the source node with the same timestamp but
// different parameters. This should now succeed for both SQL and KV
// stores. The SQL store uses UpsertSourceNode which removes the
// strict timestamp constraint, allowing last-write-wins semantics.
require.NoError(t, graph.SetSourceNode(ctx, modifiedNode))
// Verify that the parameter changes actually persisted.
updatedNode, err := graph.SourceNode(ctx)
require.NoError(t, err)
require.Equal(t, "different-alias", updatedNode.Alias.UnwrapOr(""))
require.Equal(
t, color.RGBA{R: 100, G: 200, B: 50, A: 0},
updatedNode.Color.UnwrapOr(color.RGBA{}),
)
require.Equal(t, testNode.LastUpdate, updatedNode.LastUpdate)
}
// testEdgeInsertionDeletion tests the basic CRUD operations for channel edges.
func testEdgeInsertionDeletion(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(
MakeTestGraph(t, WithSyncGraphCachePopulation()), v,
)
// We'd like to test the insertion/deletion of edges, so we create two
// vertexes to connect.
node1 := createTestVertex(t, v)
node2 := createTestVertex(t, v)
// Create a fake channel and add it to the graph.
const (
blockHeight = 1234
txIndex = 1
txPosition = 0
outPointIndex = 9
)
edgeInfo, shortChanID := createEdge(
v, blockHeight, txIndex, txPosition, outPointIndex, node1,
node2,
)
chanID := shortChanID.ToUint64()
outpoint := wire.OutPoint{
Hash: rev,
Index: outPointIndex,
}
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
assertEdgeWithNoPoliciesInCache(t, graph.ChannelGraph, edgeInfo)
// Show that trying to insert the same channel again will return the
// expected error.
err := graph.AddChannelEdge(ctx, edgeInfo)
require.ErrorIs(t, err, ErrEdgeAlreadyExist)
// Ensure that both policies are returned as unknown (nil) and that
// the edge info round-trips correctly.
dbEdge, e1, e2, err := graph.FetchChannelEdgesByID(ctx, chanID)
require.NoError(t, err)
require.Nil(t, e1)
require.Nil(t, e2)
// Verify core fields match.
require.Equal(t, edgeInfo.ChannelID, dbEdge.ChannelID)
require.Equal(t, edgeInfo.Version, dbEdge.Version)
require.Equal(t, edgeInfo.NodeKey1Bytes, dbEdge.NodeKey1Bytes)
require.Equal(t, edgeInfo.NodeKey2Bytes, dbEdge.NodeKey2Bytes)
require.Equal(t, edgeInfo.ChainHash, dbEdge.ChainHash)
require.Equal(t, edgeInfo.ChannelPoint, dbEdge.ChannelPoint)
require.Equal(t, edgeInfo.Capacity, dbEdge.Capacity)
// Verify auth proof round-trips.
require.NotNil(t, dbEdge.AuthProof)
require.Equal(t, edgeInfo.AuthProof.Version, dbEdge.AuthProof.Version)
// Verify version-specific fields.
switch v {
case lnwire.GossipVersion1:
require.Equal(t,
edgeInfo.BitcoinKey1Bytes, dbEdge.BitcoinKey1Bytes,
)
require.Equal(t,
edgeInfo.BitcoinKey2Bytes, dbEdge.BitcoinKey2Bytes,
)
require.Equal(t,
edgeInfo.ExtraOpaqueData, dbEdge.ExtraOpaqueData,
)
case lnwire.GossipVersion2:
require.Equal(t,
edgeInfo.BitcoinKey1Bytes, dbEdge.BitcoinKey1Bytes,
)
require.Equal(t,
edgeInfo.BitcoinKey2Bytes, dbEdge.BitcoinKey2Bytes,
)
require.Equal(t,
edgeInfo.MerkleRootHash, dbEdge.MerkleRootHash,
)
require.Equal(t,
edgeInfo.FundingScript, dbEdge.FundingScript,
)
require.Equal(t,
edgeInfo.ExtraSignedFields, dbEdge.ExtraSignedFields,
)
}
// Also verify fetching by outpoint returns the same data.
dbEdge2, _, _, err := graph.FetchChannelEdgesByOutpoint(
ctx, &outpoint,
)
require.NoError(t, err)
require.Equal(t, dbEdge.ChannelID, dbEdge2.ChannelID)
// Next, attempt to delete the edge from the database, again this
// should proceed without any issues.
require.NoError(t, graph.DeleteChannelEdges(
ctx, false, true, chanID,
))
assertNoEdge(t, graph.ChannelGraph, chanID)
// Ensure that any query attempts to lookup the delete channel edge are
// properly deleted.
_, _, _, err = graph.FetchChannelEdgesByOutpoint(ctx, &outpoint)
require.ErrorIs(t, err, ErrEdgeNotFound)
// Assert that if the edge is a zombie, then FetchChannelEdgesByID
// still returns a populated models.ChannelEdgeInfo as its comment
// description promises.
edge, _, _, err := graph.FetchChannelEdgesByID(ctx, chanID)
require.ErrorIs(t, err, ErrZombieEdge)
require.NotNil(t, edge)
isZombie, _, _, err := graph.IsZombieEdge(ctx, chanID)
require.NoError(t, err)
require.True(t, isZombie)
// Finally, attempt to delete a (now) non-existent edge within the
// database, this should result in an error.
err = graph.DeleteChannelEdges(ctx, false, true, chanID)
require.ErrorIs(t, err, ErrEdgeNotFound)
}
func createEdge(version lnwire.GossipVersion, height, txIndex uint32,
txPosition uint16, outPointIndex uint32, node1, node2 *models.Node,
skipProof ...bool) (*models.ChannelEdgeInfo, lnwire.ShortChannelID) {
shouldSkipProof := len(skipProof) > 0 && skipProof[0]
shortChanID := lnwire.ShortChannelID{
BlockHeight: height,
TxIndex: txIndex,
TxPosition: txPosition,
}
outpoint := wire.OutPoint{
Hash: rev,
Index: outPointIndex,
}
node1Pub, _ := node1.PubKey()
node2Pub, _ := node2.PubKey()
node1Vertex, _ := route.NewVertexFromBytes(
node1Pub.SerializeCompressed(),
)
node2Vertex, _ := route.NewVertexFromBytes(
node2Pub.SerializeCompressed(),
)
var edgeInfo *models.ChannelEdgeInfo
switch version {
case lnwire.GossipVersion1:
btcKey1, _ := route.NewVertexFromBytes(
node1Pub.SerializeCompressed(),
)
btcKey2, _ := route.NewVertexFromBytes(
node2Pub.SerializeCompressed(),
)
opts := []models.EdgeModifier{
models.WithChannelPoint(outpoint),
models.WithCapacity(9000),
}
if !shouldSkipProof {
proof := models.NewV1ChannelAuthProof(
testSig.Serialize(),
testSig.Serialize(),
testSig.Serialize(),
testSig.Serialize(),
)
opts = append(opts, models.WithChanProof(proof))
}
edgeInfo, _ = models.NewV1Channel(
shortChanID.ToUint64(),
*chaincfg.MainNetParams.GenesisHash,
node1Vertex,
node2Vertex,
&models.ChannelV1Fields{
BitcoinKey1Bytes: btcKey1,
BitcoinKey2Bytes: btcKey2,
ExtraOpaqueData: make([]byte, 0),
},
opts...,
)
case lnwire.GossipVersion2:
btcKey1, _ := route.NewVertexFromBytes(
node1Pub.SerializeCompressed(),
)
btcKey2, _ := route.NewVertexFromBytes(
node2Pub.SerializeCompressed(),
)
// Create a test merkle root hash.
var merkleRoot chainhash.Hash
copy(merkleRoot[:], bytes.Repeat([]byte{0xaa}, 32))
// Create a test funding script.
fundingScript := []byte{0x00, 0x20}
fundingScript = append(
fundingScript, bytes.Repeat([]byte{0xbb}, 32)...,
)
opts := []models.EdgeModifier{
models.WithChannelPoint(outpoint),
models.WithCapacity(9000),
}
if !shouldSkipProof {
proof := models.NewV2ChannelAuthProof(
testSig.Serialize(),
)
opts = append(opts, models.WithChanProof(proof))
}
edgeInfo, _ = models.NewV2Channel(
shortChanID.ToUint64(),
*chaincfg.MainNetParams.GenesisHash,
node1Vertex,
node2Vertex,
&models.ChannelV2Fields{
BitcoinKey1Bytes: fn.Some(btcKey1),
BitcoinKey2Bytes: fn.Some(btcKey2),
MerkleRootHash: fn.Some(merkleRoot),
FundingScript: fn.Some(fundingScript),
ExtraSignedFields: make(map[uint64][]byte),
},
opts...,
)
}
return edgeInfo, shortChanID
}
// testDisconnectBlockAtHeight checks that the pruned state of the channel
// database is what we expect after calling DisconnectBlockAtHeight.
func testDisconnectBlockAtHeight(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t, WithSyncGraphCachePopulation())
sourceNode := createTestVertex(t, v)
require.NoError(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, v)
node2 := createTestVertex(t, v)
// In addition to the fake vertexes we create some fake channel
// identifiers.
var spendOutputs []*wire.OutPoint
var blockHash chainhash.Hash
copy(blockHash[:], bytes.Repeat([]byte{1}, 32))
// Prune the graph a few times to make sure we have entries in the
// prune log.
_, err := graph.PruneGraph(ctx, spendOutputs, &blockHash, 155)
require.NoError(t, err, "unable to prune graph")
var blockHash2 chainhash.Hash
copy(blockHash2[:], bytes.Repeat([]byte{2}, 32))
_, err = graph.PruneGraph(ctx, spendOutputs, &blockHash2, 156)
require.NoError(t, err, "unable to prune graph")
// Create an edge which has its block height at 156.
height := uint32(156)
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 gets
// edges from the entire range.
edgeInfo2, _ := createEdge(
v, height+1, math.MaxUint32&0x00ffffff, math.MaxUint16,
1, node1, node2,
)
// Create a third edge, this with a block height of 155.
edgeInfo3, _ := createEdge(v, height-1, 0, 0, 2, node1, node2)
// Now add all these new edges to the database.
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo2))
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo3))
assertEdgeWithNoPoliciesInCache(t, graph, edgeInfo)
assertEdgeWithNoPoliciesInCache(t, graph, edgeInfo2)
assertEdgeWithNoPoliciesInCache(t, graph, edgeInfo3)
// Call DisconnectBlockAtHeight, which should prune every channel
// that has a funding height of 'height' or greater.
removed, err := graph.DisconnectBlockAtHeight(ctx, height)
require.NoError(t, err)
assertNoEdge(t, graph, edgeInfo.ChannelID)
assertNoEdge(t, graph, edgeInfo2.ChannelID)
assertEdgeWithNoPoliciesInCache(t, graph, edgeInfo3)
// The two edges should have been removed.
require.Len(t, removed, 2)
require.Equal(t, edgeInfo.ChannelID, removed[0].ChannelID)
require.Equal(t, edgeInfo2.ChannelID, removed[1].ChannelID)
// The two first edges should be removed from the db.
has, isZombie, err := graph.HasChannelEdge(
ctx, v, edgeInfo.ChannelID,
)
require.NoError(t, err, "unable to query for edge")
require.False(t, has)
require.False(t, isZombie)
has, isZombie, err = graph.HasChannelEdge(
ctx, v, edgeInfo2.ChannelID,
)
require.NoError(t, err, "unable to query for edge")
require.False(t, has)
require.False(t, isZombie)
// Edge 3 should not be removed.
has, isZombie, err = graph.HasChannelEdge(
ctx, v, edgeInfo3.ChannelID,
)
require.NoError(t, err, "unable to query for edge")
require.True(t, has)
require.False(t, isZombie)
// PruneTip should be set to the blockHash we specified for the block
// at height 155.
hash, h, err := graph.PruneTip(ctx)
require.NoError(t, err, "unable to get prune tip")
require.True(t, blockHash.IsEqual(hash))
require.Equal(t, h, height-1)
}
func assertEdgeInfoEqual(t *testing.T, e1 *models.ChannelEdgeInfo,
e2 *models.ChannelEdgeInfo) {
require.Equal(t, e2.ChannelID, e1.ChannelID)
require.Equal(t, e2.ChainHash, e1.ChainHash)
require.Equal(t, e2.NodeKey1Bytes[:], e1.NodeKey1Bytes[:])
require.Equal(t, e2.NodeKey2Bytes[:], e1.NodeKey2Bytes[:])
btcKey1E1 := e1.BitcoinKey1Bytes.UnwrapOr(route.Vertex{})
btcKey1E2 := e2.BitcoinKey1Bytes.UnwrapOr(route.Vertex{})
require.Equal(t, btcKey1E2[:], btcKey1E1[:])
btcKey2E1 := e1.BitcoinKey2Bytes.UnwrapOr(route.Vertex{})
btcKey2E2 := e2.BitcoinKey2Bytes.UnwrapOr(route.Vertex{})
require.Equal(t, btcKey2E2[:], btcKey2E1[:])
require.True(t, e1.Features.Equals(e2.Features.RawFeatureVector))
require.True(t, bytes.Equal(
e1.AuthProof.NodeSig1(),
e2.AuthProof.NodeSig1(),
))
require.True(t, bytes.Equal(
e1.AuthProof.NodeSig2(),
e2.AuthProof.NodeSig2(),
))
require.True(t, bytes.Equal(
e1.AuthProof.BitcoinSig1(),
e2.AuthProof.BitcoinSig1(),
))
require.True(t, bytes.Equal(
e1.AuthProof.BitcoinSig2(),
e2.AuthProof.BitcoinSig2(),
))
require.Equal(t, e2.ChannelPoint, e1.ChannelPoint)
require.Equal(t, e2.Capacity, e1.Capacity)
require.Equal(t, e2.ExtraOpaqueData, e1.ExtraOpaqueData)
}
func createChannelEdge(node1, node2 *models.Node,
v lnwire.GossipVersion) (*models.ChannelEdgeInfo,
*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) {
var (
firstNode [33]byte
secondNode [33]byte
)
if bytes.Compare(node1.PubKeyBytes[:], node2.PubKeyBytes[:]) == -1 {
firstNode = node1.PubKeyBytes
secondNode = node2.PubKeyBytes
} else {
firstNode = node2.PubKeyBytes
secondNode = node1.PubKeyBytes
}
// In addition to the fake vertexes we create some fake channel
// identifiers.
chanID := uint64(prand.Int63())
outpoint := wire.OutPoint{
Hash: rev,
Index: prand.Uint32(),
}
// Add the new edge to the database, this should proceed without any
// errors.
var node1Key, node2Key route.Vertex
copy(node1Key[:], firstNode[:])
copy(node2Key[:], secondNode[:])
extraData := []byte{
1, 1, 1,
2, 2, 2, 2,
3, 3, 3, 3, 3,
}
var (
edgeInfo *models.ChannelEdgeInfo
edge1 *models.ChannelEdgePolicy
edge2 *models.ChannelEdgePolicy
)
switch v {
case gossipV1:
proof := models.NewV1ChannelAuthProof(
testSig.Serialize(),
testSig.Serialize(),
testSig.Serialize(),
testSig.Serialize(),
)
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, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(
MakeTestGraph(t, WithSyncGraphCachePopulation()), v,
)
// We'd like to test the update of edges inserted into the database, so
// we create two vertexes to connect.
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
assertNodeInCache(t, graph.ChannelGraph, node1, testFeatures)
node2 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node2))
assertNodeInCache(t, graph.ChannelGraph, node2, testFeatures)
// Create an edge and add it to the db.
edgeInfo, edge1, edge2 := createChannelEdge(node1, node2, v)
// Make sure inserting the policy at this point, before the edge info
// is added, will fail.
err := graph.UpdateEdgePolicy(ctx, edge1)
require.ErrorIs(t, err, ErrEdgeNotFound)
require.Len(t, graph.cache.graphCache.nodeChannels, 0)
// Add the edge info.
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
assertEdgeWithNoPoliciesInCache(t, graph.ChannelGraph, edgeInfo)
chanID := edgeInfo.ChannelID
outpoint := edgeInfo.ChannelPoint
// Next, insert both edge policies into the database, they should both
// be inserted without any issues.
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
assertEdgeWithPolicyInCache(
t, graph.ChannelGraph, edgeInfo, edge1, true,
)
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
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(ctx, chanID)
require.NoError(t, err, "unable to query for edge")
require.True(t, found)
require.False(t, isZombie)
// We should also be able to retrieve the channelID only knowing the
// channel point of the channel.
dbChanID, err := graph.ChannelID(ctx, &outpoint)
require.NoError(t, err, "unable to retrieve channel ID")
require.Equal(t, chanID, dbChanID)
// With the edges inserted, perform some queries to ensure that they've
// been inserted properly.
dbEdgeInfo, dbEdge1, dbEdge2, err := graph.FetchChannelEdgesByID(
ctx, chanID,
)
require.NoError(t, err, "unable to fetch channel by ID")
compareEdgePolicies(t, dbEdge1, edge1)
compareEdgePolicies(t, dbEdge2, edge2)
assertEdgeInfoEqual(t, dbEdgeInfo, edgeInfo)
// Next, attempt to query the channel edges according to the outpoint
// of the channel.
dbEdgeInfo, dbEdge1, dbEdge2, err = graph.FetchChannelEdgesByOutpoint(
ctx, &outpoint,
)
require.NoError(t, err, "unable to fetch channel by ID")
compareEdgePolicies(t, dbEdge1, edge1)
compareEdgePolicies(t, dbEdge2, edge2)
assertEdgeInfoEqual(t, dbEdgeInfo, edgeInfo)
}
// testEdgePolicyCRUD tests basic CRUD operations for edge policies.
func testEdgePolicyCRUD(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
node1 := createTestVertex(t, v)
node2 := createTestVertex(t, v)
// Create an edge. Don't add it to the DB yet.
edgeInfo, shortChanID := createEdge(
v, 100, 1, 0, 0, node1, node2,
)
chanID := shortChanID.ToUint64()
edge1 := newEdgePolicy(v, chanID, nextUpdateTime().Unix(), true)
edge2 := newEdgePolicy(v, chanID, nextUpdateTime().Unix(), false)
edge1.ToNode = edgeInfo.NodeKey2Bytes
edge2.ToNode = edgeInfo.NodeKey1Bytes
edge1.SigBytes = testSig.Serialize()
edge2.SigBytes = testSig.Serialize()
updateAndAssertPolicies := func() {
// Make copies of the policies before calling UpdateEdgePolicy
// to avoid any data race's that can occur during async calls
// that UpdateEdgePolicy may trigger.
edge1 := copyEdgePolicy(edge1)
edge2 := copyEdgePolicy(edge2)
switch v {
case lnwire.GossipVersion1:
edge1.LastUpdate = nextUpdateTime()
edge2.LastUpdate = nextUpdateTime()
case lnwire.GossipVersion2:
edge1.LastBlockHeight = nextBlockHeight()
edge2.LastBlockHeight = nextBlockHeight()
}
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
// Even though we assert at the DB level that any newer edge
// update has a newer timestamp, we need to still gracefully
// handle the case where the same exact policy is re-added since
// it could be possible that our batch executor has two of the
// same policy updates in the same batch.
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
// Use the ForEachChannel method to fetch the policies and
// assert that the deserialized policies match the original
// ones.
err := graph.ForEachChannel(
ctx,
func(info *models.ChannelEdgeInfo,
policy1 *models.ChannelEdgePolicy,
policy2 *models.ChannelEdgePolicy) error {
compareEdgePolicies(t, edge1, policy1)
compareEdgePolicies(t, edge2, policy2)
return nil
}, func() {},
)
require.NoError(t, err)
}
// Make sure inserting the policy at this point, before the edge info
// is added, will fail.
require.ErrorIs(t, graph.UpdateEdgePolicy(ctx, edge1), ErrEdgeNotFound)
// Now add the edge.
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
updateAndAssertPolicies()
// Update one of the edges to have no extra opaque data.
edge1.ExtraOpaqueData = nil
updateAndAssertPolicies()
switch v {
case lnwire.GossipVersion1:
// Update one of the edges to have ChannelFlags include a bit
// unknown to us.
edge1.ChannelFlags |= 1 << 6
// Update the other edge to have MessageFlags include a bit
// unknown to us.
edge2.MessageFlags |= 1 << 4
case lnwire.GossipVersion2:
// Update one of the edges to have DisableFlags include a bit
// unknown to us.
edge1.DisableFlags |= 1 << 6
// Update the other edge to have a modified extra signed field.
edge2.ExtraSignedFields = map[uint64][]byte{
200: {0x4, 0x5},
}
}
updateAndAssertPolicies()
}
func assertNodeInCache(t *testing.T, g *ChannelGraph, n *models.Node,
expectedFeatures *lnwire.FeatureVector) {
// Let's check the internal view first.
nodeFeatures := g.cache.graphCache.nodeFeatures
require.Equal(
t, expectedFeatures, nodeFeatures[n.PubKeyBytes],
)
// The external view should reflect this as well. Except when we expect
// the features to be nil internally, we return an empty feature vector
// on the public interface instead.
if expectedFeatures == nil {
expectedFeatures = lnwire.EmptyFeatureVector()
}
features := g.cache.graphCache.GetFeatures(n.PubKeyBytes)
require.Equal(t, expectedFeatures, features)
}
func assertNodeNotInCache(t *testing.T, g *ChannelGraph, n route.Vertex) {
_, ok := g.cache.graphCache.nodeFeatures[n]
require.False(t, ok)
_, ok = g.cache.graphCache.nodeChannels[n]
require.False(t, ok)
// We should get the default features for this node.
features := g.cache.graphCache.GetFeatures(n)
require.Equal(t, lnwire.EmptyFeatureVector(), features)
}
func assertEdgeWithNoPoliciesInCache(t *testing.T, g *ChannelGraph,
e *models.ChannelEdgeInfo) {
// Let's check the internal view first.
require.NotEmpty(t, g.cache.graphCache.nodeChannels[e.NodeKey1Bytes])
require.NotEmpty(t, g.cache.graphCache.nodeChannels[e.NodeKey2Bytes])
expectedNode1Channel := &DirectedChannel{
ChannelID: e.ChannelID,
IsNode1: true,
OtherNode: e.NodeKey2Bytes,
Capacity: e.Capacity,
OutPolicySet: false,
InPolicy: nil,
}
nodeChannels := g.cache.graphCache.nodeChannels
require.Contains(
t, nodeChannels[e.NodeKey1Bytes], e.ChannelID,
)
require.Equal(
t, expectedNode1Channel,
nodeChannels[e.NodeKey1Bytes][e.ChannelID],
)
expectedNode2Channel := &DirectedChannel{
ChannelID: e.ChannelID,
IsNode1: false,
OtherNode: e.NodeKey1Bytes,
Capacity: e.Capacity,
OutPolicySet: false,
InPolicy: nil,
}
require.Contains(
t, nodeChannels[e.NodeKey2Bytes], e.ChannelID,
)
require.Equal(
t, expectedNode2Channel,
nodeChannels[e.NodeKey2Bytes][e.ChannelID],
)
// The external view should reflect this as well.
var foundChannel *DirectedChannel
err := g.cache.graphCache.ForEachChannel(
e.NodeKey1Bytes, func(c *DirectedChannel) error {
if c.ChannelID == e.ChannelID {
foundChannel = c
}
return nil
},
)
require.NoError(t, err)
require.NotNil(t, foundChannel)
require.Equal(t, expectedNode1Channel, foundChannel)
err = g.cache.graphCache.ForEachChannel(
e.NodeKey2Bytes, func(c *DirectedChannel) error {
if c.ChannelID == e.ChannelID {
foundChannel = c
}
return nil
},
)
require.NoError(t, err)
require.NotNil(t, foundChannel)
require.Equal(t, expectedNode2Channel, foundChannel)
}
func assertNoEdge(t *testing.T, g *ChannelGraph, chanID uint64) {
// Make sure no channel in the cache has the given channel ID. If there
// are no channels at all, that is fine as well.
for _, channels := range g.cache.graphCache.nodeChannels {
for _, channel := range channels {
require.NotEqual(t, channel.ChannelID, chanID)
}
}
}
func assertEdgeWithPolicyInCache(t *testing.T, g *ChannelGraph,
e *models.ChannelEdgeInfo, p *models.ChannelEdgePolicy, policy1 bool) {
// Check the internal state first.
c1, ok := g.cache.graphCache.nodeChannels[e.NodeKey1Bytes][e.ChannelID]
require.True(t, ok)
if policy1 {
require.True(t, c1.OutPolicySet)
} else {
require.NotNil(t, c1.InPolicy)
require.Equal(
t, p.FeeProportionalMillionths,
c1.InPolicy.FeeProportionalMillionths,
)
}
c2, ok := g.cache.graphCache.nodeChannels[e.NodeKey2Bytes][e.ChannelID]
require.True(t, ok)
if policy1 {
require.NotNil(t, c2.InPolicy)
require.Equal(
t, p.FeeProportionalMillionths,
c2.InPolicy.FeeProportionalMillionths,
)
} else {
require.True(t, c2.OutPolicySet)
}
// Now for both nodes make sure that the external view is also correct.
var (
c1Ext *DirectedChannel
c2Ext *DirectedChannel
)
require.NoError(t, g.cache.graphCache.ForEachChannel(
e.NodeKey1Bytes, func(c *DirectedChannel) error {
c1Ext = c
return nil
},
))
require.NoError(t, g.cache.graphCache.ForEachChannel(
e.NodeKey2Bytes, func(c *DirectedChannel) error {
c2Ext = c
return nil
},
))
// Only compare the fields that are actually copied, then compare the
// values of the functions separately.
require.Equal(t, c1, c1Ext.DeepCopy())
require.Equal(t, c2, c2Ext.DeepCopy())
if policy1 {
require.Equal(
t, p.FeeProportionalMillionths,
c2Ext.InPolicy.FeeProportionalMillionths,
)
require.Equal(
t, route.Vertex(e.NodeKey2Bytes),
c2Ext.InPolicy.ToNodePubKey(),
)
require.Equal(t, testFeatures, c2Ext.InPolicy.ToNodeFeatures)
} else {
require.Equal(
t, p.FeeProportionalMillionths,
c1Ext.InPolicy.FeeProportionalMillionths,
)
require.Equal(
t, route.Vertex(e.NodeKey1Bytes),
c1Ext.InPolicy.ToNodePubKey(),
)
require.Equal(t, testFeatures, c1Ext.InPolicy.ToNodeFeatures)
}
}
func randEdgePolicy(chanID uint64) *models.ChannelEdgePolicy {
update := prand.Int63()
return newEdgePolicy(lnwire.GossipVersion1, chanID, update, true)
}
func copyEdgePolicy(p *models.ChannelEdgePolicy) *models.ChannelEdgePolicy {
return &models.ChannelEdgePolicy{
Version: p.Version,
SigBytes: p.SigBytes,
ChannelID: p.ChannelID,
LastUpdate: p.LastUpdate,
LastBlockHeight: p.LastBlockHeight,
SecondPeer: p.SecondPeer,
MessageFlags: p.MessageFlags,
ChannelFlags: p.ChannelFlags,
DisableFlags: p.DisableFlags,
TimeLockDelta: p.TimeLockDelta,
MinHTLC: p.MinHTLC,
MaxHTLC: p.MaxHTLC,
FeeBaseMSat: p.FeeBaseMSat,
FeeProportionalMillionths: p.FeeProportionalMillionths,
ToNode: p.ToNode,
ExtraOpaqueData: p.ExtraOpaqueData,
ExtraSignedFields: p.ExtraSignedFields,
}
}
func newEdgePolicy(v lnwire.GossipVersion, chanID uint64,
updateTime int64, isNode1 bool) *models.ChannelEdgePolicy {
policy := &models.ChannelEdgePolicy{
Version: v,
SecondPeer: !isNode1,
ChannelID: chanID,
TimeLockDelta: uint16(prand.Int63()),
MinHTLC: lnwire.MilliSatoshi(prand.Int63()),
MaxHTLC: lnwire.MilliSatoshi(prand.Int63()),
FeeBaseMSat: lnwire.MilliSatoshi(prand.Int63()),
FeeProportionalMillionths: lnwire.MilliSatoshi(prand.Int63()),
}
if v == lnwire.GossipVersion1 {
policy.LastUpdate = time.Unix(updateTime, 0)
policy.MessageFlags = 1
if !isNode1 {
policy.ChannelFlags = lnwire.ChanUpdateDirection
}
policy.ExtraOpaqueData = []byte{1, 0}
} else {
policy.LastBlockHeight = nextBlockHeight()
policy.DisableFlags = 0
policy.ExtraSignedFields = map[uint64][]byte{
100: {0x1, 0x2, 0x3},
}
}
return policy
}
// testAddEdgeProof tests the ability to add an edge proof to an existing edge.
func testAddEdgeProof(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// Add an edge with no proof.
node1 := createTestVertex(t, v)
node2 := createTestVertex(t, v)
// Create edge without proof (skipProof = true).
edge1, _ := createEdge(v, 100, 0, 0, 0, node1, node2, true)
require.NoError(t, graph.AddChannelEdge(ctx, edge1))
// Fetch the edge and assert that the proof is nil.
dbEdge, _, _, err := graph.FetchChannelEdgesByID(
ctx, edge1.ChannelID,
)
require.NoError(t, err)
require.Nil(t, dbEdge.AuthProof)
// Create a proof appropriate for the version.
var proof *models.ChannelAuthProof
switch v {
case lnwire.GossipVersion1:
proof = models.NewV1ChannelAuthProof(
testSig.Serialize(),
testSig.Serialize(),
testSig.Serialize(),
testSig.Serialize(),
)
case lnwire.GossipVersion2:
proof = models.NewV2ChannelAuthProof(testSig.Serialize())
}
// First, add the proof to the rest of the channel edge info and try
// to call AddChannelEdge again - this should fail due to the channel
// already existing.
edge1.AuthProof = proof
err = graph.AddChannelEdge(ctx, edge1)
require.ErrorIs(t, err, ErrEdgeAlreadyExist)
// Now add just the proof via AddEdgeProof.
scid1 := lnwire.NewShortChanIDFromInt(edge1.ChannelID)
require.NoError(t, graph.AddEdgeProof(ctx, scid1, proof))
// Fetch the edge again and assert that the proof is now set.
dbEdge, _, _, err = graph.FetchChannelEdgesByID(
ctx, edge1.ChannelID,
)
require.NoError(t, err)
require.NotNil(t, dbEdge.AuthProof)
// For completeness, also test the case where we insert a new edge with
// an edge proof from the start. Show that the proof is present.
edge2, _ := createEdge(v, 200, 0, 0, 1, node1, node2)
require.NoError(t, graph.AddChannelEdge(ctx, edge2))
// Fetch the edge and assert that the proof is set.
dbEdge2, _, _, err := graph.FetchChannelEdgesByID(
ctx, edge2.ChannelID,
)
require.NoError(t, err)
require.NotNil(t, dbEdge2.AuthProof)
}
// testForEachSourceNodeChannel tests that the ForEachSourceNodeChannel
// correctly iterates through the channels of the set source node.
func testForEachSourceNodeChannel(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// Create a source node (A) and set it as such in the DB.
nodeA := createTestVertex(t, v)
require.NoError(t, graph.SetSourceNode(ctx, nodeA))
// Now, create a few more nodes (B, C, D) along with some channels
// between them. We'll create the following graph:
//
// A -- B -- D
// |
// C
//
// The graph includes a channel (B-D) that does not belong to the source
// node along with 2 channels (A-B and A-C) that do belong to the source
// node. For the A-B channel, we will let the source node set an
// outgoing policy but for the A-C channel, we will set only an incoming
// policy.
nodeB := createTestVertex(t, v)
nodeC := createTestVertex(t, v)
nodeD := createTestVertex(t, v)
abEdge, _ := createEdge(v, 100, 0, 0, 0, nodeA, nodeB)
require.NoError(t, graph.AddChannelEdge(ctx, abEdge))
acEdge, _ := createEdge(v, 200, 0, 0, 1, nodeA, nodeC)
require.NoError(t, graph.AddChannelEdge(ctx, acEdge))
bdEdge, _ := createEdge(v, 300, 0, 0, 2, nodeB, nodeD)
require.NoError(t, graph.AddChannelEdge(ctx, bdEdge))
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 := newPolicy(
acEdge, nodeC.PubKeyBytes, nodeA.PubKeyBytes,
)
require.NoError(t, graph.UpdateEdgePolicy(ctx, acPolicyAIncoming))
type sourceNodeChan struct {
otherNode route.Vertex
havePolicy bool
}
// Put together our expected source node channels.
expectedSrcChans := map[wire.OutPoint]*sourceNodeChan{
abEdge.ChannelPoint: {
otherNode: nodeB.PubKeyBytes,
havePolicy: true,
},
acEdge.ChannelPoint: {
otherNode: nodeC.PubKeyBytes,
havePolicy: false,
},
}
// 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 {
require.Contains(t, expectedSrcChans, chanPoint)
expected := expectedSrcChans[chanPoint]
require.Equal(
t, expected.otherNode[:],
otherNode.PubKeyBytes[:],
)
require.Equal(t, expected.havePolicy, havePolicy)
delete(expectedSrcChans, chanPoint)
return nil
}, func() {},
)
require.NoError(t, err)
require.Empty(t, expectedSrcChans)
}
// TestGraphTraversal tests that we can traverse the graph and find all
// nodes and channels that we expect to find.
func TestGraphTraversal(t *testing.T) {
t.Parallel()
ctx := t.Context()
// If we turn the channel graph cache _off_, then iterate through the
// set of channels (to force the fall back), we should find all the
// channel as well as the nodes included.
graph := MakeTestGraph(t, WithUseGraphCache(false))
// 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 each node pair.
const numNodes = 20
const numChannels = 5
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{})
for _, node := range nodeList {
nodeIndex[node.PubKeyBytes] = struct{}{}
}
err := graph.ForEachNodeCached(ctx, lnwire.GossipVersion1,
func(_ context.Context, node route.Vertex,
chans map[uint64]*DirectedChannel) error {
if _, ok := nodeIndex[node]; !ok {
return fmt.Errorf("node %x not found in graph",
node)
}
for chanID := range chans {
if _, ok := chanIndex[chanID]; !ok {
return fmt.Errorf(
"chan %v not found in graph",
chanID,
)
}
}
return nil
}, func() {})
require.NoError(t, err)
// Iterate through all the known channels within the graph DB, once
// again if the map is empty that indicates that all edges have
// properly been reached.
err = graph.ForEachChannel(ctx, lnwire.GossipVersion1,
func(ei *models.ChannelEdgeInfo,
_ *models.ChannelEdgePolicy,
_ *models.ChannelEdgePolicy) error {
delete(chanIndex, ei.ChannelID)
return nil
}, func() {})
require.NoError(t, err)
require.Len(t, chanIndex, 0)
// Finally, we want to test the ability to iterate over all the
// outgoing channels for a particular node.
numNodeChans := 0
firstNode, secondNode := nodeList[0], nodeList[1]
err = graph.ForEachNodeChannel(
ctx, lnwire.GossipVersion1, firstNode.PubKeyBytes,
func(_ *models.ChannelEdgeInfo, outEdge,
inEdge *models.ChannelEdgePolicy) error {
// All channels between first and second node should
// have fully (both sides) specified policies.
if inEdge == nil || outEdge == nil {
return fmt.Errorf("channel policy not present")
}
// Each should indicate that it's outgoing (pointed
// towards the second node).
if !bytes.Equal(
outEdge.ToNode[:], secondNode.PubKeyBytes[:],
) {
return fmt.Errorf("wrong outgoing edge")
}
// The incoming edge should also indicate that it's
// pointing to the origin node.
if !bytes.Equal(
inEdge.ToNode[:], firstNode.PubKeyBytes[:],
) {
return fmt.Errorf("wrong outgoing edge")
}
numNodeChans++
return nil
}, func() {},
)
require.NoError(t, err)
require.Equal(t, numChannels, numNodeChans)
}
// testGraphTraversalCacheable tests that the memory optimized node traversal is
// working correctly.
func testGraphTraversalCacheable(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
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, nodeList := fillTestGraph(
t, graph.ChannelGraph, numNodes, numChannels, v,
)
// Create a map of all nodes with the nodes we just inserted.
nodeMap := make(map[route.Vertex]struct{})
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 {
delete(nodeMap, node)
nodes = append(nodes, node)
return nil
}, func() {
nodes = nil
})
require.NoError(t, err)
require.Len(t, nodeMap, 0)
// Duplicate the map before we start deleting from it so that we can
// check that both the cached and db version of
// ForEachNodeDirectedChannel works as expected here.
chanIndex2 := make(map[uint64]struct{})
for k, v := range chanIndex {
chanIndex2[k] = v
}
for _, node := range nodes {
// Query the VersionedGraph which uses the cache to iterate
// through the channels for each node.
err = graph.ForEachNodeDirectedChannel(
ctx, node, func(d *DirectedChannel) error {
delete(chanIndex, d.ChannelID)
return nil
}, func() {},
)
require.NoError(t, err)
// Now skip the cache and query the DB directly.
err = graph.db.ForEachNodeDirectedChannel(
ctx, v, node, func(d *DirectedChannel) error {
delete(chanIndex2, d.ChannelID)
return nil
}, func() {},
)
require.NoError(t, err)
}
require.Len(t, chanIndex, 0)
require.Len(t, chanIndex2, 0)
}
// TestGraphCacheTraversal tests traversal of the graph via the graph cache.
func TestGraphCacheTraversal(t *testing.T) {
t.Parallel()
ctx := t.Context()
// Explicitly enable the graph cache so that the
// ForEachNodeDirectedChannel call below will use the cache.
graph := MakeTestGraph(t, WithUseGraphCache(true))
// 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 each node pair.
const numNodes = 20
const numChannels = 5
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
// properly been reached.
numNodeChans := 0
for _, node := range nodeList {
err := graph.ForEachNodeDirectedChannel(
ctx, node.PubKeyBytes, func(d *DirectedChannel) error {
delete(chanIndex, d.ChannelID)
if !d.OutPolicySet || d.InPolicy == nil {
return fmt.Errorf("channel policy " +
"not present")
}
// The incoming edge should also indicate that
// it's pointing to the origin node.
inPolicyNodeKey := d.InPolicy.ToNodePubKey()
if !bytes.Equal(
inPolicyNodeKey[:], node.PubKeyBytes[:],
) {
return fmt.Errorf("wrong outgoing edge")
}
numNodeChans++
return nil
}, func() {
numNodeChans = 0
},
)
require.NoError(t, err)
}
require.Len(t, chanIndex, 0)
// We count the channels for both nodes, so there should be double the
// amount now. Except for the very last node, that doesn't have any
// channels to make the loop easier in fillTestGraph().
require.Equal(t, numChannels*2*(numNodes-1), numNodeChans)
}
// fillTestGraph fills the graph with nodes and channels using the requested
// gossip version.
func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes,
numChannels int, v lnwire.GossipVersion) (map[uint64]struct{},
[]*models.Node) {
ctx := t.Context()
nodes := make([]*models.Node, numNodes)
nodeIndex := map[route.Vertex]struct{}{}
for i := 0; i < numNodes; i++ {
node := createTestVertex(t, v)
nodes[i] = node
nodeIndex[node.PubKeyBytes] = struct{}{}
}
// Add each of the nodes into the graph, they should be inserted
// without error.
for _, node := range nodes {
require.NoError(t, graph.AddNode(ctx, node))
}
// Iterate over each node as returned by the graph, if all nodes are
// reached, then the map created above should be empty.
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]
if bytes.Compare(
node1.PubKeyBytes[:], node2.PubKeyBytes[:],
) == -1 {
node1, node2 = node2, node1
}
for i := 0; i < numChannels; i++ {
txHash := sha256.Sum256([]byte{byte(i)})
chanID := uint64((n << 8) + i + 1)
op := wire.OutPoint{
Hash: txHash,
Index: 0,
}
var node1Key, node2Key route.Vertex
copy(node1Key[:], node1.PubKeyBytes[:])
copy(node2Key[:], node2.PubKeyBytes[:])
edgeInfo := buildEdgeInfo(
chanID, node1Key, node2Key, op, v,
)
err = graph.AddChannelEdge(ctx, edgeInfo)
require.NoError(t, err)
// Create and add an edge with random data that points
// from node1 -> node2.
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 = newEdgePolicy(
v, chanID, prand.Int63(), false,
)
edge.ToNode = node1.PubKeyBytes
edge.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge))
chanIndex[chanID] = struct{}{}
}
}
return chanIndex, nodes
}
func assertPruneTip(t *testing.T, graph *ChannelGraph,
blockHash *chainhash.Hash, blockHeight uint32) {
pruneHash, pruneHeight, err := graph.PruneTip(t.Context())
require.NoError(t, err)
require.Equal(t, blockHash[:], pruneHash[:])
require.Equal(t, blockHeight, pruneHeight)
}
func assertNumChans(t *testing.T, graph *ChannelGraph, n int) {
numChans := 0
err := graph.ForEachChannel(
t.Context(), lnwire.GossipVersion1,
func(*models.ChannelEdgeInfo,
*models.ChannelEdgePolicy,
*models.ChannelEdgePolicy) error {
numChans++
return nil
}, func() {
numChans = 0
},
)
require.NoError(t, err)
require.Equal(t, n, numChans)
}
func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) {
numNodes := 0
v1Graph := NewVersionedGraph(graph, lnwire.GossipVersion1)
err := v1Graph.ForEachNode(t.Context(), func(_ *models.Node) error {
numNodes++
return nil
}, func() {})
require.NoError(t, err)
require.Equal(t, n, numNodes)
}
func assertChanViewEqual(t *testing.T, a []EdgePoint, b []EdgePoint) {
require.Len(t, b, len(a))
chanViewSet := make(map[wire.OutPoint]struct{})
for _, op := range a {
chanViewSet[op.OutPoint] = struct{}{}
}
for _, op := range b {
_, ok := chanViewSet[op.OutPoint]
require.True(t, ok)
}
}
func assertChanViewEqualChanPoints(t *testing.T, a []EdgePoint,
b []*wire.OutPoint) {
require.Len(t, b, len(a))
chanViewSet := make(map[wire.OutPoint]struct{})
for _, op := range a {
chanViewSet[op.OutPoint] = struct{}{}
}
for _, op := range b {
_, ok := chanViewSet[*op]
require.True(t, ok)
}
}
func TestGraphPruning(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
sourceNode := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.SetSourceNode(ctx, sourceNode))
// As initial set up for the test, we'll create a graph with 5 vertexes
// and enough edges to create a fully connected graph. The graph will
// be rather simple, representing a straight line.
const numNodes = 5
graphNodes := make([]*models.Node, numNodes)
for i := 0; i < numNodes; i++ {
node := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node))
graphNodes[i] = node
}
// With the vertexes created, we'll next create a series of channels
// between them.
channelPoints := make([]*wire.OutPoint, 0, numNodes-1)
edgePoints := make([]EdgePoint, 0, numNodes-1)
for i := 0; i < numNodes-1; i++ {
txHash := sha256.Sum256([]byte{byte(i)})
chanID := uint64(i + 1)
op := wire.OutPoint{
Hash: txHash,
Index: 0,
}
channelPoints = append(channelPoints, &op)
var node1Key, node2Key route.Vertex
copy(node1Key[:], graphNodes[i].PubKeyBytes[:])
copy(node2Key[:], graphNodes[i+1].PubKeyBytes[:])
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)
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
pkScript, err := edgeInfo.FundingPKScript()
require.NoError(t, err)
edgePoints = append(edgePoints, EdgePoint{
FundingPkScript: pkScript,
OutPoint: op,
})
// Create and add an edge with random data that points from
// node_i -> node_i+1
edge := randEdgePolicy(chanID)
edge.ChannelFlags = 0
edge.ToNode = graphNodes[i].PubKeyBytes
edge.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge))
// Create another random edge that points from node_i+1 ->
// node_i this time.
edge = randEdgePolicy(chanID)
edge.ChannelFlags = 1
edge.ToNode = graphNodes[i].PubKeyBytes
edge.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge))
}
v1Graph := NewVersionedGraph(graph, lnwire.GossipVersion1)
// With all the channel points added, we'll consult the graph to ensure
// it has the same channel view as the one we just constructed.
channelView, err := v1Graph.ChannelView(ctx)
require.NoError(t, err, "unable to get graph channel view")
assertChanViewEqual(t, channelView, edgePoints)
// Now with our test graph created, we can test the pruning
// capabilities of the channel graph.
// First we create a mock block that ends up closing the first two
// channels.
var blockHash chainhash.Hash
copy(blockHash[:], bytes.Repeat([]byte{1}, 32))
blockHeight := uint32(1)
block := channelPoints[:2]
prunedChans, err := graph.PruneGraph(
ctx, block, &blockHash, blockHeight,
)
require.NoError(t, err, "unable to prune graph")
require.Len(t, prunedChans, 2)
// Now ensure that the prune tip has been updated.
assertPruneTip(t, graph, &blockHash, blockHeight)
// Count up the number of channels known within the graph, only 2
// should be remaining.
assertNumChans(t, graph, 2)
// Those channels should also be missing from the channel view.
channelView, err = v1Graph.ChannelView(ctx)
require.NoError(t, err, "unable to get graph channel view")
assertChanViewEqualChanPoints(t, channelView, channelPoints[2:])
// Next we'll create a block that doesn't close any channels within the
// graph to test the negative error case.
fakeHash := sha256.Sum256([]byte("test prune"))
nonChannel := &wire.OutPoint{
Hash: fakeHash,
Index: 9,
}
blockHash = sha256.Sum256(blockHash[:])
blockHeight = 2
prunedChans, err = graph.PruneGraph(
ctx, []*wire.OutPoint{nonChannel}, &blockHash, blockHeight,
)
require.NoError(t, err, "unable to prune graph")
// No channels should have been detected as pruned.
require.Empty(t, prunedChans)
// Once again, the prune tip should have been updated. We should still
// see both channels and their participants, along with the source node.
assertPruneTip(t, graph, &blockHash, blockHeight)
assertNumChans(t, graph, 2)
assertNumNodes(t, graph, 4)
// Finally, create a block that prunes the remainder of the channels
// from the graph.
blockHash = sha256.Sum256(blockHash[:])
blockHeight = 3
prunedChans, err = graph.PruneGraph(
ctx, channelPoints[2:], &blockHash, blockHeight,
)
require.NoError(t, err, "unable to prune graph")
// The remainder of the channels should have been pruned from the
// graph.
require.Len(t, prunedChans, 2)
// The prune tip should be updated, no channels should be found, and
// only the source node should remain within the current graph.
assertPruneTip(t, graph, &blockHash, blockHeight)
assertNumChans(t, graph, 0)
assertNumNodes(t, graph, 1)
// Finally, the channel view at this point in the graph should now be
// completely empty. Those channels should also be missing from the
// channel view.
channelView, err = v1Graph.ChannelView(ctx)
require.NoError(t, err, "unable to get graph channel view")
require.Empty(t, channelView)
}
// TestHighestChanID tests that we're able to properly retrieve the highest
// known channel ID in the database.
func testHighestChanID(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
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")
require.Zero(t, bestID)
// Next, we'll insert two channels into the database, with each channel
// connecting the same two nodes.
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(v, 10, 0, 0, 0, node1, node2)
edge2, chanID2 := createEdge(v, 100, 0, 0, 0, node1, node2)
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.
bestID, err = graph.HighestChanID(ctx)
require.NoError(t, err, "unable to get highest ID")
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(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")
require.Equal(t, chanID3.ToUint64(), bestID)
}
// TestChanUpdatesInHorizon tests the we're able to properly retrieve all known
// channel updates within a specific time horizon. It also tests that upon
// insertion of a new edge, the edge update index is updated properly.
func TestChanUpdatesInHorizon(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
// If we issue an arbitrary query before any channel updates are
// inserted in the database, we should get zero results.
chanIter := graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartTime: fn.Some(time.Unix(999, 0)),
EndTime: fn.Some(time.Unix(9999, 0)),
},
)
chanUpdates, err := fn.CollectErr(chanIter)
require.NoError(t, err, "unable to updates for updates")
require.Empty(t, chanUpdates)
// We'll start by creating two nodes which will seed our test graph.
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node2))
// We'll now create 10 channels between the two nodes, with update
// times 10 seconds after each other.
const numChans = 10
startTime := time.Unix(1234, 0)
endTime := startTime
edges := make([]ChannelEdge, 0, numChans)
for i := 0; i < numChans; i++ {
channel, chanID := createEdge(
lnwire.GossipVersion1, uint32(i*10), 0, 0, 0,
node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel))
edge1UpdateTime := endTime
edge2UpdateTime := edge1UpdateTime.Add(time.Second)
endTime = endTime.Add(time.Second * 10)
edge1 := newEdgePolicy(
lnwire.GossipVersion1, chanID.ToUint64(),
edge1UpdateTime.Unix(), true,
)
edge1.ChannelFlags = 0
edge1.ToNode = node2.PubKeyBytes
edge1.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
edge2 := newEdgePolicy(
lnwire.GossipVersion1, chanID.ToUint64(),
edge2UpdateTime.Unix(), false,
)
edge2.ChannelFlags = 1
edge2.ToNode = node1.PubKeyBytes
edge2.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
edges = append(edges, ChannelEdge{
Info: channel,
Policy1: edge1,
Policy2: edge2,
})
}
// With our channels loaded, we'll now start our series of queries.
queryCases := []struct {
start time.Time
end time.Time
resp []ChannelEdge
}{
// If we query for a time range that's strictly below our set
// of updates, then we'll get an empty result back.
{
start: time.Unix(100, 0),
end: time.Unix(200, 0),
},
// If we query for a time range that's well beyond our set of
// updates, we should get an empty set of results back.
{
start: time.Unix(99999, 0),
end: time.Unix(999999, 0),
},
// If we query for the start time, and 10 seconds directly
// after it, we should only get a single update, that first
// one.
{
start: time.Unix(1234, 0),
end: startTime.Add(time.Second * 10),
resp: []ChannelEdge{edges[0]},
},
// If we add 10 seconds past the first update, and then
// subtract 10 from the last update, then we should only get
// the 8 edges in the middle.
{
start: startTime.Add(time.Second * 10),
end: endTime.Add(-time.Second * 10),
resp: edges[1:9],
},
// If we use the start and end time as is, we should get the
// entire range.
{
start: startTime,
end: endTime,
resp: edges,
},
}
for _, queryCase := range queryCases {
respIter := graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartTime: fn.Some(queryCase.start),
EndTime: fn.Some(queryCase.end),
},
)
resp, err := fn.CollectErr(respIter)
require.NoError(t, err)
require.Len(t, resp, len(queryCase.resp))
for i := 0; i < len(resp); i++ {
chanExp := queryCase.resp[i]
chanRet := resp[i]
assertEdgeInfoEqual(t, chanExp.Info, chanRet.Info)
compareEdgePolicies(
t, chanExp.Policy1, chanRet.Policy1,
)
compareEdgePolicies(
t, chanExp.Policy2, chanRet.Policy2,
)
}
}
}
// TestNodeUpdatesInHorizon tests that we're able to properly scan and retrieve
// the most recent node updates within a particular time horizon.
func TestNodeUpdatesInHorizon(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
startTime := time.Unix(1234, 0)
endTime := startTime
// If we issue an arbitrary query before we insert any nodes into the
// database, then we shouldn't get any results back.
nodeUpdatesIter := graph.NodeUpdatesInHorizon(
ctx, NodeUpdateRange{
StartTime: fn.Some(time.Unix(999, 0)),
EndTime: fn.Some(time.Unix(9999, 0)),
},
)
nodeUpdates, err := fn.CollectErr(nodeUpdatesIter)
require.NoError(t, err, "unable to query for node updates")
require.Len(t, nodeUpdates, 0)
// We'll create 10 node announcements, each with an update timestamp 10
// seconds after the other.
const numNodes = 10
nodeAnns := make([]models.Node, 0, numNodes)
for i := 0; i < numNodes; i++ {
nodeAnn := createTestVertex(t, lnwire.GossipVersion1)
// The node ann will use the current end time as its last
// update them, then we'll add 10 seconds in order to create
// the proper update time for the next node announcement.
updateTime := endTime
endTime = updateTime.Add(time.Second * 10)
nodeAnn.LastUpdate = updateTime
nodeAnns = append(nodeAnns, *nodeAnn)
require.NoError(t, graph.AddNode(ctx, nodeAnn))
}
queryCases := []struct {
start time.Time
end time.Time
resp []models.Node
}{
// If we query for a time range that's strictly below our set
// of updates, then we'll get an empty result back.
{
start: time.Unix(100, 0),
end: time.Unix(200, 0),
},
// If we query for a time range that's well beyond our set of
// updates, we should get an empty set of results back.
{
start: time.Unix(99999, 0),
end: time.Unix(999999, 0),
},
// If we skip he first time epoch with out start time, then we
// should get back every now but the first.
{
start: startTime.Add(time.Second * 10),
end: endTime,
resp: nodeAnns[1:],
},
// If we query for the range as is, we should get all 10
// announcements back.
{
start: startTime,
end: endTime,
resp: nodeAnns,
},
// If we reduce the ending time by 1 nanosecond before the last
// node's timestamp, then we should get all but the last node.
{
start: startTime,
end: endTime.Add(-time.Second*10 - time.Nanosecond),
resp: nodeAnns[:9],
},
}
for _, queryCase := range queryCases {
iter := graph.NodeUpdatesInHorizon(
ctx, NodeUpdateRange{
StartTime: fn.Some(queryCase.start),
EndTime: fn.Some(queryCase.end),
},
)
resp, err := fn.CollectErr(iter)
require.NoError(t, err, "unable to query for node updates")
require.Len(t, resp, len(queryCase.resp))
for i := 0; i < len(resp); i++ {
compareNodes(t, &queryCase.resp[i], resp[i])
}
}
}
// TestNodeUpdatesInHorizonPublicOnly tests that NodeUpdatesInHorizon with
// WithIterPublicNodesOnly returns only nodes that have at least one public
// channel (one with a channel announcement proof).
func TestNodeUpdatesInHorizonPublicOnly(t *testing.T) {
t.Parallel()
ctx := t.Context()
chanGraph := MakeTestGraph(t)
graph := NewVersionedGraph(chanGraph, lnwire.GossipVersion1)
startTime := time.Unix(1000, 0)
// Create 4 nodes: we'll make node pairs where one pair has a public
// channel (with proof) and the other has a private channel (no proof).
publicNode1 := createTestVertex(t, lnwire.GossipVersion1)
publicNode1.LastUpdate = startTime.Add(10 * time.Second)
// Set publicNode1 as the source node (required before adding
// channel edges).
require.NoError(t, chanGraph.SetSourceNode(ctx, publicNode1))
publicNode2 := createTestVertex(t, lnwire.GossipVersion1)
publicNode2.LastUpdate = startTime.Add(20 * time.Second)
require.NoError(t, chanGraph.AddNode(ctx, publicNode2))
// privateNode has a channel to the source node (publicNode1) but
// without a proof, so it remains private in both KV and SQL backends.
privateNode := createTestVertex(t, lnwire.GossipVersion1)
privateNode.LastUpdate = startTime.Add(30 * time.Second)
require.NoError(t, chanGraph.AddNode(ctx, privateNode))
// Create a standalone node with no channels at all.
lonelyNode := createTestVertex(t, lnwire.GossipVersion1)
lonelyNode.LastUpdate = startTime.Add(40 * time.Second)
require.NoError(t, chanGraph.AddNode(ctx, lonelyNode))
// Add a public channel between publicNode1 and publicNode2
// (with proof, making both nodes public).
publicEdge, _ := createEdge(
lnwire.GossipVersion1, 100, 0, 0, 0,
publicNode1, publicNode2,
)
require.NoError(t, chanGraph.AddChannelEdge(ctx, publicEdge))
// Add a private channel between publicNode1 (source) and
// privateNode (no proof, so privateNode remains private).
privateEdge, _ := createEdge(
lnwire.GossipVersion1, 200, 0, 0, 1,
publicNode1, privateNode, true, // skipProof
)
require.NoError(t, chanGraph.AddChannelEdge(ctx, privateEdge))
// Query without the public-only filter — should return all 4 nodes.
endTime := startTime.Add(60 * time.Second)
r := NodeUpdateRange{
StartTime: fn.Some(startTime),
EndTime: fn.Some(endTime),
}
allIter := graph.NodeUpdatesInHorizon(ctx, r)
allNodes, err := fn.CollectErr(allIter)
require.NoError(t, err)
require.Len(t, allNodes, 4)
// Query with the public-only filter — should return only the 2
// public nodes.
publicIter := graph.NodeUpdatesInHorizon(
ctx, r, WithIterPublicNodesOnly(),
)
publicNodes, err := fn.CollectErr(publicIter)
require.NoError(t, err)
require.Len(t, publicNodes, 2)
// Verify the returned nodes are exactly the public ones.
pub1Key := publicNode1.PubKeyBytes
pub2Key := publicNode2.PubKeyBytes
for _, node := range publicNodes {
require.True(
t, node.PubKeyBytes == pub1Key ||
node.PubKeyBytes == pub2Key,
"unexpected node in public-only results: %x",
node.PubKeyBytes,
)
}
}
// testNodeUpdatesWithBatchSize is a helper function that tests node updates
// with a specific batch size to ensure the iterator works correctly across
// batch boundaries.
func testNodeUpdatesWithBatchSize(t *testing.T, ctx context.Context,
batchSize int) {
// Create a fresh graph for each test.
testGraph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
// Add 25 nodes with increasing timestamps.
startTime := time.Unix(1234567890, 0)
var nodeAnns []models.Node
for i := 0; i < 25; i++ {
nodeAnn := createTestVertex(t, lnwire.GossipVersion1)
nodeAnn.LastUpdate = startTime.Add(
time.Duration(i) * time.Hour,
)
nodeAnns = append(nodeAnns, *nodeAnn)
require.NoError(
t, testGraph.AddNode(ctx, nodeAnn),
)
}
testCases := []struct {
name string
start time.Time
end time.Time
want int
}{
{
name: "all nodes",
start: startTime,
end: startTime.Add(26 * time.Hour),
want: 25,
},
// The end time is exclusive per BOLT 07, so we
// add one extra hour to include the last node in
// the desired range.
{
name: "first batch only",
start: startTime,
end: startTime.Add(
time.Duration(
min(batchSize, 25),
) * time.Hour,
),
want: min(batchSize, 25),
},
{
name: "cross batch boundary",
start: startTime,
end: startTime.Add(
time.Duration(
min(batchSize+1, 25),
) * time.Hour,
),
want: min(batchSize+1, 25),
},
{
name: "exact boundary",
start: func() time.Time {
// Test querying exactly at a
// batch boundary.
if batchSize <= 25 {
return startTime.Add(
time.Duration(
batchSize-1,
) * time.Hour,
)
}
// For batch sizes > 25, test
// beyond our data range.
return startTime.Add(
time.Duration(25) * time.Hour,
)
}(),
end: func() time.Time {
// End is exclusive, so we add
// one hour to include the node
// at exactly the start time.
if batchSize <= 25 {
return startTime.Add(
time.Duration(
batchSize,
) * time.Hour,
)
}
return startTime.Add(
time.Duration(26) * time.Hour,
)
}(),
want: func() int {
if batchSize <= 25 {
return 1
}
// No nodes exist at hour 25 or
// beyond.
return 0
}(),
},
{
name: "empty range before",
start: startTime.Add(-time.Hour),
end: startTime.Add(-time.Minute),
want: 0,
},
{
name: "empty range after",
start: startTime.Add(30 * time.Hour),
end: startTime.Add(40 * time.Hour),
want: 0,
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
iter := testGraph.NodeUpdatesInHorizon(
ctx, NodeUpdateRange{
StartTime: fn.Some(tc.start),
EndTime: fn.Some(tc.end),
},
WithNodeUpdateIterBatchSize(
batchSize,
),
)
nodes, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Len(
t, nodes, tc.want,
"expected %d nodes, got %d",
tc.want, len(nodes),
)
// Verify nodes are in the correct time
// order.
for i := 1; i < len(nodes); i++ {
require.True(t,
nodes[i-1].LastUpdate.Before(
nodes[i].LastUpdate,
) || nodes[i-1].LastUpdate.Equal(
nodes[i].LastUpdate,
),
"nodes should be in "+
"chronological order",
)
}
})
}
}
// TestNodeUpdatesInHorizonBoundaryConditions tests the iterator boundary
// conditions, specifically around batch boundaries and edge cases.
func TestNodeUpdatesInHorizonBoundaryConditions(t *testing.T) {
t.Parallel()
ctx := t.Context()
// Test with various batch sizes to ensure the iterator works correctly
// across batch boundaries.
batchSizes := []int{1, 3, 5, 10, 25, 100}
for _, batchSize := range batchSizes {
testName := fmt.Sprintf("BatchSize%d", batchSize)
t.Run(testName, func(t *testing.T) {
testNodeUpdatesWithBatchSize(t, ctx, batchSize)
})
}
}
// TestNodeUpdatesInHorizonEarlyTermination tests that the iterator properly
// handles early termination when the caller stops iterating.
func TestNodeUpdatesInHorizonEarlyTermination(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
// We'll start by creating 100 nodes, each with an update time spaced
// one hour apart.
startTime := time.Unix(1234567890, 0)
for i := 0; i < 100; i++ {
nodeAnn := createTestVertex(t, lnwire.GossipVersion1)
nodeAnn.LastUpdate = startTime.Add(time.Duration(i) * time.Hour)
require.NoError(t, graph.AddNode(ctx, nodeAnn))
}
// Test early termination at various points
terminationPoints := []int{0, 1, 5, 10, 23, 50, 99}
for _, stopAt := range terminationPoints {
t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) {
iter := graph.NodeUpdatesInHorizon(
ctx, NodeUpdateRange{
StartTime: fn.Some(startTime),
EndTime: fn.Some(
startTime.Add(200 * time.Hour),
),
},
WithNodeUpdateIterBatchSize(10),
)
// Collect only up to stopAt nodes, breaking afterwards.
var collected []*models.Node
count := 0
for node := range iter {
if count >= stopAt {
break
}
collected = append(collected, node)
count++
}
require.Len(
t, collected, stopAt,
"should have collected exactly %d nodes",
stopAt,
)
})
}
}
// TestChanUpdatesInHorizonBoundaryConditions tests the channel iterator
// boundary conditions.
func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) {
t.Parallel()
ctx := t.Context()
batchSizes := []int{1, 3, 5, 10}
for _, batchSize := range batchSizes {
testName := fmt.Sprintf("BatchSize%d", batchSize)
t.Run(testName, func(t *testing.T) {
// Create a fresh graph for each test, then add two new
// nodes to the graph.
graph := NewVersionedGraph(
MakeTestGraph(t), lnwire.GossipVersion1,
)
node1 := createTestVertex(t, lnwire.GossipVersion1)
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node1))
require.NoError(t, graph.AddNode(ctx, node2))
// Next, we'll create 25 channels between the two nodes,
// each with increasing timestamps.
startTime := time.Unix(1234567890, 0)
const numChans = 25
for i := 0; i < numChans; i++ {
updateTime := startTime.Add(
time.Duration(i) * time.Hour,
)
channel, chanID := createEdge(
lnwire.GossipVersion1, uint32(i*10), 0,
0, 0, node1, node2,
)
require.NoError(
t, graph.AddChannelEdge(ctx, channel),
)
edge1 := newEdgePolicy(
lnwire.GossipVersion1,
chanID.ToUint64(), updateTime.Unix(),
true,
)
edge1.ChannelFlags = 0
edge1.ToNode = node2.PubKeyBytes
edge1.SigBytes = testSig.Serialize()
require.NoError(
t, graph.UpdateEdgePolicy(ctx, edge1),
)
edge2 := newEdgePolicy(
lnwire.GossipVersion1,
chanID.ToUint64(), updateTime.Unix(),
false,
)
edge2.ChannelFlags = 1
edge2.ToNode = node1.PubKeyBytes
edge2.SigBytes = testSig.Serialize()
require.NoError(
t, graph.UpdateEdgePolicy(ctx, edge2),
)
}
// Now we'll run the main query, and verify that we get
// back the expected number of channels.
iter := graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartTime: fn.Some(startTime),
EndTime: fn.Some(
startTime.Add(26 * time.Hour),
),
},
WithChanUpdateIterBatchSize(batchSize),
)
channels, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Len(
t, channels, numChans,
"expected %d channels, got %d", numChans,
len(channels),
)
})
}
}
// TestNodeUpdatesInHorizonExclusiveEnd verifies that NodeUpdatesInHorizon uses
// an exclusive end time per BOLT 07: "timestamp is greater or equal to
// first_timestamp, and less than first_timestamp plus timestamp_range".
func TestNodeUpdatesInHorizonExclusiveEnd(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
// Create three nodes at timestamps 100, 200, and 300.
timestamps := []int64{100, 200, 300}
for _, ts := range timestamps {
node := createTestVertex(t, lnwire.GossipVersion1)
node.LastUpdate = time.Unix(ts, 0)
require.NoError(t, graph.AddNode(ctx, node))
}
tests := []struct {
name string
start time.Time
end time.Time
want int
}{
{
// Start is inclusive: node at exactly startTime
// should be included.
name: "start time is inclusive",
start: time.Unix(100, 0),
end: time.Unix(101, 0),
want: 1,
},
{
// End is exclusive: node at exactly endTime should
// NOT be included.
name: "end time is exclusive",
start: time.Unix(100, 0),
end: time.Unix(200, 0),
want: 1,
},
{
// One second past the boundary includes the node.
name: "one past end includes boundary node",
start: time.Unix(100, 0),
end: time.Unix(201, 0),
want: 2,
},
{
// Range [200, 300) should include node at 200 but
// not node at 300.
name: "mid range excludes end",
start: time.Unix(200, 0),
end: time.Unix(300, 0),
want: 1,
},
{
// Range [200, 301) should include nodes at 200
// and 300.
name: "mid range includes end plus one",
start: time.Unix(200, 0),
end: time.Unix(301, 0),
want: 2,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
iter := graph.NodeUpdatesInHorizon(
ctx, NodeUpdateRange{
StartTime: fn.Some(tc.start),
EndTime: fn.Some(tc.end),
},
)
nodes, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Len(t, nodes, tc.want)
})
}
}
// TestNodeUpdatesInHorizonV2 tests that NodeUpdatesInHorizon works correctly
// for v2 gossip using block-height-based ranges with [start, end) semantics.
func TestNodeUpdatesInHorizonV2(t *testing.T) {
t.Parallel()
if !isSQLDB {
t.Skip("v2 gossip only supported with SQL backend")
}
ctx := t.Context()
graph := NewVersionedGraph(
MakeTestGraph(t), lnwire.GossipVersion2,
)
// Query before any nodes exist — should return empty.
iter := graph.NodeUpdatesInHorizon(
ctx, NodeUpdateRange{
StartHeight: fn.Some(uint32(0)),
EndHeight: fn.Some(uint32(9999)),
},
)
nodes, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Empty(t, nodes)
// Create 10 v2 nodes at block heights 100, 110, 120, ..., 190.
const numNodes = 10
const startHeight uint32 = 100
const heightStep uint32 = 10
nodeAnns := make([]models.Node, 0, numNodes)
for i := 0; i < numNodes; i++ {
node := createTestVertex(t, lnwire.GossipVersion2)
node.LastBlockHeight = startHeight + uint32(i)*heightStep
nodeAnns = append(nodeAnns, *node)
require.NoError(t, graph.AddNode(ctx, node))
}
// endHeight is one past the last node's height (exclusive).
endHeight := startHeight + uint32(numNodes)*heightStep
tests := []struct {
name string
start uint32
end uint32
want int
}{
{
// Range strictly below all nodes.
name: "below range",
start: 0,
end: 50,
want: 0,
},
{
// Range strictly above all nodes.
name: "above range",
start: 500,
end: 600,
want: 0,
},
{
// Start is inclusive: node at exactly startHeight
// should be included.
name: "start height is inclusive",
start: startHeight,
end: startHeight + 1,
want: 1,
},
{
// End is exclusive: node at exactly endHeight-10
// (=190) should NOT be included when end=190.
name: "end height is exclusive",
start: startHeight,
end: endHeight - heightStep,
want: numNodes - 1,
},
{
// One past the last node includes it.
name: "one past end includes last",
start: startHeight,
end: endHeight - heightStep + 1,
want: numNodes,
},
{
// Full range returns all nodes.
name: "full range",
start: startHeight,
end: endHeight,
want: numNodes,
},
{
// Skip the first node.
name: "skip first",
start: startHeight + heightStep,
end: endHeight,
want: numNodes - 1,
},
{
// Middle slice: heights [120, 170) = nodes at
// 120, 130, 140, 150, 160 = 5 nodes.
name: "middle slice",
start: 120,
end: 170,
want: 5,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
iter := graph.NodeUpdatesInHorizon(
ctx, NodeUpdateRange{
StartHeight: fn.Some(tc.start),
EndHeight: fn.Some(tc.end),
},
)
results, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Len(t, results, tc.want)
// Verify nodes are in ascending block height
// order.
for i := 1; i < len(results); i++ {
require.LessOrEqual(
t,
results[i-1].LastBlockHeight,
results[i].LastBlockHeight,
"nodes should be in ascending "+
"block height order",
)
}
})
}
}
// TestChanUpdatesInHorizonExclusiveEnd verifies that ChanUpdatesInHorizon uses
// an exclusive end time per BOLT 07: "timestamp is greater or equal to
// first_timestamp, and less than first_timestamp plus timestamp_range".
func TestChanUpdatesInHorizonExclusiveEnd(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
node1 := createTestVertex(t, lnwire.GossipVersion1)
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node1))
require.NoError(t, graph.AddNode(ctx, node2))
// Create three channels with policy updates at timestamps 100, 200,
// and 300.
timestamps := []int64{100, 200, 300}
for i, ts := range timestamps {
channel, chanID := createEdge(
lnwire.GossipVersion1, uint32(i*10), 0, 0, 0,
node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel))
edge := newEdgePolicy(
lnwire.GossipVersion1, chanID.ToUint64(), ts, true,
)
edge.ChannelFlags = 0
edge.ToNode = node2.PubKeyBytes
edge.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge))
}
tests := []struct {
name string
start time.Time
end time.Time
want int
}{
{
// Start is inclusive: channel at exactly startTime
// should be included.
name: "start time is inclusive",
start: time.Unix(100, 0),
end: time.Unix(101, 0),
want: 1,
},
{
// End is exclusive: channel at exactly endTime
// should NOT be included.
name: "end time is exclusive",
start: time.Unix(100, 0),
end: time.Unix(200, 0),
want: 1,
},
{
// One second past the boundary includes the
// channel.
name: "one past end includes boundary channel",
start: time.Unix(100, 0),
end: time.Unix(201, 0),
want: 2,
},
{
// Range [200, 300) should include channel at 200
// but not channel at 300.
name: "mid range excludes end",
start: time.Unix(200, 0),
end: time.Unix(300, 0),
want: 1,
},
{
// Range [200, 301) should include channels at 200
// and 300.
name: "mid range includes end plus one",
start: time.Unix(200, 0),
end: time.Unix(301, 0),
want: 2,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
iter := graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartTime: fn.Some(tc.start),
EndTime: fn.Some(tc.end),
},
)
channels, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Len(t, channels, tc.want)
})
}
}
// TestChanUpdatesInHorizonV2 tests that ChanUpdatesInHorizon works correctly
// for v2 gossip using block-height-based ranges with [start, end) semantics.
func TestChanUpdatesInHorizonV2(t *testing.T) {
t.Parallel()
if !isSQLDB {
t.Skip("v2 gossip only supported with SQL backend")
}
ctx := t.Context()
graph := NewVersionedGraph(
MakeTestGraph(t), lnwire.GossipVersion2,
)
node1 := createTestVertex(t, lnwire.GossipVersion2)
node2 := createTestVertex(t, lnwire.GossipVersion2)
require.NoError(t, graph.AddNode(ctx, node1))
require.NoError(t, graph.AddNode(ctx, node2))
// Query before any channels exist — should return empty.
iter := graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartHeight: fn.Some(uint32(0)),
EndHeight: fn.Some(uint32(9999)),
},
)
channels, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Empty(t, channels)
// Create 10 v2 channels with policy block heights at
// 100, 110, 120, ..., 190.
const numChans = 10
const startHeight uint32 = 100
const heightStep uint32 = 10
for i := 0; i < numChans; i++ {
height := startHeight + uint32(i)*heightStep
channel, chanID := createEdge(
lnwire.GossipVersion2, uint32(i*10), 0, 0, 0,
node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel))
edge1 := newEdgePolicy(
lnwire.GossipVersion2, chanID.ToUint64(), 0, true,
)
edge1.LastBlockHeight = height
edge1.ToNode = node2.PubKeyBytes
edge1.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
edge2 := newEdgePolicy(
lnwire.GossipVersion2, chanID.ToUint64(), 0, false,
)
edge2.LastBlockHeight = height
edge2.ToNode = node1.PubKeyBytes
edge2.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
}
endHeight := startHeight + uint32(numChans)*heightStep
tests := []struct {
name string
start uint32
end uint32
want int
}{
{
name: "below range",
start: 0,
end: 50,
want: 0,
},
{
name: "above range",
start: 500,
end: 600,
want: 0,
},
{
name: "start height is inclusive",
start: startHeight,
end: startHeight + 1,
want: 1,
},
{
// End is exclusive: channel at exactly
// endHeight-10 (=190) should NOT be included
// when end=190.
name: "end height is exclusive",
start: startHeight,
end: endHeight - heightStep,
want: numChans - 1,
},
{
name: "one past end includes last",
start: startHeight,
end: endHeight - heightStep + 1,
want: numChans,
},
{
name: "full range",
start: startHeight,
end: endHeight,
want: numChans,
},
{
name: "skip first",
start: startHeight + heightStep,
end: endHeight,
want: numChans - 1,
},
{
// Heights [120, 170) = channels at
// 120, 130, 140, 150, 160 = 5 channels.
name: "middle slice",
start: 120,
end: 170,
want: 5,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
iter := graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartHeight: fn.Some(tc.start),
EndHeight: fn.Some(tc.end),
},
)
results, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Len(t, results, tc.want)
})
}
// Test with asymmetric policy block heights: one policy inside
// the range, the other outside. The SQL query uses OR across the
// two policies, so the channel should still be returned if
// either policy is in range.
t.Run("asymmetric policy heights", func(t *testing.T) {
channel, chanID := createEdge(
lnwire.GossipVersion2, 500, 0, 0, 0,
node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel))
// Policy 1 at height 300 (inside range).
edge1 := newEdgePolicy(
lnwire.GossipVersion2,
chanID.ToUint64(), 0, true,
)
edge1.LastBlockHeight = 300
edge1.ToNode = node2.PubKeyBytes
edge1.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
// Policy 2 at height 900 (outside range).
edge2 := newEdgePolicy(
lnwire.GossipVersion2,
chanID.ToUint64(), 0, false,
)
edge2.LastBlockHeight = 900
edge2.ToNode = node1.PubKeyBytes
edge2.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
// Query [250, 350) — only policy 1 is in range, but the
// channel should still be returned.
iter := graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartHeight: fn.Some(uint32(250)),
EndHeight: fn.Some(uint32(350)),
},
)
results, err := fn.CollectErr(iter)
require.NoError(t, err)
require.Len(t, results, 1)
// Query [850, 950) — only policy 2 is in range, channel
// should still be returned.
iter = graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartHeight: fn.Some(uint32(850)),
EndHeight: fn.Some(uint32(950)),
},
)
results, err = fn.CollectErr(iter)
require.NoError(t, err)
require.Len(t, results, 1)
// Query [400, 500) — neither policy is in range.
iter = graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartHeight: fn.Some(uint32(400)),
EndHeight: fn.Some(uint32(500)),
},
)
results, err = fn.CollectErr(iter)
require.NoError(t, err)
require.Empty(t, results)
})
}
// testFilterKnownChanIDsZombieRevival tests that if a ChannelUpdateInfo is
// passed to FilterKnownChanIDs that contains a channel that we have marked as
// a zombie, then we will mark it as live again if the new ChannelUpdate has
// timestamps that would make the channel be considered live again.
//
// NOTE: this test focuses on zombie revival. The main logic of
// FilterKnownChanIDs is tested in testFilterKnownChanIDs.
func testFilterKnownChanIDsZombieRevival(t *testing.T,
v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
var (
scid1 = lnwire.ShortChannelID{BlockHeight: 1}
scid2 = lnwire.ShortChannelID{BlockHeight: 2}
scid3 = lnwire.ShortChannelID{BlockHeight: 3}
)
vGraph := NewVersionedGraph(graph, v)
isZombie := func(scid lnwire.ShortChannelID) bool {
zombie, _, _, err := vGraph.IsZombieEdge(
ctx, scid.ToUint64(),
)
require.NoError(t, err)
return zombie
}
// Mark channel 1 and 2 as zombies.
err := graph.MarkEdgeZombie(
ctx, v, scid1.ToUint64(), [33]byte{}, [33]byte{},
)
require.NoError(t, err)
err = graph.MarkEdgeZombie(
ctx, v, scid2.ToUint64(), [33]byte{}, [33]byte{},
)
require.NoError(t, err)
require.True(t, isZombie(scid1))
require.True(t, isZombie(scid2))
require.False(t, isZombie(scid3))
// Build a freshness marker appropriate for the gossip version. V1
// uses unix timestamps, v2 uses block heights.
var revivalFreshness lnwire.Timestamp
switch v {
case lnwire.GossipVersion1:
revivalFreshness = lnwire.UnixTimestamp(1000)
case lnwire.GossipVersion2:
revivalFreshness = lnwire.BlockHeightTimestamp(1000)
}
// Call FilterKnownChanIDs with an isStillZombie call-back that would
// result in the current zombies still be considered as zombies.
_, err = vGraph.FilterKnownChanIDs(ctx, []ChannelUpdateInfo{
{ShortChannelID: scid1, Version: v},
{ShortChannelID: scid2, Version: v},
{ShortChannelID: scid3, Version: v},
}, func(_ ChannelUpdateInfo) bool {
return true
})
require.NoError(t, err)
require.True(t, isZombie(scid1))
require.True(t, isZombie(scid2))
require.False(t, isZombie(scid3))
// Now call it again but this time with an isStillZombie call-back
// that would result in channel with SCID 2 no longer being
// considered a zombie.
_, err = vGraph.FilterKnownChanIDs(ctx, []ChannelUpdateInfo{
{ShortChannelID: scid1, Version: v},
{
ShortChannelID: scid2,
Version: v,
Node1Freshness: revivalFreshness,
},
{ShortChannelID: scid3, Version: v},
}, func(info ChannelUpdateInfo) bool {
return info.Node1Freshness != revivalFreshness
})
require.NoError(t, err)
// Show that SCID 2 has been marked as live.
require.True(t, isZombie(scid1))
require.False(t, isZombie(scid2))
require.False(t, isZombie(scid3))
}
// testFilterKnownChanIDs tests that we're able to properly perform the set
// differences of an incoming set of channel ID's, and those that we already
// know of on disk.
func testFilterKnownChanIDs(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
vGraph := NewVersionedGraph(graph, v)
isZombieUpdate := func(_ ChannelUpdateInfo) bool {
return true
}
// newChanUpdateInfo builds a ChannelUpdateInfo for the given SCID with
// the test's gossip version and zero freshness.
newChanUpdateInfo := func(
scid lnwire.ShortChannelID,
) ChannelUpdateInfo {
return ChannelUpdateInfo{
ShortChannelID: scid,
Version: v,
}
}
var (
scid1 = lnwire.ShortChannelID{BlockHeight: 1}
scid2 = lnwire.ShortChannelID{BlockHeight: 2}
scid3 = lnwire.ShortChannelID{BlockHeight: 3}
)
// If we try to filter out a set of channel ID's before we even know of
// any channels, then we should get the entire set back.
preChanIDs := []ChannelUpdateInfo{
newChanUpdateInfo(scid1),
newChanUpdateInfo(scid2),
newChanUpdateInfo(scid3),
}
filteredIDs, err := vGraph.FilterKnownChanIDs(
ctx, preChanIDs, isZombieUpdate,
)
require.NoError(t, err, "unable to filter chan IDs")
require.EqualValues(t, []uint64{
scid1.ToUint64(),
scid2.ToUint64(),
scid3.ToUint64(),
}, filteredIDs)
// We'll start by creating two nodes which will seed our test graph.
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node2))
// Next, we'll add 5 channel ID's to the graph, each of them having a
// block height 10 blocks after the previous.
const numChans = 5
chanIDs := make([]ChannelUpdateInfo, 0, numChans)
for i := 0; i < numChans; i++ {
channel, chanID := createEdge(
v, uint32(i*10), 0, 0, 0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel))
chanIDs = append(chanIDs, newChanUpdateInfo(chanID))
}
const numZombies = 5
zombieIDs := make([]ChannelUpdateInfo, 0, numZombies)
for i := 0; i < numZombies; i++ {
channel, chanID := createEdge(
v, uint32(i*10+1), 0, 0, 0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel))
err := graph.DeleteChannelEdges(
ctx, v, false, true, channel.ChannelID,
)
require.NoError(t, err)
zombieIDs = append(zombieIDs, newChanUpdateInfo(chanID))
}
queryCases := []struct {
queryIDs []ChannelUpdateInfo
resp []ChannelUpdateInfo
}{
// If we attempt to filter out all chanIDs we know of, the
// response should be the empty set.
{
queryIDs: chanIDs,
},
// If we attempt to filter out all zombies that we know of,
// the response should be the empty set.
{
queryIDs: zombieIDs,
},
// If we query for a set of ID's that we didn't insert, we
// should get the same set back.
{
queryIDs: []ChannelUpdateInfo{
newChanUpdateInfo(lnwire.ShortChannelID{
BlockHeight: 99,
}),
newChanUpdateInfo(lnwire.ShortChannelID{
BlockHeight: 100,
}),
},
resp: []ChannelUpdateInfo{
newChanUpdateInfo(lnwire.ShortChannelID{
BlockHeight: 99,
}),
newChanUpdateInfo(lnwire.ShortChannelID{
BlockHeight: 100,
}),
},
},
// If we query for a super-set of our the chan ID's inserted,
// we should only get those new chanIDs back.
{
queryIDs: append(chanIDs, []ChannelUpdateInfo{
newChanUpdateInfo(lnwire.ShortChannelID{
BlockHeight: 99,
}),
newChanUpdateInfo(lnwire.ShortChannelID{
BlockHeight: 101,
}),
}...),
resp: []ChannelUpdateInfo{
newChanUpdateInfo(lnwire.ShortChannelID{
BlockHeight: 99,
}),
newChanUpdateInfo(lnwire.ShortChannelID{
BlockHeight: 101,
}),
},
},
}
for _, queryCase := range queryCases {
resp, err := vGraph.FilterKnownChanIDs(
ctx, queryCase.queryIDs, isZombieUpdate,
)
require.NoError(t, err)
expectedSCIDs := make([]uint64, len(queryCase.resp))
for i, info := range queryCase.resp {
expectedSCIDs[i] = info.ShortChannelID.ToUint64()
}
if len(expectedSCIDs) == 0 {
expectedSCIDs = nil
}
require.EqualValues(t, expectedSCIDs, resp)
}
}
// TestStressTestChannelGraphAPI is a stress test that concurrently calls some
// of the ChannelGraph methods in various orders in order to ensure that no
// deadlock can occur. This test currently focuses on stress testing all the
// methods that acquire the cache mutex along with the DB mutex.
func TestStressTestChannelGraphAPI(t *testing.T) {
t.Parallel()
if testing.Short() {
t.Skipf("Skipping test in short mode")
}
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node2))
// We need to update the node's timestamp since this call to
// SetSourceNode will trigger an upsert which will only be allowed if
// the newest LastUpdate time is greater than the current one.
node1.LastUpdate = node1.LastUpdate.Add(time.Second)
require.NoError(t, graph.SetSourceNode(ctx, node1))
type chanInfo struct {
info models.ChannelEdgeInfo
id lnwire.ShortChannelID
}
var (
chans []*chanInfo
mu sync.RWMutex
)
// newBlockHeight returns a random block height between 0 and 100.
newBlockHeight := func() uint32 {
return uint32(rand.Int31n(100))
}
// addNewChan is a will create and return a new random channel and will
// add it to the set of channels.
addNewChan := func() *chanInfo {
mu.Lock()
defer mu.Unlock()
channel, chanID := createEdge(
lnwire.GossipVersion1, newBlockHeight(),
rand.Uint32(), uint16(rand.Int()), rand.Uint32(),
node1, node2,
)
newChan := &chanInfo{
info: *channel,
id: chanID,
}
chans = append(chans, newChan)
return newChan
}
// getRandChan picks a random channel from the set and returns it.
getRandChan := func() *chanInfo {
mu.RLock()
defer mu.RUnlock()
if len(chans) == 0 {
return nil
}
return chans[rand.Intn(len(chans))]
}
// getRandChanSet returns a random set of channels.
getRandChanSet := func() []*chanInfo {
mu.RLock()
defer mu.RUnlock()
if len(chans) == 0 {
return nil
}
start := rand.Intn(len(chans))
end := rand.Intn(len(chans))
if end < start {
start, end = end, start
}
var infoCopy []*chanInfo
for i := start; i < end; i++ {
infoCopy = append(infoCopy, &chanInfo{
info: chans[i].info,
id: chans[i].id,
})
}
return infoCopy
}
// delChan deletes the channel with the given ID from the set if it
// exists.
delChan := func(id lnwire.ShortChannelID) {
mu.Lock()
defer mu.Unlock()
index := -1
for i, c := range chans {
if c.id == id {
index = i
break
}
}
if index == -1 {
return
}
chans = append(chans[:index], chans[index+1:]...)
}
var blockHash chainhash.Hash
copy(blockHash[:], bytes.Repeat([]byte{2}, 32))
var methodsMu sync.Mutex
methods := []struct {
name string
fn func() error
}{
{
name: "MarkEdgeZombie",
fn: func() error {
channel := getRandChan()
if channel == nil {
return nil
}
return graph.MarkEdgeZombie(
ctx, lnwire.GossipVersion1,
channel.id.ToUint64(),
node1.PubKeyBytes,
node2.PubKeyBytes,
)
},
},
{
name: "FilterKnownChanIDs",
fn: func() error {
chanSet := getRandChanSet()
var chanIDs []ChannelUpdateInfo
ver := lnwire.GossipVersion1
for _, c := range chanSet {
info := ChannelUpdateInfo{
ShortChannelID: c.id,
Version: ver,
}
chanIDs = append(chanIDs, info)
}
_, err := graph.FilterKnownChanIDs(
ctx, chanIDs,
func(_ ChannelUpdateInfo) bool {
return rand.Intn(2) == 0
},
)
return err
},
},
{
name: "HasChannelEdge",
fn: func() error {
channel := getRandChan()
if channel == nil {
return nil
}
_, _, err := graph.HasChannelEdge(
ctx, channel.id.ToUint64(),
)
return err
},
},
{
name: "PruneGraph",
fn: func() error {
chanSet := getRandChanSet()
var spentOutpoints []*wire.OutPoint
for _, c := range chanSet {
spentOutpoints = append(
spentOutpoints,
&c.info.ChannelPoint,
)
}
_, err := graph.PruneGraph(
ctx, spentOutpoints, &blockHash, 100,
)
return err
},
},
{
name: "ChanUpdateInHorizon",
fn: func() error {
now := time.Now()
iter := graph.ChanUpdatesInHorizon(
ctx, ChanUpdateRange{
StartTime: fn.Some(
now.Add(-time.Hour),
),
EndTime: fn.Some(now),
},
)
_, err := fn.CollectErr(iter)
return err
},
},
{
name: "DeleteChannelEdges",
fn: func() error {
var (
strictPruning = rand.Intn(2) == 0
markZombie = rand.Intn(2) == 0
channels = getRandChanSet()
chanIDs []uint64
)
for _, c := range channels {
chanIDs = append(
chanIDs, c.id.ToUint64(),
)
delChan(c.id)
}
err := graph.DeleteChannelEdges(
ctx, strictPruning, markZombie,
chanIDs...,
)
if err != nil &&
!errors.Is(err, ErrEdgeNotFound) {
return err
}
return nil
},
},
{
name: "DisconnectBlockAtHeight",
fn: func() error {
_, err := graph.DisconnectBlockAtHeight(
ctx, newBlockHeight(),
)
return err
},
},
{
name: "AddChannelEdge",
fn: func() error {
channel := addNewChan()
return graph.AddChannelEdge(ctx, &channel.info)
},
},
}
const (
// concurrencyLevel is the number of concurrent goroutines that
// will be run simultaneously.
concurrencyLevel = 10
// executionCount is the number of methods that will be called
// per goroutine.
executionCount = 100
)
for i := 0; i < concurrencyLevel; i++ {
t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
t.Parallel()
for j := 0; j < executionCount; j++ {
// Randomly select a method to execute.
methodIndex := rand.Intn(len(methods))
methodsMu.Lock()
fn := methods[methodIndex].fn
name := methods[methodIndex].name
methodsMu.Unlock()
err := fn()
require.NoErrorf(t, err, name)
}
})
}
}
// TestFilterChannelRange tests that we're able to properly retrieve the full
// set of short channel ID's for a given block range.
func TestFilterChannelRange(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
// 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)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node2))
// If we try to filter a channel range before we have any channels
// inserted, we should get an empty slice of results.
resp, err := graph.FilterChannelRange(
ctx, lnwire.GossipVersion1, 10, 100, false,
)
require.NoError(t, err)
require.Empty(t, resp)
// To start, we'll create a set of channels, two mined in a block 10
// blocks after the prior one.
startHeight := uint32(100)
endHeight := startHeight
const numChans = 10
var (
channelRanges = make(
[]BlockChannelRange, 0, numChans/2,
)
channelRangesWithTimestamps = make(
[]BlockChannelRange, 0, numChans/2,
)
)
updateTimeSeed := time.Now().Unix()
maybeAddPolicy := func(chanID uint64, node *models.Node,
node2 bool) time.Time {
var chanFlags lnwire.ChanUpdateChanFlags
if node2 {
chanFlags = lnwire.ChanUpdateDirection
}
var updateTime = time.Unix(0, 0)
if rand.Int31n(2) == 0 {
updateTime = time.Unix(updateTimeSeed, 0)
err = graph.UpdateEdgePolicy(
ctx, &models.ChannelEdgePolicy{
Version: lnwire.GossipVersion1,
ToNode: node.PubKeyBytes,
ChannelFlags: chanFlags,
ChannelID: chanID,
LastUpdate: updateTime,
},
)
require.NoError(t, err)
}
updateTimeSeed++
return updateTime
}
for i := 0; i < numChans/2; i++ {
chanHeight := endHeight
channel1, chanID1 := createEdge(
lnwire.GossipVersion1, chanHeight, uint32(i+1), 0,
0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel1))
channel2, chanID2 := createEdge(
lnwire.GossipVersion1, chanHeight, uint32(i+2), 0,
0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel2))
chanInfo1 := NewV1ChannelUpdateInfo(
chanID1, time.Time{}, time.Time{},
)
chanInfo2 := NewV1ChannelUpdateInfo(
chanID2, time.Time{}, time.Time{},
)
channelRanges = append(channelRanges, BlockChannelRange{
Height: chanHeight,
Channels: []ChannelUpdateInfo{
chanInfo1, chanInfo2,
},
})
var (
time1 = maybeAddPolicy(channel1.ChannelID, node1, false)
time2 = maybeAddPolicy(channel1.ChannelID, node2, true)
time3 = maybeAddPolicy(channel2.ChannelID, node1, false)
time4 = maybeAddPolicy(channel2.ChannelID, node2, true)
)
chanInfo1 = NewV1ChannelUpdateInfo(chanID1, time1, time2)
chanInfo2 = NewV1ChannelUpdateInfo(chanID2, time3, time4)
channelRangesWithTimestamps = append(
channelRangesWithTimestamps, BlockChannelRange{
Height: chanHeight,
Channels: []ChannelUpdateInfo{
chanInfo1, chanInfo2,
},
},
)
endHeight += 10
}
// With our channels inserted, we'll construct a series of queries that
// we'll execute below in order to exercise the features of the
// FilterKnownChanIDs method.
tests := []struct {
name string
startHeight uint32
endHeight uint32
resp []BlockChannelRange
expStartIndex int
expEndIndex int
}{
// If we query for the entire range, then we should get the same
// set of short channel IDs back.
{
name: "entire range",
startHeight: startHeight,
endHeight: endHeight,
resp: channelRanges,
expStartIndex: 0,
expEndIndex: len(channelRanges),
},
// If we query for a range of channels right before our range,
// we shouldn't get any results back.
{
name: "range before",
startHeight: 0,
endHeight: 10,
},
// If we only query for the last height (range wise), we should
// only get that last channel.
{
name: "last height",
startHeight: endHeight - 10,
endHeight: endHeight - 10,
resp: channelRanges[4:],
expStartIndex: 4,
expEndIndex: len(channelRanges),
},
// If we query for just the first height, we should only get a
// single channel back (the first one).
{
name: "first height",
startHeight: startHeight,
endHeight: startHeight,
resp: channelRanges[:1],
expStartIndex: 0,
expEndIndex: 1,
},
{
name: "subset",
startHeight: startHeight + 10,
endHeight: endHeight - 10,
resp: channelRanges[1:5],
expStartIndex: 1,
expEndIndex: 5,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
t.Parallel()
// First, do the query without requesting timestamps.
resp, err := graph.FilterChannelRange(
ctx, lnwire.GossipVersion1, test.startHeight,
test.endHeight, false,
)
require.NoError(t, err)
expRes := channelRanges[test.expStartIndex:test.expEndIndex] //nolint:ll
if len(expRes) == 0 {
require.Nil(t, resp)
} else {
require.Equal(t, expRes, resp)
}
// Now, query the timestamps as well.
resp, err = graph.FilterChannelRange(
ctx, lnwire.GossipVersion1, test.startHeight,
test.endHeight, true,
)
require.NoError(t, err)
expRes = channelRangesWithTimestamps[test.expStartIndex:test.expEndIndex] //nolint:ll
if len(expRes) == 0 {
require.Nil(t, resp)
} else {
require.Equal(t, expRes, resp)
}
})
}
}
// TestFilterChannelRangeVersionGuard checks that FilterChannelRange correctly
// handles version-specific requests. For gossip v1, the KV store returns
// results as normal; for v2, the KV store returns
// ErrVersionNotSupportedForKVDB while the SQL store returns empty results
// (a v2-aware query is a follow-up).
func TestFilterChannelRangeVersionGuard(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
resp, err := store.FilterChannelRange(
ctx, lnwire.GossipVersion2, 0, 1000, false,
)
if isSQLDB {
// The SQL store accepts any known version and returns empty
// results since no v2 channels have been added.
require.NoError(t, err)
require.Empty(t, resp)
} else {
// The KV store does not support v2 and must return the
// sentinel error.
require.ErrorIs(t, err, ErrVersionNotSupportedForKVDB)
}
}
// 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, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
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, 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.
const numChans = 5
startTime := time.Unix(1234, 0)
endTime := startTime
edges := make([]ChannelEdge, 0, numChans)
edgeQuery := make([]uint64, 0, numChans)
for i := 0; i < numChans; i++ {
channel, chanID := createEdge(
v, uint32(i*10), 0, 0, 0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel))
updateTime := endTime
endTime = updateTime.Add(time.Second * 10)
edge1 := newEdgePolicy(
v, chanID.ToUint64(),
updateTime.Unix(), true,
)
if v == lnwire.GossipVersion1 {
edge1.ChannelFlags = 0
}
edge1.ToNode = node2.PubKeyBytes
edge1.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
edge2 := newEdgePolicy(
v, chanID.ToUint64(),
updateTime.Unix(), false,
)
if v == lnwire.GossipVersion1 {
edge2.ChannelFlags = 1
}
edge2.ToNode = node1.PubKeyBytes
edge2.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
edges = append(edges, ChannelEdge{
Info: channel,
Policy1: edge1,
Policy2: edge2,
})
edgeQuery = append(edgeQuery, chanID.ToUint64())
}
// Add an additional edge that does not exist. The query should skip
// this channel and return only infos for the edges that exist.
edgeQuery = append(edgeQuery, 500)
// 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(
v, 666, 0, 0, 0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, zombieChan))
err := graph.DeleteChannelEdges(
ctx, false, true, zombieChan.ChannelID,
)
require.NoError(t, err, "unable to delete and mark edge zombie")
edgeQuery = append(edgeQuery, zombieChanID.ToUint64())
// We'll now attempt to query for the range of channel ID's we just
// inserted into the database. We should get the exact same set of
// edges back.
resp, err := graph.FetchChanInfos(ctx, edgeQuery)
require.NoError(t, err, "unable to fetch chan edges")
require.Len(t, resp, len(edges))
for i := 0; i < len(resp); i++ {
compareEdgePolicies(t, resp[i].Policy1, edges[i].Policy1)
compareEdgePolicies(t, resp[i].Policy2, edges[i].Policy2)
assertEdgeInfoEqual(t, resp[i].Info, edges[i].Info)
}
}
// testChannelView tests that ChannelView returns the correct edge points for
// each active channel in the graph.
func testChannelView(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// Initially the channel view should be empty.
channelView, err := graph.ChannelView(ctx)
require.NoError(t, err)
require.Empty(t, channelView)
// Add some nodes and a set of channels between them.
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node2))
const numChans = 3
edgePoints := make([]EdgePoint, 0, numChans)
for i := 0; i < numChans; i++ {
edge, _ := createEdge(
v, uint32(i+1), 0, 0, uint32(i), node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, edge))
pkScript, err := edge.FundingPKScript()
require.NoError(t, err)
edgePoints = append(edgePoints, EdgePoint{
FundingPkScript: pkScript,
OutPoint: wire.OutPoint{
Hash: rev,
Index: uint32(i),
},
})
}
// Fetch the channel view and ensure it matches the expected edge
// points.
channelView, err = graph.ChannelView(ctx)
require.NoError(t, err)
assertChanViewEqual(t, channelView, edgePoints)
}
// testChannelViewTaprootV1RoundTrip tests that a taproot channel persisted as a
// v1 edge can be read back from ChannelView() with the correct taproot funding
// script.
func testChannelViewTaprootV1RoundTrip(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
if v != lnwire.GossipVersion1 {
t.Skip("only relevant for v1 taproot workaround channels")
}
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node2))
node1Pub, err := node1.PubKey()
require.NoError(t, err)
node2Pub, err := node2.PubKey()
require.NoError(t, err)
node1Vertex := route.NewVertex(node1Pub)
node2Vertex := route.NewVertex(node2Pub)
outpoint := wire.OutPoint{
Hash: rev,
Index: 1,
}
// Persist a synthetic v1 channel that advertises the taproot staging
// bit. This reproduces the serialization path exercised by older graph
// entries.
edgeInfo, err := models.NewV1Channel(
1, *chaincfg.MainNetParams.GenesisHash,
node1Vertex, node2Vertex,
&models.ChannelV1Fields{
BitcoinKey1Bytes: node1Vertex,
BitcoinKey2Bytes: node2Vertex,
ExtraOpaqueData: make([]byte, 0),
},
models.WithChannelPoint(outpoint),
models.WithCapacity(9000),
models.WithFeatures(lnwire.NewRawFeatureVector(
lnwire.SimpleTaprootChannelsRequiredStaging,
)),
)
require.NoError(t, err)
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
// The fix should make ChannelView reconstruct the taproot funding
// script for v1 channels that advertise the taproot staging bit.
expectedScript, _, err := input.GenTaprootFundingScript(
node1Pub, node2Pub, 0, fn.None[chainhash.Hash](),
)
require.NoError(t, err)
channelView, err := graph.ChannelView(ctx)
require.NoError(t, err)
require.Len(t, channelView, 1)
require.Equal(t, expectedScript, channelView[0].FundingPkScript)
require.Equal(t, outpoint, channelView[0].OutPoint)
}
// testIncompleteChannelPolicies tests that a channel that only has a policy
// specified on one end is properly returned in ForEachChannel calls from
// both sides.
func testIncompleteChannelPolicies(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// Create two nodes.
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node2))
channel, chanID := createEdge(
v, uint32(0), 0, 0, 0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, channel))
// Ensure that channel is reported with unknown policies.
checkPolicies := func(node *models.Node, expectedIn,
expectedOut bool) {
calls := 0
err := graph.ForEachNodeChannel(
ctx, node.PubKeyBytes,
func(_ *models.ChannelEdgeInfo, outEdge,
inEdge *models.ChannelEdgePolicy) error {
require.Equal(t, expectedOut, outEdge != nil)
require.Equal(t, expectedIn, inEdge != nil)
calls++
return nil
}, func() {},
)
require.NoError(t, err)
require.Equal(t, 1, calls)
}
checkPolicies(node2, false, false)
newTestEdgePolicy := func(isNode1 bool,
toNode route.Vertex) *models.ChannelEdgePolicy {
policy := newEdgePolicy(
v, chanID.ToUint64(), nextUpdateTime().Unix(), isNode1,
)
policy.ToNode = toNode
policy.SigBytes = testSig.Serialize()
return policy
}
// Only create an edge policy for node1 and leave the policy for node2
// unknown.
edgePolicy := newTestEdgePolicy(true, node2.PubKeyBytes)
require.NoError(t, graph.UpdateEdgePolicy(ctx, edgePolicy))
checkPolicies(node1, false, true)
checkPolicies(node2, true, false)
// Create second policy and assert that both policies are reported
// as present.
edgePolicy = newTestEdgePolicy(false, node1.PubKeyBytes)
require.NoError(t, graph.UpdateEdgePolicy(ctx, edgePolicy))
checkPolicies(node1, true, true)
checkPolicies(node2, true, true)
}
// TestChannelEdgePruningUpdateIndexDeletion tests that once edges are deleted
// from the graph, then their entries within the update index are also cleaned
// up.
func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
// The update index only applies to the bbolt graph.
boltStore, ok := graph.db.(*KVStore)
if !ok {
t.Skipf("skipping test that is aimed at a bbolt graph DB")
}
sourceNode := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.SetSourceNode(ctx, sourceNode))
// 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)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node2))
// With the two nodes created, we'll now create a random channel, as
// well as two edges in the database with distinct update times.
edgeInfo, chanID := createEdge(
lnwire.GossipVersion1, 100, 0, 0, 0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
edge1 := randEdgePolicy(chanID.ToUint64())
edge1.ChannelFlags = 0
edge1.ToNode = node1.PubKeyBytes
edge1.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
edge1 = copyEdgePolicy(edge1) // Avoid read/write race conditions.
edge2 := randEdgePolicy(chanID.ToUint64())
edge2.ChannelFlags = 1
edge2.ToNode = node2.PubKeyBytes
edge2.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
edge2 = copyEdgePolicy(edge2) // Avoid read/write race conditions.
// checkIndexTimestamps is a helper function that checks the edge update
// index only includes the given timestamps.
checkIndexTimestamps := func(timestamps ...uint64) {
timestampSet := make(map[uint64]struct{})
for _, t := range timestamps {
timestampSet[t] = struct{}{}
}
err := kvdb.View(boltStore.db, func(tx kvdb.RTx) error {
edges := tx.ReadBucket(edgeBucket)
if edges == nil {
return ErrGraphNoEdgesFound
}
edgeUpdateIndex := edges.NestedReadBucket(
edgeUpdateIndexBucket,
)
if edgeUpdateIndex == nil {
return ErrGraphNoEdgesFound
}
var numEntries int
err := edgeUpdateIndex.ForEach(func(k, v []byte) error {
numEntries++
return nil
})
if err != nil {
return err
}
expectedEntries := len(timestampSet)
if numEntries != expectedEntries {
return fmt.Errorf("expected %v entries in the "+
"update index, got %v", expectedEntries,
numEntries)
}
return edgeUpdateIndex.ForEach(func(k, _ []byte) error {
t := byteOrder.Uint64(k[:8])
if _, ok := timestampSet[t]; !ok {
return fmt.Errorf("found unexpected "+
"timestamp "+"%d", t)
}
return nil
})
}, func() {})
require.NoError(t, err)
}
// With both edges policies added, we'll make sure to check they exist
// within the edge update index.
checkIndexTimestamps(
uint64(edge1.LastUpdate.Unix()),
uint64(edge2.LastUpdate.Unix()),
)
// Now, we'll update the edge policies to ensure the old timestamps are
// removed from the update index.
edge1.ChannelFlags = 2
edge1.LastUpdate = time.Now()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
edge2.ChannelFlags = 3
edge2.LastUpdate = edge1.LastUpdate.Add(time.Hour)
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
// With the policies updated, we should now be able to find their
// updated entries within the update index.
checkIndexTimestamps(
uint64(edge1.LastUpdate.Unix()),
uint64(edge2.LastUpdate.Unix()),
)
// Now we'll prune the graph, removing the edges, and also the update
// index entries from the database all together.
var blockHash chainhash.Hash
copy(blockHash[:], bytes.Repeat([]byte{2}, 32))
_, err := graph.PruneGraph(
ctx, []*wire.OutPoint{&edgeInfo.ChannelPoint}, &blockHash,
101,
)
require.NoError(t, err, "unable to prune graph")
// Finally, we'll check the database state one last time to conclude
// that we should no longer be able to locate _any_ entries within the
// edge update index.
checkIndexTimestamps()
}
// TestPruneGraphNodes tests that unconnected vertexes are pruned via the
// PruneSyncState method.
func TestPruneGraphNodes(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
// We'll start off by inserting our source node, to ensure that it's
// the only node left after we prune the graph.
sourceNode := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.SetSourceNode(ctx, sourceNode))
// With the source node inserted, we'll now add three nodes to the
// channel graph, at the end of the scenario, only two of these nodes
// should still be in the graph.
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node2))
node3 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node3))
// We'll now add a new edge to the graph, but only actually advertise
// the edge of *one* of the nodes.
edgeInfo, chanID := createEdge(
lnwire.GossipVersion1, 100, 0, 0, 0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
// We'll now insert an advertised edge, but it'll only be the edge that
// points from the first to the second node.
edge1 := randEdgePolicy(chanID.ToUint64())
edge1.ChannelFlags = 0
edge1.ToNode = node1.PubKeyBytes
edge1.SigBytes = testSig.Serialize()
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
// We'll now initiate a around of graph pruning.
require.NoError(t, graph.PruneGraphNodes(ctx))
// At this point, there should be 3 nodes left in the graph still: the
// source node (which can't be pruned), and node 1+2. Nodes 1 and two
// should still be left in the graph as there's half of an advertised
// edge between them.
assertNumNodes(t, graph.ChannelGraph, 3)
// Finally, we'll ensure that node3, the only fully unconnected node as
// properly deleted from the graph and not another node in its place.
_, err := graph.FetchNode(ctx, node3.PubKeyBytes)
require.NotNil(t, err)
}
// 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, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
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, v)
require.NoError(t, graph.SetSourceNode(ctx, node1))
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(
v, 100, 0, 0, 0, node1, node2,
)
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
// Ensure that node1 was inserted as a full node, while node2 only has
// a shell node present.
node1, err := graph.FetchNode(ctx, node1.PubKeyBytes)
require.NoError(t, err, "unable to fetch node1")
require.True(t, node1.HaveAnnouncement())
node2, err = graph.FetchNode(ctx, node2.PubKeyBytes)
require.NoError(t, err, "unable to fetch node2")
require.False(t, node2.HaveAnnouncement())
// Show that attempting to add the channel again will result in an
// error.
err = graph.AddChannelEdge(ctx, edgeInfo)
require.ErrorIs(t, err, ErrEdgeAlreadyExist)
// Show that updating the shell node to a full node record works.
require.NoError(t, graph.AddNode(ctx, node2))
}
// TestNodePruningUpdateIndexDeletion tests that once a node has been removed
// from the channel graph, we also remove the entry from the update index as
// well.
// testNodePruningUpdateIndexDeletion verifies that deleting a node also removes
// it from the update index used by NodeUpdatesInHorizon.
func testNodePruningUpdateIndexDeletion(t *testing.T,
v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// We'll first populate our graph with a single node that will be
// removed shortly.
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
// Build a NodeUpdateRange that covers the node we just inserted. V1
// uses time-based ranges, v2 uses block-height-based ranges.
var updateRange NodeUpdateRange
switch v {
case lnwire.GossipVersion1:
updateRange = NodeUpdateRange{
StartTime: fn.Some(time.Unix(9, 0)),
EndTime: fn.Some(
node1.LastUpdate.Add(time.Minute),
),
}
case lnwire.GossipVersion2:
updateRange = NodeUpdateRange{
StartHeight: fn.Some(uint32(0)),
EndHeight: fn.Some(
node1.LastBlockHeight + 1,
),
}
}
// We'll confirm that we can retrieve the node using
// NodeUpdatesInHorizon.
nodesInHorizonIter := graph.NodeUpdatesInHorizon(
ctx, updateRange,
)
// We should only have a single node, and that node should exactly
// match the node we just inserted.
nodesInHorizon, err := fn.CollectErr(nodesInHorizonIter)
require.NoError(t, err, "unable to fetch nodes in horizon")
require.Len(t, nodesInHorizon, 1)
compareNodes(t, node1, nodesInHorizon[0])
// We'll now delete the node from the graph, this should result in it
// being removed from the update index as well.
err = graph.DeleteNode(ctx, node1.PubKeyBytes)
require.NoError(t, err)
// Now that the node has been deleted, we'll again query the nodes in
// the horizon. This time we should have no nodes at all.
nodesInHorizonIter = graph.NodeUpdatesInHorizon(
ctx, updateRange,
)
nodesInHorizon, err = fn.CollectErr(nodesInHorizonIter)
require.NoError(t, err, "unable to fetch nodes in horizon")
require.Empty(t, nodesInHorizon)
}
var (
updateTime = prand.Int63()
updateTimeMu sync.Mutex
updateBlock = prand.Uint32()
)
func nextUpdateTime() time.Time {
updateTimeMu.Lock()
defer updateTimeMu.Unlock()
updateTime++
return time.Unix(updateTime, 0)
}
func nextBlockHeight() uint32 {
updateTimeMu.Lock()
defer updateTimeMu.Unlock()
updateBlock++
return updateBlock
}
// testNodeIsPublic ensures that we properly detect nodes that are seen as
// public within the network graph.
func testNodeIsPublic(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
// We'll start off the test by creating a small network of 3
// participants with the following graph:
//
// Alice <-> Bob <-> Carol
//
// We'll need to create a separate database and channel graph for each
// participant to replicate real-world scenarios (private edges being in
// some graphs but not others, etc.).
aliceGraph := NewVersionedGraph(MakeTestGraph(t), v)
aliceNode := createTestVertex(t, v)
err := aliceGraph.SetSourceNode(ctx, aliceNode)
require.NoError(t, err, "unable to set source node")
bobGraph := NewVersionedGraph(MakeTestGraph(t), v)
bobNode := createTestVertex(t, v)
err = bobGraph.SetSourceNode(ctx, bobNode)
require.NoError(t, err, "unable to set source node")
carolGraph := NewVersionedGraph(MakeTestGraph(t), v)
carolNode := createTestVertex(t, v)
err = carolGraph.SetSourceNode(ctx, carolNode)
require.NoError(t, err, "unable to set source node")
aliceBobEdge, _ := createEdge(v, 10, 0, 0, 0, aliceNode, bobNode)
bobCarolEdge, _ := createEdge(v, 10, 1, 0, 1, bobNode, carolNode)
// After creating all of our nodes and edges, we'll add them to each
// participant's graph.
nodes := []*models.Node{aliceNode, bobNode, carolNode}
edges := []*models.ChannelEdgeInfo{aliceBobEdge, bobCarolEdge}
graphs := []*VersionedGraph{aliceGraph, bobGraph, carolGraph}
for _, graph := range graphs {
for _, node := range nodes {
node.LastUpdate = nextUpdateTime()
err := graph.AddNode(ctx, node)
require.NoError(t, err)
}
for _, edge := range edges {
err := graph.AddChannelEdge(ctx, edge)
require.NoError(t, err)
}
}
// checkNodes is a helper closure that will be used to assert that the
// given nodes are seen as public/private within the given graphs.
checkNodes := func(nodes []*models.Node,
graphs []*VersionedGraph, public bool) {
t.Helper()
for _, node := range nodes {
for _, graph := range graphs {
isPublic, err := graph.IsPublicNode(
ctx, node.PubKeyBytes,
)
require.NoError(t, err)
require.Equal(t, public, isPublic)
}
}
}
// Due to the way the edges were set up above, we'll make sure each node
// can correctly determine that every other node is public.
checkNodes(nodes, graphs, true)
// Now, we'll remove the edge between Alice and Bob from everyone's
// graph. This will make Alice be seen as a private node as it no longer
// has any advertised edges.
for _, graph := range graphs {
err := graph.DeleteChannelEdges(
ctx, false, true, aliceBobEdge.ChannelID,
)
require.NoError(t, err, "unable to remove edge")
}
checkNodes(
[]*models.Node{aliceNode},
[]*VersionedGraph{bobGraph, carolGraph},
false,
)
// We'll also make the edge between Bob and Carol private. Within Bob's
// and Carol's graph, the edge will exist, but it will not have a proof
// that allows it to be advertised. Within Alice's graph, we'll
// completely remove the edge as it is not possible for her to know of
// it without it being advertised.
for _, graph := range graphs {
err := graph.DeleteChannelEdges(
ctx, false, true, bobCarolEdge.ChannelID,
)
require.NoError(t, err, "unable to remove edge")
if graph == aliceGraph {
continue
}
bobCarolEdge.AuthProof = nil
err = graph.AddChannelEdge(ctx, bobCarolEdge)
require.NoError(t, err, "unable to add edge")
}
// With the modifications above, Bob should now be seen as a private
// node from both Alice's and Carol's perspective.
checkNodes(
[]*models.Node{bobNode},
[]*VersionedGraph{aliceGraph, carolGraph},
false,
)
}
// testIsPublicNodeEmptyChannelSignature ensures empty channel signatures don't
// mark nodes as public.
func testIsPublicNodeEmptyChannelSignature(t *testing.T,
v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
testGraph := MakeTestGraph(t)
graph := NewVersionedGraph(testGraph, v)
// Set a source node as it's required for IsPublicNode.
sourceNode := createTestVertex(t, v)
err := graph.SetSourceNode(ctx, sourceNode)
require.NoError(t, err)
node1 := createTestVertex(t, v)
node1.LastUpdate = nextUpdateTime()
err = graph.AddNode(ctx, node1)
require.NoError(t, err)
// Create an edge between source node and node1, with
// empty signatures. This tests that empty signatures
// don't mark nodes as public.
edgeInfo, _ := createEdge(
v, 10, 0, 0, 0, sourceNode, node1,
true,
)
switch v {
case lnwire.GossipVersion1:
edgeInfo.AuthProof =
models.NewV1ChannelAuthProof(
[]byte{}, []byte{},
[]byte{}, []byte{},
)
case lnwire.GossipVersion2:
edgeInfo.AuthProof =
models.NewV2ChannelAuthProof([]byte{})
}
err = graph.AddChannelEdge(ctx, edgeInfo)
require.NoError(t, err)
// node1 should NOT be considered public because the
// channel announcement has empty signatures.
isPublic, err := graph.IsPublicNode(ctx, node1.PubKeyBytes)
require.NoError(t, err)
require.False(t, isPublic)
}
// BenchmarkIsPublicNode measures the performance of IsPublicNode when checking
// a large number of nodes.
func BenchmarkIsPublicNode(b *testing.B) {
graph := MakeTestGraph(b)
// Create a graph with a reasonable number of nodes and channels.
numNodes := 100
numChans := 4
_, nodes := fillTestGraph(
b, graph, numNodes, numChans, lnwire.GossipVersion1,
)
// Use deterministic random number generator for reproducible results.
rng := prand.New(prand.NewSource(42))
v1Graph := NewVersionedGraph(graph, lnwire.GossipVersion1)
for b.Loop() {
// Query random nodes to avoid query caching and better
// represent real-world query patterns.
nodePub := nodes[rng.Intn(len(nodes))].PubKeyBytes
_, err := v1Graph.IsPublicNode(b.Context(), nodePub)
require.NoError(b, err)
}
}
// 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, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// Create first node and add it to the graph.
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
// Create second node and add it to the graph.
node2 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node2))
// Adding a new channel edge to the graph.
edgeInfo, edge1, edge2 := createChannelEdge(node1, node2, v)
switch v {
case lnwire.GossipVersion1:
node2.LastUpdate = nextUpdateTime()
case lnwire.GossipVersion2:
node2.LastBlockHeight = nextBlockHeight()
}
require.NoError(t, graph.AddNode(ctx, node2))
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
// Ensure no disabled channels exist in the bucket on start.
disabledChanIds, err := graph.DisabledChannelIDs(ctx)
require.NoError(t, err, "unable to get disabled channel ids")
require.Empty(t, disabledChanIds)
// Add one disabled policy and ensure the channel is still not in the
// disabled list.
switch v {
case lnwire.GossipVersion1:
edge1.ChannelFlags |= lnwire.ChanUpdateDisabled
case lnwire.GossipVersion2:
edge1.DisableFlags |= lnwire.ChanUpdateDisableIncoming
}
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
disabledChanIds, err = graph.DisabledChannelIDs(ctx)
require.NoError(t, err, "unable to get disabled channel ids")
require.Empty(t, disabledChanIds)
// Add second disabled policy and ensure the channel is now in the
// disabled list.
switch v {
case lnwire.GossipVersion1:
edge2.ChannelFlags |= lnwire.ChanUpdateDisabled
case lnwire.GossipVersion2:
edge2.DisableFlags |= lnwire.ChanUpdateDisableIncoming
}
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
disabledChanIds, err = graph.DisabledChannelIDs(ctx)
require.NoError(t, err, "unable to get disabled channel ids")
require.Equal(t, []uint64{edgeInfo.ChannelID}, disabledChanIds)
// Delete the channel edge and ensure it is removed from the disabled
// list.
require.NoError(t, graph.DeleteChannelEdges(
ctx, false, true, edgeInfo.ChannelID,
))
disabledChanIds, err = graph.DisabledChannelIDs(ctx)
require.NoError(t, err, "unable to get disabled channel ids")
require.Empty(t, disabledChanIds)
}
// TestEdgePolicyMissingMaxHTLC tests that if we find a ChannelEdgePolicy in
// the DB that indicates that it should support the htlc_maximum_value_msat
// field, but it is not part of the opaque data, then we'll handle it as it is
// unknown. It also checks that we are correctly able to overwrite it when we
// receive the proper update.
func TestEdgePolicyMissingMaxHTLC(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
// This test currently directly edits the bytes stored in the bbolt DB.
boltStore, ok := graph.db.(*KVStore)
if !ok {
t.Skipf("skipping test that is aimed at a bbolt graph DB")
}
// 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)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, graph.AddNode(ctx, node2))
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
chanID := edgeInfo.ChannelID
from := edge2.ToNode[:]
to := edge1.ToNode[:]
// We'll remove the no max_htlc field from the first edge policy, and
// all other opaque data, and serialize it.
edge1.MessageFlags = 0
edge1.ExtraOpaqueData = nil
var b bytes.Buffer
require.NoError(t, serializeChanEdgePolicy(&b, edge1, to))
// Set the max_htlc field. The extra bytes added to the serialization
// will be the opaque data containing the serialized field.
edge1.MessageFlags = lnwire.ChanUpdateRequiredMaxHtlc
edge1.MaxHTLC = 13928598
var b2 bytes.Buffer
require.NoError(t, serializeChanEdgePolicy(&b2, edge1, to))
withMaxHtlc := b2.Bytes()
// Remove the opaque data from the serialization.
stripped := withMaxHtlc[:len(b.Bytes())]
// Attempting to deserialize these bytes should return an error.
r := bytes.NewReader(stripped)
_, err := deserializeChanEdgePolicy(r)
require.ErrorIs(t, err, ErrEdgePolicyOptionalFieldNotFound)
// Put the stripped bytes in the DB.
putSerializedPolicy(t, boltStore.db, from, chanID, stripped)
// And add the second, unmodified edge.
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
// Attempt to fetch the edge and policies from the DB. Since the policy
// we added is invalid according to the new format, it should be as we
// are not aware of the policy (indicated by the policy returned being
// nil)
dbEdgeInfo, dbEdge1, dbEdge2, err := graph.FetchChannelEdgesByID(
ctx, chanID,
)
require.NoError(t, err, "unable to fetch channel by ID")
// The first edge should have a nil-policy returned
require.Nil(t, dbEdge1)
compareEdgePolicies(t, dbEdge2, edge2)
assertEdgeInfoEqual(t, dbEdgeInfo, edgeInfo)
// Now add the original, unmodified edge policy, and make sure the edge
// policies then become fully populated.
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
dbEdgeInfo, dbEdge1, dbEdge2, err = graph.FetchChannelEdgesByID(
ctx, chanID,
)
require.NoError(t, err, "unable to fetch channel by ID")
compareEdgePolicies(t, dbEdge1, edge1)
compareEdgePolicies(t, dbEdge2, edge2)
assertEdgeInfoEqual(t, dbEdgeInfo, edgeInfo)
}
// putSerializedPolicy is a helper function that writes a serialized
// ChannelEdgePolicy to the edge bucket in the database.
func putSerializedPolicy(t *testing.T, db kvdb.Backend, from []byte,
chanID uint64, b []byte) {
err := kvdb.Update(db, func(tx kvdb.RwTx) error {
edges := tx.ReadWriteBucket(edgeBucket)
require.NotNil(t, edges)
edgeIndex := edges.NestedReadWriteBucket(edgeIndexBucket)
require.NotNil(t, edgeIndex)
var edgeKey [33 + 8]byte
copy(edgeKey[:], from)
byteOrder.PutUint64(edgeKey[33:], chanID)
var scratch [8]byte
var indexKey [8 + 8]byte
copy(indexKey[:], scratch[:])
byteOrder.PutUint64(indexKey[8:], chanID)
updateIndex, err := edges.CreateBucketIfNotExists(
edgeUpdateIndexBucket,
)
require.NoError(t, err)
require.NoError(t, updateIndex.Put(indexKey[:], nil))
return edges.Put(edgeKey[:], b)
}, func() {})
require.NoError(t, err, "error writing db")
}
// assertNumZombies queries the provided ChannelGraph for NumZombies for the
// given gossip version and asserts that the result equals the expected count.
func assertNumZombies(t *testing.T, graph *ChannelGraph,
v lnwire.GossipVersion, expZombies uint64) {
t.Helper()
vGraph := NewVersionedGraph(graph, v)
numZombies, err := vGraph.NumZombies(t.Context())
require.NoError(t, err, "unable to query number of zombies")
require.Equal(t, expZombies, numZombies)
}
// testGraphZombieIndex ensures that we can mark edges correctly as zombie/live.
func testGraphZombieIndex(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
// We'll start by creating our test graph along with a test edge.
graph := MakeTestGraph(t)
node1 := createTestVertex(t, v)
node2 := createTestVertex(t, v)
// Swap the nodes if the second's pubkey is smaller than the first.
// Without this, the comparisons at the end will fail probabilistically.
if bytes.Compare(node2.PubKeyBytes[:], node1.PubKeyBytes[:]) < 0 {
node1, node2 = node2, node1
}
edge, _, _ := createChannelEdge(node1, node2, v)
require.NoError(t, graph.AddChannelEdge(ctx, edge))
vGraph := NewVersionedGraph(graph, v)
// Since the edge is known the graph and it isn't a zombie, IsZombieEdge
// should not report the channel as a zombie.
isZombie, _, _, err := vGraph.IsZombieEdge(ctx, edge.ChannelID)
require.NoError(t, err)
require.False(t, isZombie)
assertNumZombies(t, graph, v, 0)
// If we delete the edge and mark it as a zombie, then we should expect
// to see it within the index.
err = graph.DeleteChannelEdges(ctx, v, false, true, edge.ChannelID)
require.NoError(t, err, "unable to mark edge as zombie")
isZombie, pubKey1, pubKey2, err := vGraph.IsZombieEdge(
ctx, edge.ChannelID,
)
require.NoError(t, err)
require.True(t, isZombie)
require.Equal(t, node1.PubKeyBytes, pubKey1)
require.Equal(t, node2.PubKeyBytes, pubKey2)
assertNumZombies(t, graph, v, 1)
// Similarly, if we mark the same edge as live, we should no longer see
// it within the index.
err = graph.MarkEdgeLive(ctx, v, edge.ChannelID)
require.NoError(t, err)
// Attempting to mark the edge as live again now that it is no longer
// in the zombie index should fail.
require.ErrorIs(
t, graph.MarkEdgeLive(ctx, v, edge.ChannelID),
ErrZombieEdgeNotFound,
)
isZombie, _, _, err = vGraph.IsZombieEdge(ctx, edge.ChannelID)
require.NoError(t, err)
require.False(t, isZombie)
assertNumZombies(t, graph, v, 0)
// If we mark the edge as a zombie manually, then it should show up as
// being a zombie once again.
err = graph.MarkEdgeZombie(
ctx, v, edge.ChannelID,
node1.PubKeyBytes, node2.PubKeyBytes,
)
require.NoError(t, err, "unable to mark edge as zombie")
isZombie, _, _, err = vGraph.IsZombieEdge(ctx, edge.ChannelID)
require.NoError(t, err)
require.True(t, isZombie)
assertNumZombies(t, graph, v, 1)
}
// testFetchZombieEdgeVersioning verifies that when a zombie edge is fetched via
// FetchChannelEdgesByID, the returned ChannelEdgeInfo carries the correct
// gossip version.
func testFetchZombieEdgeVersioning(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
node1 := createTestVertex(t, v)
node2 := createTestVertex(t, v)
if bytes.Compare(node2.PubKeyBytes[:], node1.PubKeyBytes[:]) < 0 {
node1, node2 = node2, node1
}
edge, _, _ := createChannelEdge(node1, node2, v)
require.NoError(t, graph.AddChannelEdge(ctx, edge))
// Delete the edge and mark it as a zombie.
err := graph.DeleteChannelEdges(ctx, false, true, edge.ChannelID)
require.NoError(t, err)
// Fetch the zombie edge by ID. The returned edge info should carry
// the correct gossip version even though the channel data has been
// removed.
info, _, _, err := graph.FetchChannelEdgesByID(ctx, edge.ChannelID)
require.ErrorIs(t, err, ErrZombieEdge)
require.NotNil(t, info)
require.Equal(t, v, info.Version)
require.Equal(t, edge.NodeKey1Bytes, info.NodeKey1Bytes)
require.Equal(t, edge.NodeKey2Bytes, info.NodeKey2Bytes)
}
// compareNodes is used to compare two Nodes.
func compareNodes(t *testing.T, a, b *models.Node) {
t.Helper()
// Call the PubKey method for each node to ensure that the internal
// `pubKey` field is set for both objects and so require.Equals can
// then be used to compare the structs.
_, err := a.PubKey()
require.NoError(t, err)
_, err = b.PubKey()
require.NoError(t, err)
require.Equal(t, a, b)
}
// compareEdgePolicies compares two ChannelEdgePolicy values for semantic
// equality after normalizing version-specific/backend-specific differences.
func compareEdgePolicies(t testing.TB, a, b *models.ChannelEdgePolicy) {
t.Helper()
//nolint:ll
normalize := func(p *models.ChannelEdgePolicy) *models.ChannelEdgePolicy {
if p == nil {
return nil
}
policy := copyEdgePolicy(p)
if len(policy.ExtraOpaqueData) == 0 {
policy.ExtraOpaqueData = nil
}
if len(policy.ExtraSignedFields) == 0 {
policy.ExtraSignedFields = nil
}
switch policy.Version {
case lnwire.GossipVersion1:
// SecondPeer is v2-specific; derive canonical direction
// for v1.
policy.SecondPeer = !policy.IsNode1()
policy.LastBlockHeight = 0
policy.DisableFlags = 0
policy.ExtraSignedFields = nil
case lnwire.GossipVersion2:
policy.LastUpdate = time.Time{}
policy.MessageFlags = 0
policy.ChannelFlags = 0
policy.ExtraOpaqueData = nil
}
return policy
}
normalizedA := normalize(a)
normalizedB := normalize(b)
require.Equal(t, normalizedA, normalizedB)
}
// testLightningNodeSigVerification checks that we can use the Node's pubkey to
// verify signatures. For v1 this exercises ECDSA, for v2 Schnorr.
func testLightningNodeSigVerification(t *testing.T,
v lnwire.GossipVersion) {
t.Parallel()
// Create some dummy data to sign.
var data [32]byte
_, err := prand.Read(data[:])
require.NoError(t, err)
// Create private key.
priv, err := btcec.NewPrivateKey()
require.NoError(t, err, "unable to create priv key")
// Create a Node from the same private key.
node := createNode(t, v, priv)
// Retrieve the public key from the node and verify a signature
// produced by the same private key.
nodePub, err := node.PubKey()
require.NoError(t, err, "unable to get pubkey")
// Sign the data using the appropriate scheme for the gossip version.
// V1 uses ECDSA, v2 uses Schnorr.
type verifiable interface {
Verify(hash []byte, pubKey *btcec.PublicKey) bool
}
var sig verifiable
switch v {
case lnwire.GossipVersion1:
sig = ecdsa.Sign(priv, data[:])
case lnwire.GossipVersion2:
schnorrSig, sErr := schnorr.Sign(priv, data[:])
require.NoError(t, sErr)
sig = schnorrSig
}
// Verify against the raw private key's pubkey, then against the
// pubkey extracted from the Node.
require.True(t, sig.Verify(data[:], priv.PubKey()))
require.True(t, sig.Verify(data[:], nodePub))
}
// TestComputeFee tests fee calculation based on the outgoing amt.
func TestComputeFee(t *testing.T) {
var (
policy = models.ChannelEdgePolicy{
Version: lnwire.GossipVersion1,
FeeBaseMSat: 10000,
FeeProportionalMillionths: 30000,
}
outgoingAmt = lnwire.MilliSatoshi(1000000)
expectedFee = lnwire.MilliSatoshi(40000)
)
fee := policy.ComputeFee(outgoingAmt)
require.Equal(t, expectedFee, fee)
}
// TestBatchedAddChannelEdge asserts that BatchedAddChannelEdge properly
// executes multiple AddChannelEdge requests in a single txn.
func testBatchedAddChannelEdge(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
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, v)
node2 := createTestVertex(t, v)
// In addition to the fake vertexes we create some fake channel
// identifiers.
var spendOutputs []*wire.OutPoint
var blockHash chainhash.Hash
copy(blockHash[:], bytes.Repeat([]byte{1}, 32))
// Prune the graph a few times to make sure we have entries in the
// prune log.
_, err := graph.PruneGraph(ctx, spendOutputs, &blockHash, 155)
require.Nil(t, err)
var blockHash2 chainhash.Hash
copy(blockHash2[:], bytes.Repeat([]byte{2}, 32))
_, err = graph.PruneGraph(ctx, spendOutputs, &blockHash2, 156)
require.Nil(t, err)
// We'll create 3 almost identical edges, so first create a helper
// method containing all logic for doing so.
// Create an edge which has its block height at 156.
height := uint32(156)
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(
v, height+1, math.MaxUint32&0x00ffffff, math.MaxUint16, 1,
node1, node2,
)
// Create a third edge, this with a block height of 155.
edgeInfo3, _ := createEdge(
v, height-1, 0, 0, 2, node1, node2,
)
edges := []models.ChannelEdgeInfo{*edgeInfo, *edgeInfo2, *edgeInfo3}
errChan := make(chan error, len(edges))
errTimeout := errors.New("timeout adding batched channel")
// Now add all these new edges to the database.
var wg sync.WaitGroup
for _, edge := range edges {
wg.Add(1)
go func(edge models.ChannelEdgeInfo) {
defer wg.Done()
select {
case errChan <- graph.AddChannelEdge(ctx, &edge):
case <-time.After(2 * time.Second):
errChan <- errTimeout
}
}(edge)
}
wg.Wait()
for i := 0; i < len(edges); i++ {
err := <-errChan
require.Nil(t, err)
}
}
// TestBatchedUpdateEdgePolicy asserts that BatchedUpdateEdgePolicy properly
// executes multiple UpdateEdgePolicy requests in a single txn.
func testBatchedUpdateEdgePolicy(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
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, v)
require.NoError(t, graph.AddNode(ctx, node1))
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, v)
// Make sure inserting the policy at this point, before the edge info
// is added, will fail.
require.ErrorIs(t, graph.UpdateEdgePolicy(ctx, edge1), ErrEdgeNotFound)
// Add the edge info.
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
errTimeout := errors.New("timeout adding batched channel")
updates := []*models.ChannelEdgePolicy{edge1, edge2}
errChan := make(chan error, len(updates))
// Now add all these new edges to the database.
var wg sync.WaitGroup
for _, update := range updates {
wg.Add(1)
go func(update *models.ChannelEdgePolicy) {
defer wg.Done()
select {
case errChan <- graph.UpdateEdgePolicy(ctx, update):
case <-time.After(2 * time.Second):
errChan <- errTimeout
}
}(update)
}
wg.Wait()
for i := 0; i < len(updates); i++ {
err := <-errChan
require.Nil(t, err)
}
}
// BenchmarkForEachChannel is a benchmark test that measures the number of
// allocations and the total memory consumed by the full graph traversal.
func BenchmarkForEachChannel(b *testing.B) {
graph := MakeTestGraph(b)
ctx := b.Context()
const numNodes = 100
const numChannels = 4
_, _ = fillTestGraph(
b, graph, numNodes, numChannels, lnwire.GossipVersion1,
)
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
var (
totalCapacity btcutil.Amount
maxHTLCs lnwire.MilliSatoshi
)
var nodes []route.Vertex
err := graph.ForEachNodeCacheable(
ctx, lnwire.GossipVersion1, func(node route.Vertex,
vector *lnwire.FeatureVector) error {
nodes = append(nodes, node)
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 {
// We need to do something with
// the data here, otherwise the
// compiler is going to optimize
// this away, and we get bogus
// results.
totalCapacity += info.Capacity
maxHTLCs += policy.MaxHTLC
maxHTLCs += policy2.MaxHTLC
return nil
}
err := graph.ForEachNodeChannel(
ctx, lnwire.GossipVersion1, n, cb, func() {},
)
require.NoError(b, err)
}
}
}
// TestForEachNodeDirectedChannel tests that the ForEachNodeDirectedChannel
// method works as expected, and is able to handle nil self edges.
func testGraphCacheForEachNodeChannel(t *testing.T,
v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
// Unset the channel graph cache to simulate the user running with the
// option turned off. This forces the V1Store ForEachNodeDirectedChannel
// to be queried instead of the graph cache's ForEachChannel method.
graph := NewVersionedGraph(
MakeTestGraph(t, WithUseGraphCache(false)), v,
)
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
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, 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
// runtime.
var edge1 *models.ChannelEdgePolicy
if e1.ToNode == node2.PubKeyBytes {
edge1 = e1
} else {
edge1 = e2
}
// Add the channel, but only insert a single edge into the graph.
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
getSingleChannel := func() *DirectedChannel {
var ch *DirectedChannel
err := graph.db.ForEachNodeDirectedChannel(
ctx, v, node1.PubKeyBytes,
func(c *DirectedChannel) error {
require.Nil(t, ch)
ch = c
return nil
}, func() {},
)
require.NoError(t, err)
return ch
}
// We should be able to accumulate the single channel added, even
// though we have a nil edge policy here.
require.NotNil(t, getSingleChannel())
// Set an inbound fee and check that it is properly returned.
edge1.ExtraOpaqueData = []byte{
253, 217, 3, 8, 0, 0, 0, 10, 0, 0, 0, 20,
}
inboundFee := lnwire.Fee{
BaseFee: 10,
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.
directedChan := getSingleChannel()
require.NotNil(t, directedChan)
require.Equal(t, inboundFee, directedChan.InboundFee)
// 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)
}
}
// TestGraphLoading asserts that the cache is properly reconstructed after a
// restart.
func TestGraphLoading(t *testing.T) {
t.Parallel()
// Next, create the graph for the first time.
graphStore := NewTestDB(t)
graph, err := NewChannelGraph(
graphStore, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
// Populate the graph with test data.
const numNodes = 100
const numChannels = 4
_, _ = fillTestGraph(
t, graph, numNodes, numChannels, lnwire.GossipVersion1,
)
// Recreate the graph. This should cause the graph cache to be
// populated.
graphReloaded, err := NewChannelGraph(
graphStore, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, graphReloaded.Start())
t.Cleanup(func() {
require.NoError(t, graphReloaded.Stop())
})
// Assert that the cache content is identical.
require.Equal(
t, graph.cache.graphCache.nodeChannels,
graphReloaded.cache.graphCache.nodeChannels,
)
require.Equal(
t, graph.cache.graphCache.nodeFeatures,
graphReloaded.cache.graphCache.nodeFeatures,
)
}
// TestAsyncGraphCache tests the behaviour of the ChannelGraph when the graph
// cache is populated asynchronously.
func TestAsyncGraphCache(t *testing.T) {
t.Parallel()
ctx := t.Context()
const (
numNodes = 100
numChannels = 3
)
// Next, create the graph for the first time.
graphStore := NewTestDB(t)
// The first time we spin up the graph, we Start is as normal and fill
// it with test data. This will ensure that the graph cache has
// something to load on the next Start.
graph, err := NewChannelGraph(graphStore)
require.NoError(t, err)
require.NoError(t, graph.Start())
channels, nodes := fillTestGraph(
t, graph, numNodes, numChannels, lnwire.GossipVersion1,
)
assertGraphState := func() {
var (
numNodes int
chanIndex = make(map[uint64]struct{}, numChannels)
)
// We query the graph for all nodes and channels, and
// assert that we get the expected number of nodes and
// channels.
err := graph.ForEachNodeCached(
ctx, lnwire.GossipVersion1,
func(_ context.Context, node route.Vertex,
chans map[uint64]*DirectedChannel) error {
numNodes++
for chanID := range chans {
chanIndex[chanID] = struct{}{}
}
return nil
}, func() {
numNodes = 0
chanIndex = make(
map[uint64]struct{}, numChannels,
)
},
)
require.NoError(t, err)
require.Equal(t, len(nodes), numNodes)
require.Equal(t, len(channels), len(chanIndex))
}
assertGraphState()
// Now we stop the graph.
require.NoError(t, graph.Stop())
// Recreate it but don't start it yet.
graph, err = NewChannelGraph(graphStore)
require.NoError(t, err)
// Spin off a goroutine that starts to make queries to the ChannelGraph.
// We start this before we start the graph, so that we can ensure that
// the queries are made while the graph cache is being populated.
var (
wg sync.WaitGroup
numRuns = 10
)
for i := 0; i < numRuns; i++ {
wg.Add(1)
go func() {
defer wg.Done()
assertGraphState()
}()
}
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
wg.Wait()
// Wait for the cache to be fully populated.
err = wait.Predicate(func() bool {
return graph.cache.isLoaded()
}, wait.DefaultTimeout)
require.NoError(t, err)
// And then assert that all the expected nodes and channels are
// present in the graph cache.
for _, node := range nodes {
_, ok := graph.cache.graphCache.nodeChannels[node.PubKeyBytes]
require.True(t, ok)
}
}
type blockingCacheLoadStore struct {
Store
cacheLoadStarted chan struct{}
allowCacheLoad chan struct{}
blockOnce sync.Once
}
// ForEachChannelCacheable pauses the first cacheable channel iteration until
// the test allows it to continue.
func (s *blockingCacheLoadStore) ForEachChannelCacheable(ctx context.Context,
v lnwire.GossipVersion, cb func(*models.CachedEdgeInfo,
*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
reset func()) error {
return s.Store.ForEachChannelCacheable(
ctx, v, func(info *models.CachedEdgeInfo,
policy1,
policy2 *models.CachedEdgePolicy) error {
s.blockOnce.Do(func() {
close(s.cacheLoadStarted)
<-s.allowCacheLoad
})
return cb(info, policy1, policy2)
}, reset,
)
}
type shutdownBlockingCacheLoadStore struct {
Store
cacheLoadStarted chan struct{}
blockOnce sync.Once
}
// ForEachChannelCacheable blocks until the context is canceled so tests can
// assert that Stop interrupts async cache population.
func (s *shutdownBlockingCacheLoadStore) ForEachChannelCacheable(
ctx context.Context, v lnwire.GossipVersion,
cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
*models.CachedEdgePolicy) error, reset func()) error {
return s.Store.ForEachChannelCacheable(
ctx, v, func(info *models.CachedEdgeInfo,
policy1,
policy2 *models.CachedEdgePolicy) error {
s.blockOnce.Do(func() {
close(s.cacheLoadStarted)
<-ctx.Done()
})
return ctx.Err()
}, reset,
)
}
type failingCacheLoadStore struct {
Store
cacheLoadAttempted chan struct{}
populateErr error
}
// ForEachChannelCacheable fails the initial cache population after signaling
// that the async load reached channel iteration.
func (s *failingCacheLoadStore) ForEachChannelCacheable(ctx context.Context,
v lnwire.GossipVersion, cb func(*models.CachedEdgeInfo,
*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
reset func()) error {
close(s.cacheLoadAttempted)
return s.populateErr
}
// TestAsyncGraphCacheReplaysConcurrentWrites asserts that graph mutations that
// happen while the async cache population is running are replayed onto the
// cache before it becomes readable.
func TestAsyncGraphCacheReplaysConcurrentWrites(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
setupGraph, err := NewChannelGraph(
store, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, setupGraph.Start())
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, setupGraph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge2))
require.NoError(t, setupGraph.Stop())
blockingStore := &blockingCacheLoadStore{
Store: store,
cacheLoadStarted: make(chan struct{}),
allowCacheLoad: make(chan struct{}),
}
graph, err := NewChannelGraph(blockingStore)
require.NoError(t, err)
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
<-blockingStore.cacheLoadStarted
updatedEdge := *edge1
updatedEdge.LastUpdate = nextUpdateTime()
updatedEdge.FeeBaseMSat++
require.NoError(t, graph.UpdateEdgePolicy(ctx, &updatedEdge))
close(blockingStore.allowCacheLoad)
err = wait.Predicate(func() bool {
return graph.cache.isLoaded()
}, wait.DefaultTimeout)
require.NoError(t, err)
var cachedFee lnwire.MilliSatoshi
err = graph.ForEachNodeDirectedChannel(
ctx, updatedEdge.ToNode,
func(channel *DirectedChannel) error {
if channel.ChannelID != updatedEdge.ChannelID {
return nil
}
require.NotNil(t, channel.InPolicy)
cachedFee = channel.InPolicy.FeeBaseMSat
return nil
}, func() {},
)
require.NoError(t, err)
require.Equal(t, updatedEdge.FeeBaseMSat, cachedFee)
}
// TestAsyncGraphCacheStopCancelsLoad asserts that Stop interrupts async cache
// population instead of waiting for the full load to finish.
func TestAsyncGraphCacheStopCancelsLoad(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
setupGraph, err := NewChannelGraph(
store, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, setupGraph.Start())
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, setupGraph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge2))
require.NoError(t, setupGraph.Stop())
blockingStore := &shutdownBlockingCacheLoadStore{
Store: store,
cacheLoadStarted: make(chan struct{}),
}
graph, err := NewChannelGraph(blockingStore)
require.NoError(t, err)
require.NoError(t, graph.Start())
<-blockingStore.cacheLoadStarted
stopErr := make(chan error, 1)
go func() {
stopErr <- graph.Stop()
}()
select {
case err := <-stopErr:
require.NoError(t, err)
case <-time.After(wait.DefaultTimeout):
t.Fatal("Stop did not cancel graph cache loading")
}
}
// TestAsyncGraphCachePopulationFailureFallsBackToDB asserts that cache
// population errors leave the cache unreadable while reads continue to succeed
// through the DB-backed path.
func TestAsyncGraphCachePopulationFailureFallsBackToDB(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
setupGraph, err := NewChannelGraph(
store, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, setupGraph.Start())
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, setupGraph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge2))
require.NoError(t, setupGraph.Stop())
populateErr := errors.New("cache population failed")
failingStore := &failingCacheLoadStore{
Store: store,
cacheLoadAttempted: make(chan struct{}),
populateErr: populateErr,
}
graph, err := NewChannelGraph(failingStore)
require.NoError(t, err)
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
<-failingStore.cacheLoadAttempted
err = wait.Predicate(func() bool {
return graph.GraphCacheStatus() == GraphCacheStatusFailed
}, wait.DefaultTimeout)
require.NoError(t, err)
require.False(t, graph.cache.isLoaded())
var numChannels int
err = graph.ForEachNodeDirectedChannel(
ctx, edge1.ToNode,
func(channel *DirectedChannel) error {
if channel.ChannelID != edge1.ChannelID {
return nil
}
numChannels++
require.NotNil(t, channel.InPolicy)
require.Equal(t, edge1.FeeBaseMSat,
channel.InPolicy.FeeBaseMSat)
return nil
}, func() {},
)
require.NoError(t, err)
require.Equal(t, 1, numChannels)
}
// TestGraphCacheStatus asserts that the graph cache reports disabled, loading,
// loaded and failed states as expected.
func TestGraphCacheStatus(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
disabledGraph, err := NewChannelGraph(
store, WithUseGraphCache(false),
)
require.NoError(t, err)
require.Equal(
t, GraphCacheStatusDisabled, disabledGraph.GraphCacheStatus(),
)
require.NoError(t, disabledGraph.Start())
require.Equal(
t, GraphCacheStatusDisabled, disabledGraph.GraphCacheStatus(),
)
require.NoError(t, disabledGraph.Stop())
setupGraph, err := NewChannelGraph(
store, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, setupGraph.Start())
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, setupGraph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge2))
require.NoError(t, setupGraph.Stop())
blockingStore := &blockingCacheLoadStore{
Store: store,
cacheLoadStarted: make(chan struct{}),
allowCacheLoad: make(chan struct{}),
}
graph, err := NewChannelGraph(blockingStore)
require.NoError(t, err)
require.Equal(t, GraphCacheStatusLoading, graph.GraphCacheStatus())
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
<-blockingStore.cacheLoadStarted
require.Equal(t, GraphCacheStatusLoading, graph.GraphCacheStatus())
close(blockingStore.allowCacheLoad)
err = wait.Predicate(func() bool {
return graph.GraphCacheStatus() == GraphCacheStatusLoaded
}, wait.DefaultTimeout)
require.NoError(t, err)
require.NoError(t, graph.Stop())
// Assert the failed state by using a store that errors during cache
// population.
populateErr := errors.New("cache population failed")
failingStore := &failingCacheLoadStore{
Store: store,
cacheLoadAttempted: make(chan struct{}),
populateErr: populateErr,
}
failedGraph, err := NewChannelGraph(failingStore)
require.NoError(t, err)
require.NoError(t, failedGraph.Start())
t.Cleanup(func() {
require.NoError(t, failedGraph.Stop())
})
<-failingStore.cacheLoadAttempted
err = wait.Predicate(func() bool {
return failedGraph.GraphCacheStatus() == GraphCacheStatusFailed
}, wait.DefaultTimeout)
require.NoError(t, err)
}
// TestKVCacheableIteratorsRespectCancellation asserts that KV-backed cache
// iterators return when their context is canceled.
func TestKVCacheableIteratorsRespectCancellation(t *testing.T) {
t.Parallel()
if isSQLDB {
t.Skip("KV iterator cancellation is specific to KVStore")
}
ctx := t.Context()
store := NewTestDB(t)
kvStore, ok := store.(*KVStore)
require.True(t, ok)
graph, err := NewChannelGraph(
kvStore, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
canceledCtx, cancel := context.WithCancel(ctx)
cancel()
err = kvStore.ForEachNodeCacheable(
canceledCtx, lnwire.GossipVersion1,
func(route.Vertex, *lnwire.FeatureVector) error {
return nil
}, func() {},
)
require.ErrorIs(t, err, context.Canceled)
err = kvStore.ForEachChannelCacheable(
canceledCtx, lnwire.GossipVersion1,
func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
*models.CachedEdgePolicy) error {
return nil
}, func() {},
)
require.ErrorIs(t, err, context.Canceled)
}
// TestClosedScid tests that we can correctly insert a SCID into the index of
// closed short channel ids.
func TestClosedScid(t *testing.T) {
t.Parallel()
graph := MakeTestGraph(t)
scid := lnwire.ShortChannelID{}
// The scid should not exist in the closedScidBucket.
exists, err := graph.IsClosedScid(t.Context(), scid)
require.Nil(t, err)
require.False(t, exists)
// After we call PutClosedScid, the call to IsClosedScid should return
// true.
err = graph.PutClosedScid(t.Context(), scid)
require.Nil(t, err)
exists, err = graph.IsClosedScid(t.Context(), scid)
require.Nil(t, err)
require.True(t, exists)
}
// testNodeAnn is a serialized node announcement message which contains an
// address type (6) that LND is not aware of.
var testNodeAnn = "01012674c2e7ef68c73a086b7de2603f4ef1567358df84bb4edaa06c" +
"f2132965b14e2434faab04170f0089216accbd79188fa3d40dbb0438bd89782cae" +
"27cc656bf60007800088082a69a2625e7a2a024b9a1fa8e006f1e3937f65f66c40" +
"8e6da8e1ca728ea43222a7381df1cc449605024b9a424c554549524f4e2d76302e" +
"31312e307263332d362d67663963613934650000001d0180c7caa8260702240061" +
"80000000d0000000005cd2a001260706204c"
// TestLightningNodePersistence takes a raw serialized node announcement
// message, converts it to our internal models.Node type, persists it
// to disk, reads it again and converts it back to a wire message and asserts
// that the two messages are equal.
func TestLightningNodePersistence(t *testing.T) {
t.Parallel()
ctx := t.Context()
// Create a new test graph instance.
graph := NewVersionedGraph(MakeTestGraph(t), lnwire.GossipVersion1)
nodeAnnBytes, err := hex.DecodeString(testNodeAnn)
require.NoError(t, err)
// Use the raw serialized node announcement message create an
// lnwire.NodeAnnouncement1 instance.
msg, err := lnwire.ReadMessage(bytes.NewBuffer(nodeAnnBytes), 0)
require.NoError(t, err)
na, ok := msg.(*lnwire.NodeAnnouncement1)
require.True(t, ok)
// Convert the wire message to our internal node representation.
node := models.NodeFromWireAnnouncement(na)
// Persist the node to disk.
err = graph.AddNode(ctx, node)
require.NoError(t, err)
// Read the node from disk.
diskNode, err := graph.FetchNode(ctx, node.PubKeyBytes)
require.NoError(t, err)
// Convert it back to a wire message.
wireMsg, err := diskNode.NodeAnnouncement(true)
require.NoError(t, err)
// Encode it and compare against the original.
var b bytes.Buffer
_, err = lnwire.WriteMessage(&b, wireMsg, 0)
require.NoError(t, err)
require.Equal(t, nodeAnnBytes, b.Bytes())
}
// TestUpdateRangeValidateForVersion verifies that ChanUpdateRange and
// NodeUpdateRange reject invalid field combinations for each gossip version.
func TestUpdateRangeValidateForVersion(t *testing.T) {
t.Parallel()
now := time.Now()
tests := []struct {
name string
fn func() error
wantErr string
}{
{
name: "v1 chan range with time - ok",
fn: func() error {
r := ChanUpdateRange{
StartTime: fn.Some(now),
EndTime: fn.Some(now),
}
return r.validateForVersion(
lnwire.GossipVersion1,
)
},
},
{
name: "v1 chan range with height - rejected",
fn: func() error {
r := ChanUpdateRange{
StartHeight: fn.Some(uint32(1)),
EndHeight: fn.Some(uint32(100)),
}
return r.validateForVersion(
lnwire.GossipVersion1,
)
},
wantErr: "v1 chan update range must use time",
},
{
name: "v2 chan range with height - ok",
fn: func() error {
r := ChanUpdateRange{
StartHeight: fn.Some(uint32(1)),
EndHeight: fn.Some(uint32(100)),
}
return r.validateForVersion(
lnwire.GossipVersion2,
)
},
},
{
name: "v2 chan range with time - rejected",
fn: func() error {
r := ChanUpdateRange{
StartTime: fn.Some(now),
EndTime: fn.Some(now),
}
return r.validateForVersion(
lnwire.GossipVersion2,
)
},
wantErr: "v2 chan update range must use blocks",
},
{
name: "mixed chan range - rejected",
fn: func() error {
r := ChanUpdateRange{
StartTime: fn.Some(now),
StartHeight: fn.Some(uint32(1)),
}
return r.validateForVersion(
lnwire.GossipVersion1,
)
},
wantErr: "both time and block",
},
{
name: "v1 node range with time - ok",
fn: func() error {
r := NodeUpdateRange{
StartTime: fn.Some(now),
EndTime: fn.Some(now),
}
return r.validateForVersion(
lnwire.GossipVersion1,
)
},
},
{
name: "v2 node range with height - ok",
fn: func() error {
r := NodeUpdateRange{
StartHeight: fn.Some(uint32(1)),
EndHeight: fn.Some(uint32(100)),
}
return r.validateForVersion(
lnwire.GossipVersion2,
)
},
},
{
name: "v2 node range with time - rejected",
fn: func() error {
r := NodeUpdateRange{
StartTime: fn.Some(now),
EndTime: fn.Some(now),
}
return r.validateForVersion(
lnwire.GossipVersion2,
)
},
wantErr: "v2 node update range must use height",
},
{
name: "v1 chan range missing bounds - rejected",
fn: func() error {
r := ChanUpdateRange{
StartTime: fn.Some(now),
}
return r.validateForVersion(
lnwire.GossipVersion1,
)
},
wantErr: "missing time bounds",
},
{
name: "v1 chan range inverted - rejected",
fn: func() error {
r := ChanUpdateRange{
StartTime: fn.Some(now.Add(time.Hour)),
EndTime: fn.Some(now),
}
return r.validateForVersion(
lnwire.GossipVersion1,
)
},
wantErr: "start time after end time",
},
{
name: "v2 chan range inverted - rejected",
fn: func() error {
r := ChanUpdateRange{
StartHeight: fn.Some(uint32(100)),
EndHeight: fn.Some(uint32(50)),
}
return r.validateForVersion(
lnwire.GossipVersion2,
)
},
wantErr: "start height after end height",
},
{
name: "v1 node range inverted - rejected",
fn: func() error {
r := NodeUpdateRange{
StartTime: fn.Some(now.Add(time.Hour)),
EndTime: fn.Some(now),
}
return r.validateForVersion(
lnwire.GossipVersion1,
)
},
wantErr: "start time after end time",
},
{
name: "v2 node range inverted - rejected",
fn: func() error {
r := NodeUpdateRange{
StartHeight: fn.Some(uint32(100)),
EndHeight: fn.Some(uint32(50)),
}
return r.validateForVersion(
lnwire.GossipVersion2,
)
},
wantErr: "start height after end height",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
err := tc.fn()
if tc.wantErr == "" {
require.NoError(t, err)
} else {
require.ErrorContains(t, err,
tc.wantErr)
}
})
}
}
Reference in a new issue View git blame Copy permalink