Merge pull request #4934 from cfromknecht/pinned-active-syncers

discovery: pinned syncers

config.go

@@ -309,6 +309,8 @@ type Config struct {
 
 	Routing *lncfg.Routing `group:"routing" namespace:"routing"`
 
+	Gossip *lncfg.Gossip `group:"gossip" namespace:"gossip"`
+
 	Workers *lncfg.Workers `group:"workers" namespace:"workers"`
 
 	Caches *lncfg.Caches `group:"caches" namespace:"caches"`
@@ -477,6 +479,7 @@ func DefaultConfig() Config {
 				Backoff:  defaultTLSBackoff,
 			},
 		},
+		Gossip:                  &lncfg.Gossip{},
 		MaxOutgoingCltvExpiry:   htlcswitch.DefaultMaxOutgoingCltvExpiry,
 		MaxChannelFeeAllocation: htlcswitch.DefaultMaxLinkFeeAllocation,
 		MaxCommitFeeRateAnchors: lnwallet.DefaultAnchorsCommitMaxFeeRateSatPerVByte,
@@ -1289,6 +1292,10 @@ func ValidateConfig(cfg Config, usageMessage string) (*Config, error) {
 			maxRemoteHtlcs)
 	}
 
+	if err := cfg.Gossip.Parse(); err != nil {
+		return nil, err
+	}
+
 	// Validate the subconfigs for workers, caches, and the tower client.
 	err = lncfg.Validate(
 		cfg.Workers,

discovery/gossiper.go

@@ -93,6 +93,10 @@ type chanPolicyUpdateRequest struct {
 	errChan       chan error
 }
 
+// PinnedSyncers is a set of node pubkeys for which we will maintain an active
+// syncer at all times.
+type PinnedSyncers map[route.Vertex]struct{}
+
 // Config defines the configuration for the service. ALL elements within the
 // configuration MUST be non-nil for the service to carry out its duties.
 type Config struct {
@@ -233,6 +237,11 @@ type Config struct {
 	// gossip updates to once per RebroadcastInterval for any keep-alive
 	// updates, and once per block for other types of updates.
 	GossipUpdateThrottle bool
+
+	// PinnedSyncers is a set of peers that will always transition to
+	// ActiveSync upon connection. These peers will never transition to
+	// PassiveSync.
+	PinnedSyncers PinnedSyncers
 }
 
 // AuthenticatedGossiper is a subsystem which is responsible for receiving
@@ -334,6 +343,7 @@ func New(cfg Config, selfKey *btcec.PublicKey) *AuthenticatedGossiper {
 		NumActiveSyncers:        cfg.NumActiveSyncers,
 		IgnoreHistoricalFilters: cfg.IgnoreHistoricalFilters,
 		BestHeight:              gossiper.latestHeight,
+		PinnedSyncers:           cfg.PinnedSyncers,
 	})
 
 	gossiper.reliableSender = newReliableSender(&reliableSenderCfg{

discovery/gossiper_test.go

@@ -12,6 +12,7 @@ import (
 	"reflect"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -3616,6 +3617,12 @@ func TestCalculateCorrectSubBatchSizesDifferentDelay(t *testing.T) {
 	}
 }
 
+// markGraphSynced allows us to report that the initial historical sync has
+// completed.
+func (m *SyncManager) markGraphSyncing() {
+	atomic.StoreInt32(&m.initialHistoricalSyncCompleted, 0)
+}
+
 // TestBroadcastAnnsAfterGraphSynced ensures that we only broadcast
 // announcements after the graph has been considered as synced, i.e., after our
 // initial historical sync has completed.

discovery/sync_manager.go

@@ -92,6 +92,11 @@ type SyncManagerCfg struct {
 
 	// BestHeight returns the latest height known of the chain.
 	BestHeight func() uint32
+
+	// PinnedSyncers is a set of peers that will always transition to
+	// ActiveSync upon connection. These peers will never transition to
+	// PassiveSync.
+	PinnedSyncers PinnedSyncers
 }
 
 // SyncManager is a subsystem of the gossiper that manages the gossip syncers
@@ -140,6 +145,12 @@ type SyncManager struct {
 	// currently receiving new graph updates from.
 	inactiveSyncers map[route.Vertex]*GossipSyncer
 
+	// pinnedActiveSyncers is the set of all syncers which are pinned into
+	// an active sync. Pinned peers performan an initial historical sync on
+	// each connection and will continue to receive graph updates for the
+	// duration of the connection.
+	pinnedActiveSyncers map[route.Vertex]*GossipSyncer
+
 	wg   sync.WaitGroup
 	quit chan struct{}
 }
@@ -154,7 +165,10 @@ func newSyncManager(cfg *SyncManagerCfg) *SyncManager {
 			map[route.Vertex]*GossipSyncer, cfg.NumActiveSyncers,
 		),
 		inactiveSyncers: make(map[route.Vertex]*GossipSyncer),
-		quit:            make(chan struct{}),
+		pinnedActiveSyncers: make(
+			map[route.Vertex]*GossipSyncer, len(cfg.PinnedSyncers),
+		),
+		quit: make(chan struct{}),
 	}
 }
 
@@ -198,7 +212,6 @@ func (m *SyncManager) syncerHandler() {
 	m.cfg.RotateTicker.Resume()
 	defer m.cfg.RotateTicker.Stop()
 
-	m.cfg.HistoricalSyncTicker.Resume()
 	defer m.cfg.HistoricalSyncTicker.Stop()
 
 	var (
@@ -214,6 +227,19 @@ func (m *SyncManager) syncerHandler() {
 		initialHistoricalSyncSignal chan struct{}
 	)
 
+	setInitialHistoricalSyncer := func(s *GossipSyncer) {
+		initialHistoricalSyncer = s
+		initialHistoricalSyncSignal = s.ResetSyncedSignal()
+
+		// Restart the timer for our new historical sync peer. This will
+		// ensure that all initial syncers recevie an equivalent
+		// duration before attempting the next sync. Without doing so we
+		// might attempt two historical sync back to back if a peer
+		// disconnects just before the ticker fires.
+		m.cfg.HistoricalSyncTicker.Pause()
+		m.cfg.HistoricalSyncTicker.Resume()
+	}
+
 	for {
 		select {
 		// A new peer has been connected, so we'll create its
@@ -228,6 +254,8 @@ func (m *SyncManager) syncerHandler() {
 
 			s := m.createGossipSyncer(newSyncer.peer)
 
+			isPinnedSyncer := m.isPinnedSyncer(s)
+
 			// attemptHistoricalSync determines whether we should
 			// attempt an initial historical sync when a new peer
 			// connects.
@@ -235,6 +263,14 @@ func (m *SyncManager) syncerHandler() {
 
 			m.syncersMu.Lock()
 			switch {
+			// For pinned syncers, we will immediately transition
+			// the peer into an active (pinned) sync state.
+			case isPinnedSyncer:
+				attemptHistoricalSync = true
+				s.setSyncType(PinnedSync)
+				s.setSyncState(syncerIdle)
+				m.pinnedActiveSyncers[s.cfg.peerPub] = s
+
 			// Regardless of whether the initial historical sync
 			// has completed, we'll re-trigger a historical sync if
 			// we no longer have any syncers. This might be
@@ -280,7 +316,6 @@ func (m *SyncManager) syncerHandler() {
 			if !attemptHistoricalSync {
 				continue
 			}
-			m.markGraphSyncing()
 
 			log.Debugf("Attempting initial historical sync with "+
 				"GossipSyncer(%x)", s.cfg.peerPub)
@@ -297,8 +332,9 @@ func (m *SyncManager) syncerHandler() {
 			// keep track of the corresponding syncer to properly
 			// handle disconnects. We'll also use a signal to know
 			// when the historical sync completed.
-			initialHistoricalSyncer = s
+			if !isPinnedSyncer {
-			initialHistoricalSyncSignal = s.ResetSyncedSignal()
+				setInitialHistoricalSyncer(s)
+			}
 
 		// An existing peer has disconnected, so we'll tear down its
 		// corresponding GossipSyncer.
@@ -337,15 +373,13 @@ func (m *SyncManager) syncerHandler() {
 				"GossipSyncer(%v) with GossipSyncer(%x)",
 				staleSyncer.peer, s.cfg.peerPub)
 
-			initialHistoricalSyncer = s
+			setInitialHistoricalSyncer(s)
-			initialHistoricalSyncSignal = s.ResetSyncedSignal()
 
 		// Our initial historical sync signal has completed, so we'll
 		// nil all of the relevant fields as they're no longer needed.
 		case <-initialHistoricalSyncSignal:
 			initialHistoricalSyncer = nil
 			initialHistoricalSyncSignal = nil
-			m.markGraphSynced()
 
 			log.Debug("Initial historical sync completed")
 
@@ -380,12 +414,16 @@ func (m *SyncManager) syncerHandler() {
 		// Our HistoricalSyncTicker has ticked, so we'll randomly select
 		// a peer and force a historical sync with them.
 		case <-m.cfg.HistoricalSyncTicker.Ticks():
+			// If we don't have a syncer available we have nothing
+			// to do.
 			s := m.forceHistoricalSync()
+			if s == nil {
+				continue
+			}
 
-			// If we don't have a syncer available or we've already
+			// If we've already completed a historical sync, we'll
-			// performed our initial historical sync, then we have
+			// skip setting the initial historical syncer.
-			// nothing left to do.
+			if m.IsGraphSynced() {
-			if s == nil || m.IsGraphSynced() {
 				continue
 			}
 
@@ -394,8 +432,7 @@ func (m *SyncManager) syncerHandler() {
 			// where our previous historical sync peer did not
 			// respond to our queries and we haven't ingested as
 			// much of the graph as we should.
-			initialHistoricalSyncer = s
+			setInitialHistoricalSyncer(s)
-			initialHistoricalSyncSignal = s.ResetSyncedSignal()
 
 		case <-m.quit:
 			return
@@ -403,6 +440,13 @@ func (m *SyncManager) syncerHandler() {
 	}
 }
 
+// isPinnedSyncer returns true if the passed GossipSyncer is one of our pinned
+// sync peers.
+func (m *SyncManager) isPinnedSyncer(s *GossipSyncer) bool {
+	_, isPinnedSyncer := m.cfg.PinnedSyncers[s.cfg.peerPub]
+	return isPinnedSyncer
+}
+
 // createGossipSyncer creates the GossipSyncer for a newly connected peer.
 func (m *SyncManager) createGossipSyncer(peer lnpeer.Peer) *GossipSyncer {
 	nodeID := route.Vertex(peer.PubKey())
@@ -426,6 +470,7 @@ func (m *SyncManager) createGossipSyncer(peer lnpeer.Peer) *GossipSyncer {
 		maxUndelayedQueryReplies:  DefaultMaxUndelayedQueryReplies,
 		delayedQueryReplyInterval: DefaultDelayedQueryReplyInterval,
 		bestHeight:                m.cfg.BestHeight,
+		markGraphSynced:           m.markGraphSynced,
 		maxQueryChanRangeReplies:  maxQueryChanRangeReplies,
 	})
 
@@ -461,6 +506,13 @@ func (m *SyncManager) removeGossipSyncer(peer route.Vertex) {
 		return
 	}
 
+	// If it's a pinned syncer, then we can just exit as this doesn't
+	// affect our active syncer count.
+	if _, ok := m.pinnedActiveSyncers[peer]; ok {
+		delete(m.pinnedActiveSyncers, peer)
+		return
+	}
+
 	// Otherwise, we'll need find a new one to replace it, if any.
 	delete(m.activeSyncers, peer)
 	newActiveSyncer := chooseRandomSyncer(
@@ -663,6 +715,10 @@ func (m *SyncManager) gossipSyncer(peer route.Vertex) (*GossipSyncer, bool) {
 	if ok {
 		return syncer, true
 	}
+	syncer, ok = m.pinnedActiveSyncers[peer]
+	if ok {
+		return syncer, true
+	}
 	return nil, false
 }
 
@@ -694,12 +750,6 @@ func (m *SyncManager) markGraphSynced() {
 	atomic.StoreInt32(&m.initialHistoricalSyncCompleted, 1)
 }
 
-// markGraphSyncing allows us to report that the initial historical sync is
-// still undergoing.
-func (m *SyncManager) markGraphSyncing() {
-	atomic.StoreInt32(&m.initialHistoricalSyncCompleted, 0)
-}
-
 // IsGraphSynced determines whether we've completed our initial historical sync.
 // The initial historical sync is done to ensure we've ingested as much of the
 // public graph as possible.

discovery/sync_manager_test.go

@@ -7,9 +7,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btcd/btcec"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightningnetwork/lnd/lntest/wait"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/routing/route"
 	"github.com/lightningnetwork/lnd/ticker"
 	"github.com/stretchr/testify/require"
 )
@@ -18,8 +20,13 @@ import (
 func randPeer(t *testing.T, quit chan struct{}) *mockPeer {
 	t.Helper()
 
+	pk := randPubKey(t)
+	return peerWithPubkey(pk, quit)
+}
+
+func peerWithPubkey(pk *btcec.PublicKey, quit chan struct{}) *mockPeer {
 	return &mockPeer{
-		pk:       randPubKey(t),
+		pk:       pk,
 		sentMsgs: make(chan lnwire.Message),
 		quit:     quit,
 	}
@@ -28,6 +35,14 @@ func randPeer(t *testing.T, quit chan struct{}) *mockPeer {
 // newTestSyncManager creates a new test SyncManager using mock implementations
 // of its dependencies.
 func newTestSyncManager(numActiveSyncers int) *SyncManager {
+	return newPinnedTestSyncManager(numActiveSyncers, nil)
+}
+
+// newTestSyncManager creates a new test SyncManager with a set of pinned
+// syncers using mock implementations of its dependencies.
+func newPinnedTestSyncManager(numActiveSyncers int,
+	pinnedSyncers PinnedSyncers) *SyncManager {
+
 	hID := lnwire.ShortChannelID{BlockHeight: latestKnownHeight}
 	return newSyncManager(&SyncManagerCfg{
 		ChanSeries:           newMockChannelGraphTimeSeries(hID),
@@ -37,6 +52,7 @@ func newTestSyncManager(numActiveSyncers int) *SyncManager {
 		BestHeight: func() uint32 {
 			return latestKnownHeight
 		},
+		PinnedSyncers: pinnedSyncers,
 	})
 }
 
@@ -48,17 +64,45 @@ func TestSyncManagerNumActiveSyncers(t *testing.T) {
 	// We'll start by creating our test sync manager which will hold up to
 	// 3 active syncers.
 	const numActiveSyncers = 3
-	const numSyncers = numActiveSyncers + 1
+	const numPinnedSyncers = 3
+	const numInactiveSyncers = 1
 
-	syncMgr := newTestSyncManager(numActiveSyncers)
+	pinnedSyncers := make(PinnedSyncers)
+	pinnedPubkeys := make(map[route.Vertex]*btcec.PublicKey)
+	for i := 0; i < numPinnedSyncers; i++ {
+		pubkey := randPubKey(t)
+		vertex := route.NewVertex(pubkey)
+
+		pinnedSyncers[vertex] = struct{}{}
+		pinnedPubkeys[vertex] = pubkey
+
+	}
+
+	syncMgr := newPinnedTestSyncManager(numActiveSyncers, pinnedSyncers)
 	syncMgr.Start()
 	defer syncMgr.Stop()
 
+	// First we'll start by adding the pinned syncers. These should
+	// immediately be assigned PinnedSync.
+	for _, pubkey := range pinnedPubkeys {
+		peer := peerWithPubkey(pubkey, syncMgr.quit)
+		err := syncMgr.InitSyncState(peer)
+		require.NoError(t, err)
+
+		s := assertSyncerExistence(t, syncMgr, peer)
+		assertTransitionToChansSynced(t, s, peer)
+		assertActiveGossipTimestampRange(t, peer)
+		assertSyncerStatus(t, s, chansSynced, PinnedSync)
+	}
+
 	// We'll go ahead and create our syncers. We'll gather the ones which
-	// should be active and passive to check them later on.
+	// should be active and passive to check them later on. The pinned peers
+	// added above should not influence the active syncer count.
 	for i := 0; i < numActiveSyncers; i++ {
 		peer := randPeer(t, syncMgr.quit)
-		syncMgr.InitSyncState(peer)
+		err := syncMgr.InitSyncState(peer)
+		require.NoError(t, err)
+
 		s := assertSyncerExistence(t, syncMgr, peer)
 
 		// The first syncer registered always attempts a historical
@@ -70,9 +114,11 @@ func TestSyncManagerNumActiveSyncers(t *testing.T) {
 		assertSyncerStatus(t, s, chansSynced, ActiveSync)
 	}
 
-	for i := 0; i < numSyncers-numActiveSyncers; i++ {
+	for i := 0; i < numInactiveSyncers; i++ {
 		peer := randPeer(t, syncMgr.quit)
-		syncMgr.InitSyncState(peer)
+		err := syncMgr.InitSyncState(peer)
+		require.NoError(t, err)
+
 		s := assertSyncerExistence(t, syncMgr, peer)
 		assertSyncerStatus(t, s, chansSynced, PassiveSync)
 	}

discovery/syncer.go

@@ -42,6 +42,13 @@ const (
 	// They are started in a chansSynced state in order to accomplish their
 	// responsibilities above.
 	PassiveSync
+
+	// PinnedSync denotes an ActiveSync that doesn't count towards the
+	// default active syncer limits and is always active throughout the
+	// duration of the peer's connection. Each pinned syncer will begin by
+	// performing a historical sync to ensure we are well synchronized with
+	// their routing table.
+	PinnedSync
 )
 
 // String returns a human readable string describing the target SyncerType.
@@ -51,11 +58,24 @@ func (t SyncerType) String() string {
 		return "ActiveSync"
 	case PassiveSync:
 		return "PassiveSync"
+	case PinnedSync:
+		return "PinnedSync"
 	default:
 		return fmt.Sprintf("unknown sync type %d", t)
 	}
 }
 
+// IsActiveSync returns true if the SyncerType should set a GossipTimestampRange
+// allowing new gossip messages to be received from the peer.
+func (t SyncerType) IsActiveSync() bool {
+	switch t {
+	case ActiveSync, PinnedSync:
+		return true
+	default:
+		return false
+	}
+}
+
 // syncerState is an enum that represents the current state of the GossipSyncer.
 // As the syncer is a state machine, we'll gate our actions based off of the
 // current state and the next incoming message.
@@ -95,6 +115,12 @@ const (
 	// AuthenticatedGossiper, and decide if we should forward them to our
 	// target peer based on its update horizon.
 	chansSynced
+
+	// syncerIdle is a state in which the gossip syncer can handle external
+	// requests to transition or perform historical syncs. It is used as the
+	// initial state for pinned syncers, as well as a fallthrough case for
+	// chansSynced allowing fully synced peers to facilitate requests.
+	syncerIdle
 )
 
 // String returns a human readable string describing the target syncerState.
@@ -115,6 +141,9 @@ func (s syncerState) String() string {
 	case chansSynced:
 		return "chansSynced"
 
+	case syncerIdle:
+		return "syncerIdle"
+
 	default:
 		return "UNKNOWN STATE"
 	}
@@ -251,6 +280,10 @@ type gossipSyncerCfg struct {
 	// bestHeight returns the latest height known of the chain.
 	bestHeight func() uint32
 
+	// markGraphSynced updates the SyncManager's perception of whether we
+	// have completed at least one historical sync.
+	markGraphSynced func()
+
 	// maxQueryChanRangeReplies is the maximum number of replies we'll allow
 	// for a single QueryChannelRange request.
 	maxQueryChanRangeReplies uint32
@@ -521,6 +554,11 @@ func (g *GossipSyncer) channelGraphSyncer() {
 			// to our terminal state.
 			g.setSyncState(chansSynced)
 
+			// Ensure that the sync manager becomes aware that the
+			// historical sync completed so synced_to_graph is
+			// updated over rpc.
+			g.cfg.markGraphSynced()
+
 		// In this state, we've just sent off a new query for channels
 		// that we don't yet know of. We'll remain in this state until
 		// the remote party signals they've responded to our query in
@@ -560,7 +598,9 @@ func (g *GossipSyncer) channelGraphSyncer() {
 			// If we haven't yet sent out our update horizon, and
 			// we want to receive real-time channel updates, we'll
 			// do so now.
-			if g.localUpdateHorizon == nil && syncType == ActiveSync {
+			if g.localUpdateHorizon == nil &&
+				syncType.IsActiveSync() {
+
 				err := g.sendGossipTimestampRange(
 					time.Now(), math.MaxUint32,
 				)
@@ -570,10 +610,16 @@ func (g *GossipSyncer) channelGraphSyncer() {
 						g.cfg.peerPub, err)
 				}
 			}
-
 			// With our horizon set, we'll simply reply to any new
 			// messages or process any state transitions and exit if
 			// needed.
+			fallthrough
+
+		// Pinned peers will begin in this state, since they will
+		// immediately receive a request to perform a historical sync.
+		// Otherwise, we fall through after ending in chansSynced to
+		// facilitate new requests.
+		case syncerIdle:
 			select {
 			case req := <-g.syncTransitionReqs:
 				req.errChan <- g.handleSyncTransition(req)
@@ -825,6 +871,11 @@ func (g *GossipSyncer) processChanRangeReply(msg *lnwire.ReplyChannelRange) erro
 			g.cfg.peerPub[:])
 
 		g.setSyncState(chansSynced)
+
+		// Ensure that the sync manager becomes aware that the
+		// historical sync completed so synced_to_graph is updated over
+		// rpc.
+		g.cfg.markGraphSynced()
 		return nil
 	}
 
@@ -1418,7 +1469,7 @@ func (g *GossipSyncer) handleSyncTransition(req *syncTransitionReq) error {
 	switch req.newSyncType {
 	// If an active sync has been requested, then we should resume receiving
 	// new graph updates from the remote peer.
-	case ActiveSync:
+	case ActiveSync, PinnedSync:
 		firstTimestamp = time.Now()
 		timestampRange = math.MaxUint32
 

discovery/syncer_test.go

@@ -189,6 +189,7 @@ func newTestSyncer(hID lnwire.ShortChannelID,
 		bestHeight: func() uint32 {
 			return latestKnownHeight
 		},
+		markGraphSynced:          func() {},
 		maxQueryChanRangeReplies: maxQueryChanRangeReplies,
 	}
 	syncer := newGossipSyncer(cfg)

lncfg/gossip.go

@@ -0,0 +1,28 @@
+package lncfg
+
+import (
+	"github.com/lightningnetwork/lnd/discovery"
+	"github.com/lightningnetwork/lnd/routing/route"
+)
+
+type Gossip struct {
+	PinnedSyncersRaw []string `long:"pinned-syncers" description:"A set of peers that should always remain in an active sync state, which can be used to closely synchronize the routing tables of two nodes. The value should be comma separated list of hex-encoded pubkeys. Connected peers matching this pubkey will remain active for the duration of the connection and not count towards the NumActiveSyncer count."`
+
+	PinnedSyncers discovery.PinnedSyncers
+}
+
+// Parse the pubkeys for the pinned syncers.
+func (g *Gossip) Parse() error {
+	pinnedSyncers := make(discovery.PinnedSyncers)
+	for _, pubkeyStr := range g.PinnedSyncersRaw {
+		vertex, err := route.NewVertexFromStr(pubkeyStr)
+		if err != nil {
+			return err
+		}
+		pinnedSyncers[vertex] = struct{}{}
+	}
+
+	g.PinnedSyncers = pinnedSyncers
+
+	return nil
+}

lnrpc/rpc.proto

@@ -1487,6 +1487,11 @@ message Peer {
         Denotes that we are not receiving new graph updates from the peer.
         */
         PASSIVE_SYNC = 2;
+
+        /*
+        Denotes that this peer is pinned into an active sync.
+        */
+        PINNED_SYNC = 3;
     }
 
     // The type of sync we are currently performing with this peer.

lnrpc/rpc.swagger.json

@@ -2269,10 +2269,11 @@
       "enum": [
         "UNKNOWN_SYNC",
         "ACTIVE_SYNC",
-        "PASSIVE_SYNC"
+        "PASSIVE_SYNC",
+        "PINNED_SYNC"
       ],
       "default": "UNKNOWN_SYNC",
-      "description": " - UNKNOWN_SYNC: Denotes that we cannot determine the peer's current sync type.\n - ACTIVE_SYNC: Denotes that we are actively receiving new graph updates from the peer.\n - PASSIVE_SYNC: Denotes that we are not receiving new graph updates from the peer."
+      "description": " - UNKNOWN_SYNC: Denotes that we cannot determine the peer's current sync type.\n - ACTIVE_SYNC: Denotes that we are actively receiving new graph updates from the peer.\n - PASSIVE_SYNC: Denotes that we are not receiving new graph updates from the peer.\n - PINNED_SYNC: Denotes that this peer is pinned into an active sync."
     },
     "PendingChannelsResponseClosedChannel": {
       "type": "object",

lntest/itest/lnd_test.go

@@ -10330,21 +10330,119 @@ func subscribeGraphNotifications(t *harnessTest, ctxb context.Context,
 	}
 }
 
+func assertSyncType(t *harnessTest, node *lntest.HarnessNode,
+	peer string, syncType lnrpc.Peer_SyncType) {
+
+	t.t.Helper()
+
+	ctxb := context.Background()
+	ctxt, _ := context.WithTimeout(ctxb, defaultTimeout)
+	resp, err := node.ListPeers(ctxt, &lnrpc.ListPeersRequest{})
+	require.NoError(t.t, err)
+
+	for _, rpcPeer := range resp.Peers {
+		if rpcPeer.PubKey != peer {
+			continue
+		}
+
+		require.Equal(t.t, syncType, rpcPeer.SyncType)
+		return
+	}
+
+	t.t.Fatalf("unable to find peer: %s", peer)
+}
+
+func waitForGraphSync(t *harnessTest, node *lntest.HarnessNode) {
+	t.t.Helper()
+
+	err := wait.Predicate(func() bool {
+		ctxb := context.Background()
+		ctxt, _ := context.WithTimeout(ctxb, defaultTimeout)
+		resp, err := node.GetInfo(ctxt, &lnrpc.GetInfoRequest{})
+		require.NoError(t.t, err)
+
+		return resp.SyncedToGraph
+	}, defaultTimeout)
+	require.NoError(t.t, err)
+}
+
 func testGraphTopologyNotifications(net *lntest.NetworkHarness, t *harnessTest) {
+	t.t.Run("pinned", func(t *testing.T) {
+		ht := newHarnessTest(t, net)
+		testGraphTopologyNtfns(net, ht, true)
+	})
+	t.t.Run("unpinned", func(t *testing.T) {
+		ht := newHarnessTest(t, net)
+		testGraphTopologyNtfns(net, ht, false)
+	})
+}
+
+func testGraphTopologyNtfns(net *lntest.NetworkHarness, t *harnessTest, pinned bool) {
 	ctxb := context.Background()
 
 	const chanAmt = funding.MaxBtcFundingAmount
 
+	// Spin up Bob first, since we will need to grab his pubkey when
+	// starting Alice to test pinned syncing.
+	bob, err := net.NewNode("bob", nil)
+	require.NoError(t.t, err)
+	defer shutdownAndAssert(net, t, bob)
+
+	ctxt, _ := context.WithTimeout(ctxb, defaultTimeout)
+	bobInfo, err := bob.GetInfo(ctxt, &lnrpc.GetInfoRequest{})
+	require.NoError(t.t, err)
+	bobPubkey := bobInfo.IdentityPubkey
+
+	// For unpinned syncing, start Alice as usual. Otherwise grab Bob's
+	// pubkey to include in his pinned syncer set.
+	var aliceArgs []string
+	if pinned {
+		aliceArgs = []string{
+			"--numgraphsyncpeers=1",
+			fmt.Sprintf("--gossip.pinned-syncers=%s", bobPubkey),
+		}
+	}
+
+	alice, err := net.NewNode("alice", aliceArgs)
+	require.NoError(t.t, err)
+	defer shutdownAndAssert(net, t, alice)
+
+	// Connect Alice and Bob.
+	ctxt, _ = context.WithTimeout(ctxb, defaultTimeout)
+	err = net.EnsureConnected(ctxt, alice, bob)
+	require.NoError(t.t, err)
+
+	// Alice stimmy.
+	ctxt, _ = context.WithTimeout(ctxb, defaultTimeout)
+	err = net.SendCoins(ctxt, btcutil.SatoshiPerBitcoin, alice)
+	require.NoError(t.t, err)
+
+	// Bob stimmy.
+	ctxt, _ = context.WithTimeout(ctxb, defaultTimeout)
+	err = net.SendCoins(ctxt, btcutil.SatoshiPerBitcoin, bob)
+	require.NoError(t.t, err)
+
+	// Assert that Bob has the correct sync type before proceeeding.
+	if pinned {
+		assertSyncType(t, alice, bobPubkey, lnrpc.Peer_PINNED_SYNC)
+	} else {
+		assertSyncType(t, alice, bobPubkey, lnrpc.Peer_ACTIVE_SYNC)
+	}
+
+	// Regardless of syncer type, ensure that both peers report having
+	// completed their initial sync before continuing to make a channel.
+	waitForGraphSync(t, alice)
+
 	// Let Alice subscribe to graph notifications.
 	graphSub := subscribeGraphNotifications(
-		t, ctxb, net.Alice,
+		t, ctxb, alice,
 	)
 	defer close(graphSub.quit)
 
 	// Open a new channel between Alice and Bob.
-	ctxt, _ := context.WithTimeout(ctxb, channelOpenTimeout)
+	ctxt, _ = context.WithTimeout(ctxb, channelOpenTimeout)
 	chanPoint := openChannelAndAssert(
-		ctxt, t, net, net.Alice, net.Bob,
+		ctxt, t, net, alice, bob,
 		lntest.OpenChannelParams{
 			Amt: chanAmt,
 		},
@@ -10364,15 +10462,15 @@ func testGraphTopologyNotifications(net *lntest.NetworkHarness, t *harnessTest)
 			// message.
 			for _, chanUpdate := range graphUpdate.ChannelUpdates {
 				switch chanUpdate.AdvertisingNode {
-				case net.Alice.PubKeyStr:
+				case alice.PubKeyStr:
-				case net.Bob.PubKeyStr:
+				case bob.PubKeyStr:
 				default:
 					t.Fatalf("unknown advertising node: %v",
 						chanUpdate.AdvertisingNode)
 				}
 				switch chanUpdate.ConnectingNode {
-				case net.Alice.PubKeyStr:
+				case alice.PubKeyStr:
-				case net.Bob.PubKeyStr:
+				case bob.PubKeyStr:
 				default:
 					t.Fatalf("unknown connecting node: %v",
 						chanUpdate.ConnectingNode)
@@ -10388,8 +10486,8 @@ func testGraphTopologyNotifications(net *lntest.NetworkHarness, t *harnessTest)
 
 			for _, nodeUpdate := range graphUpdate.NodeUpdates {
 				switch nodeUpdate.IdentityKey {
-				case net.Alice.PubKeyStr:
+				case alice.PubKeyStr:
-				case net.Bob.PubKeyStr:
+				case bob.PubKeyStr:
 				default:
 					t.Fatalf("unknown node: %v",
 						nodeUpdate.IdentityKey)
@@ -10413,7 +10511,7 @@ func testGraphTopologyNotifications(net *lntest.NetworkHarness, t *harnessTest)
 	// Now we'll test that updates are properly sent after channels are closed
 	// within the network.
 	ctxt, _ = context.WithTimeout(ctxb, channelCloseTimeout)
-	closeChannelAndAssert(ctxt, t, net, net.Alice, chanPoint, false)
+	closeChannelAndAssert(ctxt, t, net, alice, chanPoint, false)
 
 	// Now that the channel has been closed, we should receive a
 	// notification indicating so.
@@ -10468,7 +10566,7 @@ out:
 	// we disconnect Alice and Bob, open a channel between Bob and Carol,
 	// and finally connect Alice to Bob again.
 	ctxt, _ = context.WithTimeout(ctxb, defaultTimeout)
-	if err := net.DisconnectNodes(ctxt, net.Alice, net.Bob); err != nil {
+	if err := net.DisconnectNodes(ctxt, alice, bob); err != nil {
 		t.Fatalf("unable to disconnect alice and bob: %v", err)
 	}
 	carol, err := net.NewNode("Carol", nil)
@@ -10478,12 +10576,12 @@ out:
 	defer shutdownAndAssert(net, t, carol)
 
 	ctxt, _ = context.WithTimeout(ctxb, defaultTimeout)
-	if err := net.ConnectNodes(ctxt, net.Bob, carol); err != nil {
+	if err := net.ConnectNodes(ctxt, bob, carol); err != nil {
 		t.Fatalf("unable to connect bob to carol: %v", err)
 	}
 	ctxt, _ = context.WithTimeout(ctxb, channelOpenTimeout)
 	chanPoint = openChannelAndAssert(
-		ctxt, t, net, net.Bob, carol,
+		ctxt, t, net, bob, carol,
 		lntest.OpenChannelParams{
 			Amt: chanAmt,
 		},
@@ -10496,7 +10594,7 @@ out:
 	// Bob, since a node will update its node announcement after a new
 	// channel is opened.
 	ctxt, _ = context.WithTimeout(ctxb, defaultTimeout)
-	if err := net.EnsureConnected(ctxt, net.Alice, net.Bob); err != nil {
+	if err := net.EnsureConnected(ctxt, alice, bob); err != nil {
 		t.Fatalf("unable to connect alice to bob: %v", err)
 	}
 
@@ -10510,7 +10608,7 @@ out:
 			for _, nodeUpdate := range graphUpdate.NodeUpdates {
 				switch nodeUpdate.IdentityKey {
 				case carol.PubKeyStr:
-				case net.Bob.PubKeyStr:
+				case bob.PubKeyStr:
 				default:
 					t.Fatalf("unknown node update pubey: %v",
 						nodeUpdate.IdentityKey)
@@ -10521,14 +10619,14 @@ out:
 			for _, chanUpdate := range graphUpdate.ChannelUpdates {
 				switch chanUpdate.AdvertisingNode {
 				case carol.PubKeyStr:
-				case net.Bob.PubKeyStr:
+				case bob.PubKeyStr:
 				default:
 					t.Fatalf("unknown advertising node: %v",
 						chanUpdate.AdvertisingNode)
 				}
 				switch chanUpdate.ConnectingNode {
 				case carol.PubKeyStr:
-				case net.Bob.PubKeyStr:
+				case bob.PubKeyStr:
 				default:
 					t.Fatalf("unknown connecting node: %v",
 						chanUpdate.ConnectingNode)
@@ -10552,7 +10650,7 @@ out:
 
 	// Close the channel between Bob and Carol.
 	ctxt, _ = context.WithTimeout(ctxb, channelCloseTimeout)
-	closeChannelAndAssert(ctxt, t, net, net.Bob, chanPoint, false)
+	closeChannelAndAssert(ctxt, t, net, bob, chanPoint, false)
 }
 
 // testNodeAnnouncement ensures that when a node is started with one or more
@@ -11774,11 +11872,7 @@ func testSwitchOfflineDelivery(net *lntest.NetworkHarness, t *harnessTest) {
 
 	// Wait until all outstanding htlcs in the network have been settled.
 	err = wait.Predicate(func() bool {
-		predErr = assertNumActiveHtlcs(nodes, 0)
+		return assertNumActiveHtlcs(nodes, 0) == nil
-		if predErr != nil {
-			return false
-		}
-		return true
 	}, defaultTimeout)
 	if err != nil {
 		t.Fatalf("htlc mismatch: %v", predErr)
@@ -12102,11 +12196,7 @@ func testSwitchOfflineDeliveryPersistence(net *lntest.NetworkHarness, t *harness
 	// After reconnection succeeds, the settles should be propagated all
 	// the way back to the sender. All nodes should report no active htlcs.
 	err = wait.Predicate(func() bool {
-		predErr = assertNumActiveHtlcs(nodes, 0)
+		return assertNumActiveHtlcs(nodes, 0) == nil
-		if predErr != nil {
-			return false
-		}
-		return true
 	}, defaultTimeout)
 	if err != nil {
 		t.Fatalf("htlc mismatch: %v", predErr)
@@ -12369,11 +12459,7 @@ func testSwitchOfflineDeliveryOutgoingOffline(
 	// Wait for all payments to reach Carol.
 	var predErr error
 	err = wait.Predicate(func() bool {
-		predErr = assertNumActiveHtlcs(nodes, numPayments)
+		return assertNumActiveHtlcs(nodes, numPayments) == nil
-		if predErr != nil {
-			return false
-		}
-		return true
 	}, defaultTimeout)
 	if err != nil {
 		t.Fatalf("htlc mismatch: %v", predErr)

rpcserver.go

@@ -2746,6 +2746,8 @@ func (r *rpcServer) ListPeers(ctx context.Context,
 				lnrpcSyncType = lnrpc.Peer_ACTIVE_SYNC
 			case discovery.PassiveSync:
 				lnrpcSyncType = lnrpc.Peer_PASSIVE_SYNC
+			case discovery.PinnedSync:
+				lnrpcSyncType = lnrpc.Peer_PINNED_SYNC
 			default:
 				return nil, fmt.Errorf("unhandled sync type %v",
 					syncType)

sample-lnd.conf

@@ -1019,3 +1019,20 @@ litecoin.node=ltcd
 
 ; Specify the timeout to be used when opening the database.
 ; db.bolt.dbtimeout=60s
+
+[gossip]
+; Specify a set of pinned gossip syncers, which will always be actively syncing
+; whenever the corresponding peer is online. A pinned syncer does not count
+; towards the configured `numgraphsyncpeers` since pinned syncers are not
+; rotated. Configuring a pinned syncer does not ensure a persistent connection
+; to the target peer, they will only be pinned if the connection remains active
+; via some other mechanism, e.g. having an open channel.
+;
+; This feature is useful when trying to ensure that a node keeps its
+; routing table tightly synchronized with a set of remote peers, e.g. multiple
+; lightning nodes operated by the same service.
+;
+; Each value should be a hex-encoded pubkey of the pinned peer. Multiple pinned
+; peers can be specified by setting multiple flags/fields in the config.
+; gossip.pinned-syncers=pubkey1
+; gossip.pinned-syncers=pubkey2

server.go

@@ -820,6 +820,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 		SubBatchDelay:           time.Second * 5,
 		IgnoreHistoricalFilters: cfg.IgnoreHistoricalGossipFilters,
 		GossipUpdateThrottle:    !cfg.ProtocolOptions.NoGossipThrottle(),
+		PinnedSyncers:           cfg.Gossip.PinnedSyncers,
 	},
 		s.identityECDH.PubKey(),
 	)