Skip to content
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