retry params fetch

This commit is contained in:
tycho garen
2021-09-24 16:08:33 -04:00
parent 47cd6de82a
commit e3c7b263f1
3 changed files with 50 additions and 37 deletions
+1 -1
View File
@@ -1004,7 +1004,7 @@ func (r *Reactor) fetchLightBlock(height uint64) (*types.LightBlock, error) {
func (r *Reactor) waitForEnoughPeers(ctx context.Context, numPeers int) error {
startAt := time.Now()
t := time.NewTicker(200 * time.Millisecond)
t := time.NewTicker(100 * time.Millisecond)
defer t.Stop()
for r.peers.Len() < numPeers {
select {
+1 -1
View File
@@ -525,7 +525,7 @@ func TestReactor_StateProviderP2P(t *testing.T) {
rts.reactor.cfg.UseP2P = true
rts.reactor.cfg.TrustHeight = 1
rts.reactor.cfg.TrustHash = fmt.Sprintf("%X", chain[1].Hash())
ctx, cancel := context.WithCancel(context.Background())
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
rts.reactor.mtx.Lock()
+48 -35
View File
@@ -337,44 +337,57 @@ func (s *stateProviderP2P) addProvider(p lightprovider.Provider) {
}
}
// consensusParams sends out a request for consensus params blocking until one is returned.
// If it fails to get a valid set of consensus params from any of the providers it returns an error.
// consensusParams sends out a request for consensus params blocking
// until one is returned.
//
// If it fails to get a valid set of consensus params from any of the
// providers it returns an error; however, it will retry indefinitely
// (with backoff) until the context is canceled.
func (s *stateProviderP2P) consensusParams(ctx context.Context, height int64) (types.ConsensusParams, error) {
for _, provider := range s.lc.Witnesses() {
p, ok := provider.(*BlockProvider)
if !ok {
panic("expected p2p state provider to use p2p block providers")
}
// extract the nodeID of the provider
peer, err := types.NewNodeID(p.String())
if err != nil {
return types.ConsensusParams{}, fmt.Errorf("invalid provider (%s) node id: %w", p.String(), err)
}
select {
case s.paramsSendCh <- p2p.Envelope{
To: peer,
Message: &ssproto.ParamsRequest{
Height: uint64(height),
},
}:
case <-ctx.Done():
return types.ConsensusParams{}, ctx.Err()
}
select {
// if we get no response from this provider we move on to the next one
case <-time.After(consensusParamsResponseTimeout):
continue
case <-ctx.Done():
return types.ConsensusParams{}, ctx.Err()
case params, ok := <-s.paramsRecvCh:
iterCount := 0
for {
for _, provider := range s.lc.Witnesses() {
p, ok := provider.(*BlockProvider)
if !ok {
return types.ConsensusParams{}, errors.New("params channel closed")
panic("expected p2p state provider to use p2p block providers")
}
return params, nil
// extract the nodeID of the provider
peer, err := types.NewNodeID(p.String())
if err != nil {
return types.ConsensusParams{}, fmt.Errorf("invalid provider (%s) node id: %w", p.String(), err)
}
select {
case s.paramsSendCh <- p2p.Envelope{
To: peer,
Message: &ssproto.ParamsRequest{
Height: uint64(height),
},
}:
case <-ctx.Done():
return types.ConsensusParams{}, ctx.Err()
}
select {
// if we get no response from this provider we move on to the next one
case <-time.After(consensusParamsResponseTimeout):
continue
case <-ctx.Done():
return types.ConsensusParams{}, ctx.Err()
case params, ok := <-s.paramsRecvCh:
if !ok {
return types.ConsensusParams{}, errors.New("params channel closed")
}
return params, nil
}
}
iterCount++
select {
case <-ctx.Done():
return types.ConsensusParams{}, ctx.Err()
case <-time.After(time.Duration(iterCount) * consensusParamsResponseTimeout):
}
}
return types.ConsensusParams{}, errors.New("unable to fetch consensus params from connected providers")
}