diff --git a/internal/statesync/dispatcher.go b/internal/statesync/dispatcher.go index eff58c5f0..d70a8980b 100644 --- a/internal/statesync/dispatcher.go +++ b/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() diff --git a/internal/statesync/dispatcher_test.go b/internal/statesync/dispatcher_test.go index 6ada43f57..e94341ce4 100644 --- a/internal/statesync/dispatcher_test.go +++ b/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()) } diff --git a/internal/statesync/reactor.go b/internal/statesync/reactor.go index 2a6f6fa03..38e4b5c3a 100644 --- a/internal/statesync/reactor.go +++ b/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) diff --git a/internal/statesync/stateprovider.go b/internal/statesync/stateprovider.go index b361db4b4..7870b4e02 100644 --- a/internal/statesync/stateprovider.go +++ b/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 { diff --git a/internal/statesync/syncer.go b/internal/statesync/syncer.go index c1d0522be..6062e1617 100644 --- a/internal/statesync/syncer.go +++ b/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 diff --git a/light/client_test.go b/light/client_test.go index e8a478a53..e5078eda9 100644 --- a/light/client_test.go +++ b/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( diff --git a/node/node.go b/node/node.go index 53bb38bf0..b0b35aa31 100644 --- a/node/node.go +++ b/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) diff --git a/test/e2e/runner/setup.go b/test/e2e/runner/setup.go index 0604109e7..99f029e0d 100644 --- a/test/e2e/runner/setup.go +++ b/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 = ""