changes while looking for issue

internal/statesync/dispatcher.go

@@ -3,6 +3,7 @@ package statesync
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -58,6 +59,7 @@ func (d *dispatcher) LightBlock(ctx context.Context, height int64) (*types.Light
 	d.mtx.Lock()
 	// check that the dispatcher is connected to the reactor
 	if !d.running {
+		d.mtx.Unlock()
 		return nil, "", errDisconnected
 	}
 	// check to see that the dispatcher is connected to at least one peer
@@ -70,6 +72,7 @@ func (d *dispatcher) LightBlock(ctx context.Context, height int64) (*types.Light
 	// fetch the next peer id in the list and request a light block from that
 	// peer
 	peer := d.availablePeers.Pop(ctx)
+
 	lb, err := d.lightBlock(ctx, height, peer)
 
 	// append the peer back to the list
@@ -109,8 +112,8 @@ func (d *dispatcher) stop() {
 	defer d.mtx.Unlock()
 	d.running = false
 	for peer, call := range d.calls {
-		close(call)
 		delete(d.calls, peer)
+		close(call)
 	}
 }
 
@@ -125,14 +128,15 @@ func (d *dispatcher) lightBlock(ctx context.Context, height int64, peer types.No
 		d.mtx.Lock()
 		defer d.mtx.Unlock()
 		if call, ok := d.calls[peer]; ok {
-			close(call)
 			delete(d.calls, peer)
+			close(call)
 		}
 	}()
 
 	// wait for a response, cancel or timeout
 	select {
 	case resp := <-callCh:
+		fmt.Printf("received response, height %d peer %v\n", height, peer)
 		return resp, nil
 
 	case <-ctx.Done():
@@ -146,8 +150,10 @@ func (d *dispatcher) lightBlock(ctx context.Context, height int64, peer types.No
 // respond allows the underlying process which receives requests on the
 // requestCh to respond with the respective light block
 func (d *dispatcher) respond(lb *proto.LightBlock, peer types.NodeID) error {
+	fmt.Printf("trying to respond with light block for height %d from %v\n", lb.SignedHeader.Header.Height, peer)
 	d.mtx.Lock()
 	defer d.mtx.Unlock()
+	fmt.Printf("responding with light block for height %d from %v\n", lb.SignedHeader.Header.Height, peer)
 
 	// check that the response came from a request
 	answerCh, ok := d.calls[peer]
@@ -223,12 +229,14 @@ func (d *dispatcher) dispatch(peer types.NodeID, height int64) (chan *types.Ligh
 	d.calls[peer] = ch
 
 	// send request
+	fmt.Printf("sending request dispatch, height %d peer %v\n", height, peer)
 	d.requestCh <- p2p.Envelope{
 		To: peer,
 		Message: &ssproto.LightBlockRequest{
 			Height: uint64(height),
 		},
 	}
+	fmt.Printf("sent request dispatch, height %d peer %v\n", height, peer)
 	return ch, nil
 }
 
@@ -251,6 +259,7 @@ func (p *blockProvider) LightBlock(ctx context.Context, height int64) (*types.Li
 	if !p.dispatcher.isConnected(p.peer) {
 		return nil, provider.ErrConnectionClosed
 	}
+	fmt.Println("fetching block for block provider")
 
 	ctx, cancel := context.WithTimeout(ctx, p.timeout)
 	defer cancel()

internal/statesync/dispatcher_test.go

@@ -75,6 +75,35 @@ func TestDispatcherReturnsNoBlock(t *testing.T) {
 	require.Equal(t, peerFromSet, peerResult)
 }
 
+func TestBlockProviderTimeOutWaitingOnLightBlock(t *testing.T) {
+	t.Cleanup(leaktest.Check(t))
+	ch := make(chan p2p.Envelope, 100)
+	d := newDispatcher(ch, 1*time.Second)
+	peerFromSet := createPeerSet(1)[0]
+	d.addPeer(peerFromSet)
+	p := d.CreateProvider(peerFromSet, "test-chain")
+	lb, err := p.LightBlock(context.Background(), 1)
+	require.NoError(t, err)
+	require.NotNil(t, lb)
+}
+
+func TestDispatcherTimeOutWaitingOnLightBlock(t *testing.T) {
+	t.Cleanup(leaktest.Check(t))
+	ch := make(chan p2p.Envelope, 100)
+	d := newDispatcher(ch, 1*time.Second)
+	peerFromSet := createPeerSet(1)[0]
+	d.addPeer(peerFromSet)
+	ctx, cancelFunc := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancelFunc()
+
+	lb, peerResult, err := d.LightBlock(ctx, 1)
+
+	require.Error(t, err)
+	require.Equal(t, context.DeadlineExceeded, err)
+	require.Nil(t, lb)
+	require.Equal(t, peerFromSet, peerResult)
+}
+
 func TestDispatcherErrorsWhenNoPeers(t *testing.T) {
 	t.Cleanup(leaktest.Check(t))
 	ch := make(chan p2p.Envelope, 100)
@@ -133,7 +162,7 @@ func TestDispatcherProviders(t *testing.T) {
 	t.Cleanup(leaktest.Check(t))
 
 	ch := make(chan p2p.Envelope, 100)
-	chainID := "state-sync-test"
+	chainID := "test-chain"
 	closeCh := make(chan struct{})
 	defer close(closeCh)
 
@@ -152,8 +181,8 @@ func TestDispatcherProviders(t *testing.T) {
 		require.True(t, ok)
 		assert.Equal(t, string(peers[i]), bp.String(), i)
 		lb, err := p.LightBlock(context.Background(), 10)
-		assert.Error(t, err)
-		assert.Nil(t, lb)
+		assert.NoError(t, err)
+		assert.NotNil(t, lb)
 	}
 	require.Equal(t, 0, d.peerCount())
 }

internal/statesync/reactor.go

@@ -62,7 +62,7 @@ var (
 			MsgType: new(ssproto.Message),
 			Descriptor: &p2p.ChannelDescriptor{
 				ID:                  byte(LightBlockChannel),
-				Priority:            2,
+				Priority:            1,
 				SendQueueCapacity:   10,
 				RecvMessageCapacity: lightBlockMsgSize,
 				RecvBufferCapacity:  128,
@@ -302,6 +302,7 @@ func (r *Reactor) Sync(
 		}
 	}
 
+	r.Logger.Info("sync any starting")
 	state, commit, err := r.syncer.SyncAny(ctx, r.cfg.DiscoveryTime, requestSnapshotsHook)
 	if err != nil {
 		return sm.State{}, err
@@ -560,6 +561,9 @@ func (r *Reactor) handleSnapshotMessage(envelope p2p.Envelope) error {
 			)
 			return nil
 		}
+		if msg.Height == 3 {
+			fmt.Println("received snapshot for height 3")
+		}
 
 	default:
 		return fmt.Errorf("received unknown message: %T", msg)
@@ -667,6 +671,7 @@ func (r *Reactor) handleLightBlockMessage(envelope p2p.Envelope) error {
 			r.Logger.Error("failed to retrieve light block", "err", err, "height", msg.Height)
 			return err
 		}
+		r.Logger.Info("fetched light block", "height", lb.SignedHeader.Header.Height)
 
 		lbproto, err := lb.ToProto()
 		if err != nil {
@@ -682,6 +687,7 @@ func (r *Reactor) handleLightBlockMessage(envelope p2p.Envelope) error {
 				LightBlock: lbproto,
 			},
 		}
+		r.Logger.Info("sent light block response", "height", lb.SignedHeader.Header.Height)
 
 	case *ssproto.LightBlockResponse:
 		r.Logger.Info("received light block response")
@@ -709,7 +715,7 @@ func (r *Reactor) handleParamsMessage(envelope p2p.Envelope) error {
 		}
 
 		cpproto := cp.ToProto()
-		r.blockCh.Out <- p2p.Envelope{
+		r.paramsCh.Out <- p2p.Envelope{
 			To: envelope.From,
 			Message: &ssproto.ParamsResponse{
 				Height:          msg.Height,
@@ -813,6 +819,9 @@ func (r *Reactor) processCh(ch *p2p.Channel, chName string) {
 	for {
 		select {
 		case envelope := <-ch.In:
+			if chName == "light block" {
+				fmt.Println("received p2p message for light block")
+			}
 			if err := r.handleMessage(ch.ID, envelope); err != nil {
 				r.Logger.Error(fmt.Sprintf("failed to process %s message", chName),
 					"ch_id", ch.ID, "envelope", envelope, "err", err)

internal/statesync/stateprovider.go

@@ -223,6 +223,8 @@ func NewP2PStateProvider(
 		return nil, fmt.Errorf("at least 2 peers are required, got %d", len(providers))
 	}
 
+	logger.Info(fmt.Sprintf("providers list is %d long", len(providers[1:])))
+
 	lc, err := light.NewClient(ctx, chainID, trustOptions, providers[0], providers[1:],
 		lightdb.New(dbm.NewMemDB()), light.Logger(logger))
 	if err != nil {

internal/statesync/syncer.go

@@ -156,6 +156,7 @@ func (s *syncer) SyncAny(
 	discoveryTime time.Duration,
 	requestSnapshots func(),
 ) (sm.State, *types.Commit, error) {
+	s.logger.Info("in sync any")
 
 	if discoveryTime != 0 && discoveryTime < minimumDiscoveryTime {
 		discoveryTime = minimumDiscoveryTime
@@ -197,7 +198,9 @@ func (s *syncer) SyncAny(
 			defer chunks.Close() // in case we forget to close it elsewhere
 		}
 
+		s.logger.Info("starting sync")
 		newState, commit, err := s.Sync(ctx, snapshot, chunks)
+		s.logger.Info("after sync")
 		switch {
 		case err == nil:
 			return newState, commit, nil

light/client_test.go

@@ -406,6 +406,10 @@ func TestClientLargeBisectionVerification(t *testing.T) {
 	mockNode.AssertExpectations(t)
 }
 
+func TestHeightThree(t *testing.T) {
+	primary.LightBlock(context.Background(), 3)
+}
+
 func TestClientBisectionBetweenTrustedHeaders(t *testing.T) {
 	mockFullNode := mockNodeFromHeadersAndVals(headerSet, valSet)
 	c, err := light.NewClient(

node/node.go

@@ -680,6 +680,7 @@ func (n *nodeImpl) OnStart() error {
 		// FIXME: We shouldn't allow state sync to silently error out without
 		// bubbling up the error and gracefully shutting down the rest of the node
 		go func() {
+			n.Logger.Info("staring state sync")
 			state, err := n.stateSyncReactor.Sync(context.TODO(), state.ChainID, state.InitialHeight)
 			if err != nil {
 				n.Logger.Error("state sync failed", "err", err)

test/e2e/runner/setup.go

@@ -305,18 +305,20 @@ func MakeConfig(node *e2e.Node) (*config.Config, error) {
 	if node.StateSync {
 		cfg.StateSync.Enable = true
 		cfg.StateSync.UseP2P = true
-		// cfg.StateSync.RPCServers = []string{}
+		/*
+			cfg.StateSync.RPCServers = []string{}
 
-		// for _, peer := range node.Testnet.ArchiveNodes() {
-		// 	if peer.Name == node.Name {
-		// 		continue
-		// 	}
-		// 	cfg.StateSync.RPCServers = append(cfg.StateSync.RPCServers, peer.AddressRPC())
-		// }
+			for _, peer := range node.Testnet.ArchiveNodes() {
+				if peer.Name == node.Name {
+					continue
+				}
+				cfg.StateSync.RPCServers = append(cfg.StateSync.RPCServers, peer.AddressRPC())
+			}
 
-		// if len(cfg.StateSync.RPCServers) < 2 {
-		// 	return nil, errors.New("unable to find 2 suitable state sync RPC servers")
-		// }
+			if len(cfg.StateSync.RPCServers) < 2 {
+				return nil, errors.New("unable to find 2 suitable state sync RPC servers")
+			}
+		*/
 	}
 
 	cfg.P2P.Seeds = ""