lnd.xprv/discovery/sync_manager_test.go

package discovery

import (
	"fmt"
	"math"
	"reflect"
	"sync/atomic"
	"testing"
	"time"

	"github.com/davecgh/go-spew/spew"
	"github.com/lightningnetwork/lnd/lntest"
	"github.com/lightningnetwork/lnd/lnwire"
	"github.com/lightningnetwork/lnd/ticker"
)

// randPeer creates a random peer.
func randPeer(t *testing.T, quit chan struct{}) *mockPeer {
	t.Helper()

	return &mockPeer{
		pk:       randPubKey(t),
		sentMsgs: make(chan lnwire.Message),
		quit:     quit,
	}
}

// newTestSyncManager creates a new test SyncManager using mock implementations
// of its dependencies.
func newTestSyncManager(numActiveSyncers int) *SyncManager {
	hID := lnwire.ShortChannelID{BlockHeight: latestKnownHeight}
	return newSyncManager(&SyncManagerCfg{
		ChanSeries:                newMockChannelGraphTimeSeries(hID),
		RotateTicker:              ticker.NewForce(DefaultSyncerRotationInterval),
		HistoricalSyncTicker:      ticker.NewForce(DefaultHistoricalSyncInterval),
		ActiveSyncerTimeoutTicker: ticker.NewForce(DefaultActiveSyncerTimeout),
		NumActiveSyncers:          numActiveSyncers,
	})
}

// TestSyncManagerNumActiveSyncers ensures that we are unable to have more than
// NumActiveSyncers active syncers.
func TestSyncManagerNumActiveSyncers(t *testing.T) {
	t.Parallel()

	// We'll start by creating our test sync manager which will hold up to
	// 3 active syncers.
	const numActiveSyncers = 3
	const numSyncers = numActiveSyncers + 1

	syncMgr := newTestSyncManager(numActiveSyncers)
	syncMgr.Start()
	defer syncMgr.Stop()

	// We'll go ahead and create our syncers. We'll gather the ones which
	// should be active and passive to check them later on.
	for i := 0; i < numActiveSyncers; i++ {
		peer := randPeer(t, syncMgr.quit)
		syncMgr.InitSyncState(peer)

		// The first syncer registered always attempts a historical
		// sync.
		if i == 0 {
			assertTransitionToChansSynced(t, syncMgr, peer, true)
		}

		assertPassiveSyncerTransition(t, syncMgr, peer)
		assertSyncerStatus(t, syncMgr, peer, chansSynced, ActiveSync)
	}

	for i := 0; i < numSyncers-numActiveSyncers; i++ {
		peer := randPeer(t, syncMgr.quit)
		syncMgr.InitSyncState(peer)
		assertSyncerStatus(t, syncMgr, peer, chansSynced, PassiveSync)
	}
}

// TestSyncManagerNewActiveSyncerAfterDisconnect ensures that we can regain an
// active syncer after losing one due to the peer disconnecting.
func TestSyncManagerNewActiveSyncerAfterDisconnect(t *testing.T) {
	t.Parallel()

	// We'll create our test sync manager to only have one active syncer.
	syncMgr := newTestSyncManager(1)
	syncMgr.Start()
	defer syncMgr.Stop()

	// peer1 will represent an active syncer that performs a historical
	// sync since it is the first registered peer with the SyncManager.
	peer1 := randPeer(t, syncMgr.quit)
	syncMgr.InitSyncState(peer1)
	assertTransitionToChansSynced(t, syncMgr, peer1, true)
	assertPassiveSyncerTransition(t, syncMgr, peer1)

	// It will then be torn down to simulate a disconnection. Since there
	// are no other candidate syncers available, the active syncer won't be
	// replaced.
	syncMgr.PruneSyncState(peer1.PubKey())

	// Then, we'll start our active syncer again, but this time we'll also
	// have a passive syncer available to replace the active syncer after
	// the peer disconnects.
	syncMgr.InitSyncState(peer1)
	assertPassiveSyncerTransition(t, syncMgr, peer1)

	// Create our second peer, which should be initialized as a passive
	// syncer.
	peer2 := randPeer(t, syncMgr.quit)
	syncMgr.InitSyncState(peer2)
	assertSyncerStatus(t, syncMgr, peer2, chansSynced, PassiveSync)

	// Disconnect our active syncer, which should trigger the SyncManager to
	// replace it with our passive syncer.
	syncMgr.PruneSyncState(peer1.PubKey())
	assertPassiveSyncerTransition(t, syncMgr, peer2)
}

// TestSyncManagerRotateActiveSyncerCandidate tests that we can successfully
// rotate our active syncers after a certain interval.
func TestSyncManagerRotateActiveSyncerCandidate(t *testing.T) {
	t.Parallel()

	// We'll create our sync manager with three active syncers.
	syncMgr := newTestSyncManager(1)
	syncMgr.Start()
	defer syncMgr.Stop()

	// The first syncer registered always performs a historical sync.
	activeSyncPeer := randPeer(t, syncMgr.quit)
	syncMgr.InitSyncState(activeSyncPeer)
	assertTransitionToChansSynced(t, syncMgr, activeSyncPeer, true)
	assertPassiveSyncerTransition(t, syncMgr, activeSyncPeer)

	// We'll send a tick to force a rotation. Since there aren't any
	// candidates, none of the active syncers will be rotated.
	syncMgr.cfg.RotateTicker.(*ticker.Force).Force <- time.Time{}
	assertNoMsgSent(t, activeSyncPeer)
	assertSyncerStatus(t, syncMgr, activeSyncPeer, chansSynced, ActiveSync)

	// We'll then go ahead and add a passive syncer.
	passiveSyncPeer := randPeer(t, syncMgr.quit)
	syncMgr.InitSyncState(passiveSyncPeer)
	assertSyncerStatus(t, syncMgr, passiveSyncPeer, chansSynced, PassiveSync)

	// We'll force another rotation - this time, since we have a passive
	// syncer available, they should be rotated.
	syncMgr.cfg.RotateTicker.(*ticker.Force).Force <- time.Time{}

	// The transition from an active syncer to a passive syncer causes the
	// peer to send out a new GossipTimestampRange in the past so that they
	// don't receive new graph updates.
	assertActiveSyncerTransition(t, syncMgr, activeSyncPeer)

	// The transition from a passive syncer to an active syncer causes the
	// peer to send a new GossipTimestampRange with the current timestamp to
	// signal that they would like to receive new graph updates from their
	// peers. This will also cause the gossip syncer to redo its state
	// machine, starting from its initial syncingChans state. We'll then
	// need to transition it to its final chansSynced state to ensure the
	// next syncer is properly started in the round-robin.
	assertPassiveSyncerTransition(t, syncMgr, passiveSyncPeer)
}

// TestSyncManagerHistoricalSync ensures that we only attempt a single
// historical sync during the SyncManager's startup, and that we can routinely
// force historical syncs whenever the HistoricalSyncTicker fires.
func TestSyncManagerHistoricalSync(t *testing.T) {
	t.Parallel()

	syncMgr := newTestSyncManager(0)
	syncMgr.Start()
	defer syncMgr.Stop()

	// We should expect to see a QueryChannelRange message with a
	// FirstBlockHeight of the genesis block, signaling that a historical
	// sync is being attempted.
	peer := randPeer(t, syncMgr.quit)
	syncMgr.InitSyncState(peer)
	assertMsgSent(t, peer, &lnwire.QueryChannelRange{
		FirstBlockHeight: 0,
		NumBlocks:        math.MaxUint32,
	})

	// If an additional peer connects, then a historical sync should not be
	// attempted again.
	extraPeer := randPeer(t, syncMgr.quit)
	syncMgr.InitSyncState(extraPeer)
	assertNoMsgSent(t, extraPeer)

	// Then, we'll send a tick to force a historical sync. This should
	// trigger the extra peer to also perform a historical sync since the
	// first peer is not eligible due to not being in a chansSynced state.
	syncMgr.cfg.HistoricalSyncTicker.(*ticker.Force).Force <- time.Time{}
	assertMsgSent(t, extraPeer, &lnwire.QueryChannelRange{
		FirstBlockHeight: 0,
		NumBlocks:        math.MaxUint32,
	})
}

// TestSyncManagerRoundRobinQueue ensures that any subsequent active syncers can
// only be started after the previous one has completed its state machine.
func TestSyncManagerRoundRobinQueue(t *testing.T) {
	t.Parallel()

	const numActiveSyncers = 3

	// We'll start by creating our sync manager with support for three
	// active syncers.
	syncMgr := newTestSyncManager(numActiveSyncers)
	syncMgr.Start()
	defer syncMgr.Stop()

	peers := make([]*mockPeer, 0, numActiveSyncers)

	// The first syncer registered always attempts a historical sync.
	firstPeer := randPeer(t, syncMgr.quit)
	syncMgr.InitSyncState(firstPeer)
	peers = append(peers, firstPeer)
	assertTransitionToChansSynced(t, syncMgr, firstPeer, true)

	// After completing the historical sync, a sync transition to ActiveSync
	// should happen. It should transition immediately since it has no
	// dependents.
	assertActiveGossipTimestampRange(t, firstPeer)

	// We'll create the remaining numActiveSyncers. These will be queued in
	// the round robin since the first syncer has yet to reach chansSynced.
	queuedPeers := make([]*mockPeer, 0, numActiveSyncers-1)
	for i := 0; i < numActiveSyncers-1; i++ {
		peer := randPeer(t, syncMgr.quit)
		syncMgr.InitSyncState(peer)
		peers = append(peers, peer)
		queuedPeers = append(queuedPeers, peer)
	}

	// Ensure they cannot transition without sending a GossipTimestampRange
	// message first.
	for _, peer := range queuedPeers {
		assertNoMsgSent(t, peer)
	}

	// Transition the first syncer to chansSynced, which should allow the
	// second to transition next.
	assertTransitionToChansSynced(t, syncMgr, firstPeer, false)

	// assertSyncerTransitioned ensures the target peer's syncer is the only
	// that has transitioned.
	assertSyncerTransitioned := func(target *mockPeer) {
		t.Helper()

		for _, peer := range peers {
			if peer.PubKey() != target.PubKey() {
				assertNoMsgSent(t, peer)
				continue
			}

			assertActiveGossipTimestampRange(t, target)
		}
	}

	// For each queued syncer, we'll ensure they have transitioned to an
	// ActiveSync type and reached their final chansSynced state to allow
	// the next one to transition.
	for _, peer := range queuedPeers {
		assertSyncerTransitioned(peer)
		assertTransitionToChansSynced(t, syncMgr, peer, false)
	}
}

// TestSyncManagerRoundRobinTimeout ensures that if we timeout while waiting for
// an active syncer to reach its final chansSynced state, then we will go on to
// start the next.
func TestSyncManagerRoundRobinTimeout(t *testing.T) {
	t.Parallel()

	// Create our sync manager with support for two active syncers.
	syncMgr := newTestSyncManager(2)
	syncMgr.Start()
	defer syncMgr.Stop()

	// peer1 will be the first peer we start, which will time out and cause
	// peer2 to start.
	peer1 := randPeer(t, syncMgr.quit)
	peer2 := randPeer(t, syncMgr.quit)

	// The first syncer registered always attempts a historical sync.
	syncMgr.InitSyncState(peer1)
	assertTransitionToChansSynced(t, syncMgr, peer1, true)

	// We assume the syncer for peer1 has transitioned once we see it send a
	// lnwire.GossipTimestampRange message.
	assertActiveGossipTimestampRange(t, peer1)

	// We'll then create the syncer for peer2. This should cause it to be
	// queued so that it starts once the syncer for peer1 is done.
	syncMgr.InitSyncState(peer2)
	assertNoMsgSent(t, peer2)

	// Send a force tick to pretend the sync manager has timed out waiting
	// for peer1's syncer to reach chansSynced.
	syncMgr.cfg.ActiveSyncerTimeoutTicker.(*ticker.Force).Force <- time.Time{}

	// Finally, ensure that the syncer for peer2 has transitioned.
	assertActiveGossipTimestampRange(t, peer2)
}

// TestSyncManagerRoundRobinStaleSyncer ensures that any stale active syncers we
// are currently waiting for or are queued up to start are properly removed and
// stopped.
func TestSyncManagerRoundRobinStaleSyncer(t *testing.T) {
	t.Parallel()

	const numActiveSyncers = 4

	// We'll create and start our sync manager with some active syncers.
	syncMgr := newTestSyncManager(numActiveSyncers)
	syncMgr.Start()
	defer syncMgr.Stop()

	peers := make([]*mockPeer, 0, numActiveSyncers)

	// The first syncer registered always attempts a historical sync.
	firstPeer := randPeer(t, syncMgr.quit)
	syncMgr.InitSyncState(firstPeer)
	peers = append(peers, firstPeer)
	assertTransitionToChansSynced(t, syncMgr, firstPeer, true)

	// After completing the historical sync, a sync transition to ActiveSync
	// should happen. It should transition immediately since it has no
	// dependents.
	assertActiveGossipTimestampRange(t, firstPeer)
	assertMsgSent(t, firstPeer, &lnwire.QueryChannelRange{
		FirstBlockHeight: startHeight,
		NumBlocks:        math.MaxUint32 - startHeight,
	})

	// We'll create the remaining numActiveSyncers. These will be queued in
	// the round robin since the first syncer has yet to reach chansSynced.
	queuedPeers := make([]*mockPeer, 0, numActiveSyncers-1)
	for i := 0; i < numActiveSyncers-1; i++ {
		peer := randPeer(t, syncMgr.quit)
		syncMgr.InitSyncState(peer)
		peers = append(peers, peer)
		queuedPeers = append(queuedPeers, peer)
	}

	// Ensure they cannot transition without sending a GossipTimestampRange
	// message first.
	for _, peer := range queuedPeers {
		assertNoMsgSent(t, peer)
	}

	// assertSyncerTransitioned ensures the target peer's syncer is the only
	// that has transitioned.
	assertSyncerTransitioned := func(target *mockPeer) {
		t.Helper()

		for _, peer := range peers {
			if peer.PubKey() != target.PubKey() {
				assertNoMsgSent(t, peer)
				continue
			}

			assertPassiveSyncerTransition(t, syncMgr, target)
		}
	}

	// We'll then remove the syncers in the middle to cover the case where
	// they are queued up in the sync manager's pending list.
	for i, peer := range peers {
		if i == 0 || i == len(peers)-1 {
			continue
		}

		syncMgr.PruneSyncState(peer.PubKey())
	}

	// We'll then remove the syncer we are currently waiting for. This
	// should prompt the last syncer to start since it is the only one left
	// pending. We'll do this in a goroutine since the peer behind the new
	// active syncer will need to send out its new GossipTimestampRange.
	go syncMgr.PruneSyncState(peers[0].PubKey())
	assertSyncerTransitioned(peers[len(peers)-1])
}

// assertNoMsgSent is a helper function that ensures a peer hasn't sent any
// messages.
func assertNoMsgSent(t *testing.T, peer *mockPeer) {
	t.Helper()

	select {
	case msg := <-peer.sentMsgs:
		t.Fatalf("peer %x sent unexpected message %v", peer.PubKey(),
			spew.Sdump(msg))
	case <-time.After(time.Second):
	}
}

// assertMsgSent asserts that the peer has sent the given message.
func assertMsgSent(t *testing.T, peer *mockPeer, msg lnwire.Message) {
	t.Helper()

	var msgSent lnwire.Message
	select {
	case msgSent = <-peer.sentMsgs:
	case <-time.After(time.Second):
		t.Fatalf("expected peer %x to send %T message", peer.PubKey(),
			msg)
	}

	if !reflect.DeepEqual(msgSent, msg) {
		t.Fatalf("expected peer %x to send message: %v\ngot: %v",
			peer.PubKey(), spew.Sdump(msg), spew.Sdump(msgSent))
	}
}

// assertActiveGossipTimestampRange is a helper function that ensures a peer has
// sent a lnwire.GossipTimestampRange message indicating that it would like to
// receive new graph updates.
func assertActiveGossipTimestampRange(t *testing.T, peer *mockPeer) {
	t.Helper()

	var msgSent lnwire.Message
	select {
	case msgSent = <-peer.sentMsgs:
	case <-time.After(time.Second):
		t.Fatalf("expected peer %x to send lnwire.GossipTimestampRange "+
			"message", peer.PubKey())
	}

	msg, ok := msgSent.(*lnwire.GossipTimestampRange)
	if !ok {
		t.Fatalf("expected peer %x to send %T message", peer.PubKey(),
			msg)
	}
	if msg.FirstTimestamp == 0 {
		t.Fatalf("expected *lnwire.GossipTimestampRange message with " +
			"non-zero FirstTimestamp")
	}
	if msg.TimestampRange == 0 {
		t.Fatalf("expected *lnwire.GossipTimestampRange message with " +
			"non-zero TimestampRange")
	}
}

// assertSyncerStatus asserts that the gossip syncer for the given peer matches
// the expected sync state and type.
func assertSyncerStatus(t *testing.T, syncMgr *SyncManager, peer *mockPeer,
	syncState syncerState, syncType SyncerType) {

	t.Helper()

	s, ok := syncMgr.GossipSyncer(peer.PubKey())
	if !ok {
		t.Fatalf("gossip syncer for peer %x not found", peer.PubKey())
	}

	// We'll check the status of our syncer within a WaitPredicate as some
	// sync transitions might cause this to be racy.
	err := lntest.WaitNoError(func() error {
		state := s.syncState()
		if s.syncState() != syncState {
			return fmt.Errorf("expected syncState %v for peer "+
				"%x, got %v", syncState, peer.PubKey(), state)
		}

		typ := s.SyncType()
		if s.SyncType() != syncType {
			return fmt.Errorf("expected syncType %v for peer "+
				"%x, got %v", syncType, peer.PubKey(), typ)
		}

		return nil
	}, time.Second)
	if err != nil {
		t.Fatal(err)
	}
}

// assertTransitionToChansSynced asserts the transition of an ActiveSync
// GossipSyncer to its final chansSynced state.
func assertTransitionToChansSynced(t *testing.T, syncMgr *SyncManager,
	peer *mockPeer, historicalSync bool) {

	t.Helper()

	s, ok := syncMgr.GossipSyncer(peer.PubKey())
	if !ok {
		t.Fatalf("gossip syncer for peer %x not found", peer.PubKey())
	}

	firstBlockHeight := uint32(startHeight)
	if historicalSync {
		firstBlockHeight = 0
	}
	assertMsgSent(t, peer, &lnwire.QueryChannelRange{
		FirstBlockHeight: firstBlockHeight,
		NumBlocks:        math.MaxUint32 - firstBlockHeight,
	})

	s.ProcessQueryMsg(&lnwire.ReplyChannelRange{Complete: 1}, nil)

	chanSeries := syncMgr.cfg.ChanSeries.(*mockChannelGraphTimeSeries)

	select {
	case <-chanSeries.filterReq:
		chanSeries.filterResp <- nil
	case <-time.After(2 * time.Second):
		t.Fatal("expected to receive FilterKnownChanIDs request")
	}

	err := lntest.WaitNoError(func() error {
		state := syncerState(atomic.LoadUint32(&s.state))
		if state != chansSynced {
			return fmt.Errorf("expected syncerState %v, got %v",
				chansSynced, state)
		}

		return nil
	}, time.Second)
	if err != nil {
		t.Fatal(err)
	}
}

// assertPassiveSyncerTransition asserts that a gossip syncer goes through all
// of its expected steps when transitioning from passive to active.
func assertPassiveSyncerTransition(t *testing.T, syncMgr *SyncManager,
	peer *mockPeer) {

	t.Helper()

	assertActiveGossipTimestampRange(t, peer)
	assertTransitionToChansSynced(t, syncMgr, peer, false)
}

// assertActiveSyncerTransition asserts that a gossip syncer goes through all of
// its expected steps when transitioning from active to passive.
func assertActiveSyncerTransition(t *testing.T, syncMgr *SyncManager,
	peer *mockPeer) {

	t.Helper()

	assertMsgSent(t, peer, &lnwire.GossipTimestampRange{
		FirstTimestamp: uint32(zeroTimestamp.Unix()),
		TimestampRange: 0,
	})
	assertSyncerStatus(t, syncMgr, peer, chansSynced, PassiveSync)
}