diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index f553482bb..f20c3a480 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -49,6 +49,11 @@ func newDiskChanGraph(t *testing.T) (testGraph, error) { graphDB, err := graphdb.NewChannelGraph(&graphdb.Config{KVDB: backend}) require.NoError(t, err) + require.NoError(t, graphDB.Start()) + t.Cleanup(func() { + require.NoError(t, graphDB.Stop()) + }) + return &testDBGraph{ db: graphDB, databaseChannelGraph: databaseChannelGraph{ diff --git a/graph/db/graph.go b/graph/db/graph.go index 6e2132a5f..f926aaea3 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -3,6 +3,7 @@ package graphdb import ( "errors" "sync" + "sync/atomic" "time" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -32,13 +33,20 @@ type Config struct { // KVStore. Upcoming commits will move the graph cache out of the KVStore and // into this layer so that the KVStore is only responsible for CRUD operations. type ChannelGraph struct { + started atomic.Bool + stopped atomic.Bool + // cacheMu guards any writes to the graphCache. It should be held // across the DB write call and the graphCache update to make the // two updates as atomic as possible. - cacheMu sync.Mutex + cacheMu sync.Mutex + graphCache *GraphCache *KVStore + + quit chan struct{} + wg sync.WaitGroup } // NewChannelGraph creates a new ChannelGraph instance with the given backend. @@ -58,6 +66,7 @@ func NewChannelGraph(cfg *Config, options ...ChanGraphOption) (*ChannelGraph, if !opts.useGraphCache { return &ChannelGraph{ KVStore: store, + quit: make(chan struct{}), }, nil } @@ -96,9 +105,38 @@ func NewChannelGraph(cfg *Config, options ...ChanGraphOption) (*ChannelGraph, return &ChannelGraph{ KVStore: store, graphCache: graphCache, + quit: make(chan struct{}), }, nil } +// Start kicks off any goroutines required for the ChannelGraph to function. +// If the graph cache is enabled, then it will be populated with the contents of +// the database. +func (c *ChannelGraph) Start() error { + if !c.started.CompareAndSwap(false, true) { + return nil + } + log.Debugf("ChannelGraph starting") + defer log.Debug("ChannelGraph started") + + return nil +} + +// Stop signals any active goroutines for a graceful closure. +func (c *ChannelGraph) Stop() error { + if !c.stopped.CompareAndSwap(false, true) { + return nil + } + + log.Debugf("ChannelGraph shutting down...") + defer log.Debug("ChannelGraph shutdown complete") + + close(c.quit) + c.wg.Wait() + + return nil +} + // ForEachNodeDirectedChannel iterates through all channels of a given node, // executing the passed callback on the directed edge representing the channel // and its incoming policy. If the callback returns an error, then the iteration diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index bc130c599..dc41f02a2 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -4077,6 +4077,10 @@ func TestGraphLoading(t *testing.T) { graph, err := NewChannelGraph(&Config{KVDB: backend}) 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 @@ -4087,6 +4091,10 @@ func TestGraphLoading(t *testing.T) { // populated. graphReloaded, err := NewChannelGraph(&Config{KVDB: backend}) 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( diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 0ea61a76c..aa120a39a 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -26,6 +26,7 @@ import ( "github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing/route" + "github.com/stretchr/testify/require" ) var ( @@ -4722,10 +4723,12 @@ func MakeTestGraph(t testing.TB, modifiers ...KVStoreOptionModifier) ( return nil, err } + require.NoError(t, graph.Start()) t.Cleanup(func() { _ = backend.Close() backendCleanup() + require.NoError(t, graph.Stop()) }) return graph, nil diff --git a/graph/notifications_test.go b/graph/notifications_test.go index aab1d4137..807b3fea7 100644 --- a/graph/notifications_test.go +++ b/graph/notifications_test.go @@ -1100,6 +1100,10 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph, if err != nil { return nil, nil, err } + require.NoError(t, graph.Start()) + t.Cleanup(func() { + require.NoError(t, graph.Stop()) + }) return graph, backend, nil } diff --git a/peer/test_utils.go b/peer/test_utils.go index 7d3c5ca37..87a80712f 100644 --- a/peer/test_utils.go +++ b/peer/test_utils.go @@ -619,6 +619,10 @@ func createTestPeer(t *testing.T) *peerTestCtx { KVDB: graphBackend, }) require.NoError(t, err) + require.NoError(t, dbAliceGraph.Start()) + t.Cleanup(func() { + require.NoError(t, dbAliceGraph.Stop()) + }) dbAliceChannel := channeldb.OpenForTesting(t, dbPath) diff --git a/routing/pathfind_test.go b/routing/pathfind_test.go index 029383e0b..8a0280686 100644 --- a/routing/pathfind_test.go +++ b/routing/pathfind_test.go @@ -173,6 +173,10 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph, if err != nil { return nil, nil, err } + require.NoError(t, graph.Start()) + t.Cleanup(func() { + require.NoError(t, graph.Stop()) + }) return graph, backend, nil } diff --git a/server.go b/server.go index ad521cd67..731d69797 100644 --- a/server.go +++ b/server.go @@ -2377,6 +2377,12 @@ func (s *server) Start() error { return } + cleanup = cleanup.add(s.graphDB.Stop) + if err := s.graphDB.Start(); err != nil { + startErr = err + return + } + cleanup = cleanup.add(s.graphBuilder.Stop) if err := s.graphBuilder.Start(); err != nil { startErr = err @@ -2683,6 +2689,9 @@ func (s *server) Stop() error { if err := s.graphBuilder.Stop(); err != nil { srvrLog.Warnf("failed to stop graphBuilder %v", err) } + if err := s.graphDB.Stop(); err != nil { + srvrLog.Warnf("failed to stop graphDB %v", err) + } if err := s.chainArb.Stop(); err != nil { srvrLog.Warnf("failed to stop chainArb: %v", err) }