Merge pull request onflow#6011 from onflow/leo/increase-guarantee-buffer

[Execution] Increase guarantee buffer

engine/execution/ingestion/fetcher/access_fetcher.go

@@ -69,14 +69,23 @@ func NewAccessCollectionFetcher(
  return nil, fmt.Errorf("failed to connect to collection rpc: %w", err)
  }
 
+ lg.Info().Msgf("connected to access rpc at %s", accessAddress)
+
+ // make a large enough buffer so that it is able to hold all the guarantees
+ // on startup and not block the main thread.
+ // this case would only happen if there are lots of un-executed finalized blocks.
+ // making sure the --enable-new-ingestion-engine=true flag is on to make use
+ // of the new ingestion engine for catching up, which loads less un-executed blocks
+ // during startup.
+ bufferSize := 100_000
  noopHandler := func(flow.Identifier, flow.Entity) {}
  e := &AccessCollectionFetcher{
  log: lg,
  handler: noopHandler,
  client: access.NewAccessAPIClient(collectionRPCConn),
  chain: chain,
  originID: nodeID,
- guaranteeInfos: make(chan guaranteeInfo, 100),
+ guaranteeInfos: make(chan guaranteeInfo, bufferSize),
  }
 
  builder := component.NewComponentManagerBuilder().AddWorker(e.launchWorker)
@@ -98,6 +107,9 @@ func convertAccessAddrFromState(address string) string {
 }
 
 func (f *AccessCollectionFetcher) FetchCollection(blockID flow.Identifier, height uint64, guarantee *flow.CollectionGuarantee) error {
+ f.log.Debug().Hex("blockID", blockID[:]).Uint64("height", height).Hex("col_id", guarantee.CollectionID[:]).
+ Msgf("fetching collection guarantee")
+
  f.guaranteeInfos <- guaranteeInfo{
  blockID: blockID,
  height: height,
@@ -117,6 +129,8 @@ func (f *AccessCollectionFetcher) WithHandle(handler requester.HandleFunc) {
 func (f *AccessCollectionFetcher) launchWorker(ctx irrecoverable.SignalerContext, ready component.ReadyFunc) {
  ready()
 
+ f.log.Info().Msg("launching collection fetcher worker")
+
  for {
  select {
  case <-ctx.Done():