Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

TestRequestBlockBytesErrors: Various fixes #2928

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions catchup/universalFetcher.go
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,9 @@ func (uf *universalBlockFetcher) fetchBlock(ctx context.Context, round basics.Ro
config: &uf.config,
}
fetchedBuf, err = fetcherClient.getBlockBytes(ctx, round)
if err != nil {
return nil, nil, time.Duration(0), err
}
address = fetcherClient.address()
} else if httpPeer, validHTTPPeer := peer.(network.HTTPPeer); validHTTPPeer {
fetcherClient := &HTTPFetcher{
Expand All @@ -74,14 +77,14 @@ func (uf *universalBlockFetcher) fetchBlock(ctx context.Context, round basics.Ro
log: uf.log,
config: &uf.config}
fetchedBuf, err = fetcherClient.getBlockBytes(ctx, round)
if err != nil {
return nil, nil, time.Duration(0), err
}
address = fetcherClient.address()
} else {
return nil, nil, time.Duration(0), fmt.Errorf("fetchBlock: UniversalFetcher only supports HTTPPeer and UnicastPeer")
}
downloadDuration = time.Now().Sub(blockDownloadStartTime)
if err != nil {
return nil, nil, time.Duration(0), err
}
block, cert, err := processBlockBytes(fetchedBuf, round, address)
if err != nil {
return nil, nil, time.Duration(0), err
Expand Down
4 changes: 3 additions & 1 deletion catchup/universalFetcher_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,7 @@ func TestRequestBlockBytesErrors(t *testing.T) {
t.Fatal(err)
return
}
defer ledger.Ledger.Close()

blockServiceConfig := config.GetDefaultLocal()
blockServiceConfig.EnableBlockService = true
Expand All @@ -191,14 +192,15 @@ func TestRequestBlockBytesErrors(t *testing.T) {
up := makeTestUnicastPeer(net, t)
ls := rpcs.MakeBlockService(logging.Base(), blockServiceConfig, ledger, net, "test genesisID")
ls.Start()
defer ls.Stop()

fetcher := makeUniversalBlockFetcher(logging.TestingLog(t), net, cfg)

ctx, cancel := context.WithCancel(context.Background())
cancel()
_, _, _, err = fetcher.fetchBlock(ctx, next, up)
var wrfe errWsFetcherRequestFailed
require.True(t, errors.As(err, &wrfe))
require.True(t, errors.As(err, &wrfe), "unexpected err: %w", wrfe)
require.Equal(t, "context canceled", err.(errWsFetcherRequestFailed).cause)

ctx = context.Background()
Expand Down
5 changes: 5 additions & 0 deletions rpcs/blockService.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"path"
"strconv"
"strings"
"sync"

"github.com/gorilla/mux"

Expand Down Expand Up @@ -72,6 +73,7 @@ type BlockService struct {
fallbackEndpoints fallbackEndpoints
enableArchiverFallback bool
log logging.Logger
closeWaitGroup sync.WaitGroup
}

// EncodedBlockCert defines how GetBlockBytes encodes a block and its certificate
Expand Down Expand Up @@ -125,12 +127,14 @@ func (bs *BlockService) Start() {
bs.net.RegisterHandlers(handlers)
}
bs.stop = make(chan struct{})
bs.closeWaitGroup.Add(1)
go bs.listenForCatchupReq(bs.catchupReqs, bs.stop)
}

// Stop servicing catchup requests over ws
func (bs *BlockService) Stop() {
close(bs.stop)
bs.closeWaitGroup.Wait()
}

// ServerHTTP returns blocks
Expand Down Expand Up @@ -237,6 +241,7 @@ func (bs *BlockService) processIncomingMessage(msg network.IncomingMessage) (n n

// listenForCatchupReq handles catchup getblock request
func (bs *BlockService) listenForCatchupReq(reqs <-chan network.IncomingMessage, stop chan struct{}) {
defer bs.closeWaitGroup.Done()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
for {
Expand Down