Merge pull request #9236 from ellemouton/moveGraphDBCode

[1/3] Graph RIP: refactor+graph: move all graph related DB code to the graph package

autopilot/graph.go

@@ -3,6 +3,7 @@ package autopilot
 import (
 	"bytes"
 	"encoding/hex"
+	"errors"
 	"net"
 	"sort"
 	"sync/atomic"
@@ -11,8 +12,8 @@ import (
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/btcec/v2/ecdsa"
 	"github.com/btcsuite/btcd/btcutil"
-	"github.com/lightningnetwork/lnd/channeldb"
-	"github.com/lightningnetwork/lnd/channeldb/models"
+	graphdb "github.com/lightningnetwork/lnd/graph/db"
+	"github.com/lightningnetwork/lnd/graph/db/models"
 	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/routing/route"
@@ -35,7 +36,7 @@ var (
 //
 // TODO(roasbeef): move inmpl to main package?
 type databaseChannelGraph struct {
-	db *channeldb.ChannelGraph
+	db *graphdb.ChannelGraph
 }
 
 // A compile time assertion to ensure databaseChannelGraph meets the
@@ -44,7 +45,7 @@ var _ ChannelGraph = (*databaseChannelGraph)(nil)
 
 // ChannelGraphFromDatabase returns an instance of the autopilot.ChannelGraph
 // backed by a live, open channeldb instance.
-func ChannelGraphFromDatabase(db *channeldb.ChannelGraph) ChannelGraph {
+func ChannelGraphFromDatabase(db *graphdb.ChannelGraph) ChannelGraph {
 	return &databaseChannelGraph{
 		db: db,
 	}
@@ -54,11 +55,11 @@ func ChannelGraphFromDatabase(db *channeldb.ChannelGraph) ChannelGraph {
 // channeldb.LightningNode. The wrapper method implement the autopilot.Node
 // interface.
 type dbNode struct {
-	db *channeldb.ChannelGraph
+	db *graphdb.ChannelGraph
 
 	tx kvdb.RTx
 
-	node *channeldb.LightningNode
+	node *models.LightningNode
 }
 
 // A compile time assertion to ensure dbNode meets the autopilot.Node
@@ -134,7 +135,9 @@ func (d *dbNode) ForEachChannel(cb func(ChannelEdge) error) error {
 //
 // NOTE: Part of the autopilot.ChannelGraph interface.
 func (d *databaseChannelGraph) ForEachNode(cb func(Node) error) error {
-	return d.db.ForEachNode(func(tx kvdb.RTx, n *channeldb.LightningNode) error {
+	return d.db.ForEachNode(func(tx kvdb.RTx,
+		n *models.LightningNode) error {
+
 		// We'll skip over any node that doesn't have any advertised
 		// addresses. As we won't be able to reach them to actually
 		// open any channels.
@@ -157,7 +160,7 @@ func (d *databaseChannelGraph) ForEachNode(cb func(Node) error) error {
 func (d *databaseChannelGraph) addRandChannel(node1, node2 *btcec.PublicKey,
 	capacity btcutil.Amount) (*ChannelEdge, *ChannelEdge, error) {
 
-	fetchNode := func(pub *btcec.PublicKey) (*channeldb.LightningNode, error) {
+	fetchNode := func(pub *btcec.PublicKey) (*models.LightningNode, error) {
 		if pub != nil {
 			vertex, err := route.NewVertexFromBytes(
 				pub.SerializeCompressed(),
@@ -168,10 +171,10 @@ func (d *databaseChannelGraph) addRandChannel(node1, node2 *btcec.PublicKey,
 
 			dbNode, err := d.db.FetchLightningNode(vertex)
 			switch {
-			case err == channeldb.ErrGraphNodeNotFound:
+			case errors.Is(err, graphdb.ErrGraphNodeNotFound):
 				fallthrough
-			case err == channeldb.ErrGraphNotFound:
-				graphNode := &channeldb.LightningNode{
+			case errors.Is(err, graphdb.ErrGraphNotFound):
+				graphNode := &models.LightningNode{
 					HaveNodeAnnouncement: true,
 					Addresses: []net.Addr{
 						&net.TCPAddr{
@@ -198,7 +201,7 @@ func (d *databaseChannelGraph) addRandChannel(node1, node2 *btcec.PublicKey,
 		if err != nil {
 			return nil, err
 		}
-		dbNode := &channeldb.LightningNode{
+		dbNode := &models.LightningNode{
 			HaveNodeAnnouncement: true,
 			Addresses: []net.Addr{
 				&net.TCPAddr{
@@ -302,7 +305,7 @@ func (d *databaseChannelGraph) addRandNode() (*btcec.PublicKey, error) {
 	if err != nil {
 		return nil, err
 	}
-	dbNode := &channeldb.LightningNode{
+	dbNode := &models.LightningNode{
 		HaveNodeAnnouncement: true,
 		Addresses: []net.Addr{
 			&net.TCPAddr{
@@ -478,7 +481,7 @@ func (m *memChannelGraph) addRandNode() (*btcec.PublicKey, error) {
 // databaseChannelGraphCached wraps a channeldb.ChannelGraph instance with the
 // necessary API to properly implement the autopilot.ChannelGraph interface.
 type databaseChannelGraphCached struct {
-	db *channeldb.ChannelGraph
+	db *graphdb.ChannelGraph
 }
 
 // A compile time assertion to ensure databaseChannelGraphCached meets the
@@ -487,7 +490,7 @@ var _ ChannelGraph = (*databaseChannelGraphCached)(nil)
 
 // ChannelGraphFromCachedDatabase returns an instance of the
 // autopilot.ChannelGraph backed by a live, open channeldb instance.
-func ChannelGraphFromCachedDatabase(db *channeldb.ChannelGraph) ChannelGraph {
+func ChannelGraphFromCachedDatabase(db *graphdb.ChannelGraph) ChannelGraph {
 	return &databaseChannelGraphCached{
 		db: db,
 	}
@@ -498,7 +501,7 @@ func ChannelGraphFromCachedDatabase(db *channeldb.ChannelGraph) ChannelGraph {
 // interface.
 type dbNodeCached struct {
 	node     route.Vertex
-	channels map[uint64]*channeldb.DirectedChannel
+	channels map[uint64]*graphdb.DirectedChannel
 }
 
 // A compile time assertion to ensure dbNodeCached meets the autopilot.Node
@@ -552,7 +555,7 @@ func (nc dbNodeCached) ForEachChannel(cb func(ChannelEdge) error) error {
 // NOTE: Part of the autopilot.ChannelGraph interface.
 func (dc *databaseChannelGraphCached) ForEachNode(cb func(Node) error) error {
 	return dc.db.ForEachNodeCached(func(n route.Vertex,
-		channels map[uint64]*channeldb.DirectedChannel) error {
+		channels map[uint64]*graphdb.DirectedChannel) error {
 
 		if len(channels) > 0 {
 			node := dbNodeCached{

autopilot/prefattach_test.go

@@ -8,7 +8,8 @@ import (
 
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/btcutil"
-	"github.com/lightningnetwork/lnd/channeldb"
+	graphdb "github.com/lightningnetwork/lnd/graph/db"
+	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/stretchr/testify/require"
 )
 
@@ -24,17 +25,21 @@ type testGraph interface {
 }
 
 func newDiskChanGraph(t *testing.T) (testGraph, error) {
-	// Next, create channeldb for the first time.
-	cdb, err := channeldb.Open(t.TempDir())
-	if err != nil {
-		return nil, err
-	}
-	t.Cleanup(func() {
-		require.NoError(t, cdb.Close())
+	backend, err := kvdb.GetBoltBackend(&kvdb.BoltBackendConfig{
+		DBPath:            t.TempDir(),
+		DBFileName:        "graph.db",
+		NoFreelistSync:    true,
+		AutoCompact:       false,
+		AutoCompactMinAge: kvdb.DefaultBoltAutoCompactMinAge,
+		DBTimeout:         kvdb.DefaultDBTimeout,
 	})
+	require.NoError(t, err)
+
+	graphDB, err := graphdb.NewChannelGraph(backend)
+	require.NoError(t, err)
 
 	return &databaseChannelGraph{
-		db: cdb.ChannelGraph(),
+		db: graphDB,
 	}, nil
 }
 

chainntnfs/bitcoindnotify/bitcoind_test.go

@@ -37,11 +37,7 @@ var (
 func initHintCache(t *testing.T) *channeldb.HeightHintCache {
 	t.Helper()
 
-	db, err := channeldb.Open(t.TempDir())
-	require.NoError(t, err, "unable to create db")
-	t.Cleanup(func() {
-		require.NoError(t, db.Close())
-	})
+	db := channeldb.OpenForTesting(t, t.TempDir())
 
 	testCfg := channeldb.CacheConfig{
 		QueryDisable: false,

chainntnfs/btcdnotify/btcd_test.go

@@ -33,11 +33,7 @@ var (
 func initHintCache(t *testing.T) *channeldb.HeightHintCache {
 	t.Helper()
 
-	db, err := channeldb.Open(t.TempDir())
-	require.NoError(t, err, "unable to create db")
-	t.Cleanup(func() {
-		require.NoError(t, db.Close())
-	})
+	db := channeldb.OpenForTesting(t, t.TempDir())
 
 	testCfg := channeldb.CacheConfig{
 		QueryDisable: false,

chainntnfs/test/test_interface.go

@@ -1906,10 +1906,8 @@ func TestInterfaces(t *testing.T, targetBackEnd string) {
 
 		// Initialize a height hint cache for each notifier.
 		tempDir := t.TempDir()
-		db, err := channeldb.Open(tempDir)
-		if err != nil {
-			t.Fatalf("unable to create db: %v", err)
-		}
+		db := channeldb.OpenForTesting(t, tempDir)
+
 		testCfg := channeldb.CacheConfig{
 			QueryDisable: false,
 		}

chainreg/chainregistry.go

@@ -23,8 +23,8 @@ import (
 	"github.com/lightningnetwork/lnd/chainntnfs/btcdnotify"
 	"github.com/lightningnetwork/lnd/chainntnfs/neutrinonotify"
 	"github.com/lightningnetwork/lnd/channeldb"
-	"github.com/lightningnetwork/lnd/channeldb/models"
 	"github.com/lightningnetwork/lnd/fn"
+	"github.com/lightningnetwork/lnd/graph/db/models"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/keychain"
 	"github.com/lightningnetwork/lnd/kvdb"

chainreg/no_chain_backend.go

@@ -11,7 +11,7 @@ import (
 	"github.com/btcsuite/btcwallet/chain"
 	"github.com/btcsuite/btcwallet/waddrmgr"
 	"github.com/lightningnetwork/lnd/chainntnfs"
-	"github.com/lightningnetwork/lnd/channeldb"
+	graphdb "github.com/lightningnetwork/lnd/graph/db"
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
 	"github.com/lightningnetwork/lnd/routing/chainview"
 )
@@ -94,7 +94,7 @@ func (n *NoChainBackend) DisconnectedBlocks() <-chan *chainview.FilteredBlock {
 	return make(chan *chainview.FilteredBlock)
 }
 
-func (n *NoChainBackend) UpdateFilter([]channeldb.EdgePoint, uint32) error {
+func (n *NoChainBackend) UpdateFilter([]graphdb.EdgePoint, uint32) error {
 	return nil
 }
 

chanbackup/backup.go

@@ -2,13 +2,10 @@ package chanbackup
 
 import (
 	"fmt"
-	"net"
 
-	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/fn"
-	"github.com/lightningnetwork/lnd/kvdb"
 )
 
 // LiveChannelSource is an interface that allows us to query for the set of
@@ -20,34 +17,28 @@ type LiveChannelSource interface {
 
 	// FetchChannel attempts to locate a live channel identified by the
 	// passed chanPoint. Optionally an existing db tx can be supplied.
-	FetchChannel(tx kvdb.RTx, chanPoint wire.OutPoint) (
-		*channeldb.OpenChannel, error)
-}
-
-// AddressSource is an interface that allows us to query for the set of
-// addresses a node can be connected to.
-type AddressSource interface {
-	// AddrsForNode returns all known addresses for the target node public
-	// key.
-	AddrsForNode(nodePub *btcec.PublicKey) ([]net.Addr, error)
+	FetchChannel(chanPoint wire.OutPoint) (*channeldb.OpenChannel, error)
 }
 
 // assembleChanBackup attempts to assemble a static channel backup for the
 // passed open channel. The backup includes all information required to restore
 // the channel, as well as addressing information so we can find the peer and
 // reconnect to them to initiate the protocol.
-func assembleChanBackup(addrSource AddressSource,
+func assembleChanBackup(addrSource channeldb.AddrSource,
 	openChan *channeldb.OpenChannel) (*Single, error) {
 
 	log.Debugf("Crafting backup for ChannelPoint(%v)",
 		openChan.FundingOutpoint)
 
 	// First, we'll query the channel source to obtain all the addresses
 	// that are associated with the peer for this channel.
-	nodeAddrs, err := addrSource.AddrsForNode(openChan.IdentityPub)
+	known, nodeAddrs, err := addrSource.AddrsForNode(openChan.IdentityPub)
 	if err != nil {
 		return nil, err
 	}
+	if !known {
+		return nil, fmt.Errorf("node unknown by address source")
+	}
 
 	single := NewSingle(openChan, nodeAddrs)
 
@@ -100,11 +91,11 @@ func buildCloseTxInputs(
 // the target channel identified by its channel point. If we're unable to find
 // the target channel, then an error will be returned.
 func FetchBackupForChan(chanPoint wire.OutPoint, chanSource LiveChannelSource,
-	addrSource AddressSource) (*Single, error) {
+	addrSource channeldb.AddrSource) (*Single, error) {
 
 	// First, we'll query the channel source to see if the channel is known
 	// and open within the database.
-	targetChan, err := chanSource.FetchChannel(nil, chanPoint)
+	targetChan, err := chanSource.FetchChannel(chanPoint)
 	if err != nil {
 		// If we can't find the channel, then we return with an error,
 		// as we have nothing to  backup.
@@ -124,7 +115,7 @@ func FetchBackupForChan(chanPoint wire.OutPoint, chanSource LiveChannelSource,
 // FetchStaticChanBackups will return a plaintext static channel back up for
 // all known active/open channels within the passed channel source.
 func FetchStaticChanBackups(chanSource LiveChannelSource,
-	addrSource AddressSource) ([]Single, error) {
+	addrSource channeldb.AddrSource) ([]Single, error) {
 
 	// First, we'll query the backup source for information concerning all
 	// currently open and available channels.

chanbackup/backup_test.go

@@ -8,7 +8,6 @@ import (
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/channeldb"
-	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/stretchr/testify/require"
 )
 
@@ -40,7 +39,7 @@ func (m *mockChannelSource) FetchAllChannels() ([]*channeldb.OpenChannel, error)
 	return chans, nil
 }
 
-func (m *mockChannelSource) FetchChannel(_ kvdb.RTx, chanPoint wire.OutPoint) (
+func (m *mockChannelSource) FetchChannel(chanPoint wire.OutPoint) (
 	*channeldb.OpenChannel, error) {
 
 	if m.failQuery {
@@ -62,20 +61,19 @@ func (m *mockChannelSource) addAddrsForNode(nodePub *btcec.PublicKey, addrs []ne
 	m.addrs[nodeKey] = addrs
 }
 
-func (m *mockChannelSource) AddrsForNode(nodePub *btcec.PublicKey) ([]net.Addr, error) {
+func (m *mockChannelSource) AddrsForNode(nodePub *btcec.PublicKey) (bool,
+	[]net.Addr, error) {
+
 	if m.failQuery {
-		return nil, fmt.Errorf("fail")
+		return false, nil, fmt.Errorf("fail")
 	}
 
 	var nodeKey [33]byte
 	copy(nodeKey[:], nodePub.SerializeCompressed())
 
 	addrs, ok := m.addrs[nodeKey]
-	if !ok {
-		return nil, fmt.Errorf("can't find addr")
-	}
 
-	return addrs, nil
+	return ok, addrs, nil
 }
 
 // TestFetchBackupForChan tests that we're able to construct a single channel