Feature/slowdownload (#683)

* tweak debugSubstreamsRunDelay to max 500ms
* when download or execution is too slow for live, gracefully downgrade to a merged-block-based stream
* fix test after zstd upgrade makes it more efficient
* add to logs the 'client_read_average_time_last_5_minutes' to see if they are the cause of the wait time
* tweak changelog

Author: sduchesneau

docs/release-notes/change-log.md

@@ -13,9 +13,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ### Server
 
-* Add a maximum execution time for a full tier2 segment. By default, this is 60 minutes. It will fail with `rpc error: code = DeadlineExceeded desc = request active for too long`.
+* **BREAKING** Add a maximum execution time for a full tier2 segment. By default, this is 60 minutes. It will fail with `rpc error: code = DeadlineExceeded desc = request active for too long`.
   It can be configured from the `SegmentExecutionTimeout` configuration option on Tier2Config or disabled by setting it to 0.
 * Improve log message for 'request active for a long time', adding stats.
+* Fix `subscription channel at max capacity` error: when the LIVE channel is full (ex: slow module execution or slow client reader), the request will be continued from merged files instead of failing, and gracefully recover if performance is restored.
 * Fixed a small context memory leak when using wasmtime (especially with grpc-based metering plugin)
 
 ### CLI

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/jhump/protoreflect v1.14.0
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
-	github.com/streamingfast/bstream v0.0.2-0.20250221181559-fb0809660f91
+	github.com/streamingfast/bstream v0.0.2-0.20250903174843-9c884c3356fd
 	github.com/streamingfast/cli v0.0.4-0.20241119021815-815afa473375
 	github.com/streamingfast/dauth v0.0.0-20250821195214-8e2a3c300f97
 	github.com/streamingfast/dbin v0.9.1-0.20231117225723-59790c798e2c
@@ -106,6 +106,7 @@ require (
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/manifoldco/promptui v0.9.0 // indirect
 	github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect
+	github.com/mostynb/go-grpc-compression v1.2.3 // indirect
 	github.com/mschoch/smat v0.2.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.6 // indirect
 	github.com/pinax-network/graph-networks-libs/packages/golang v0.7.0 // indirect
@@ -163,7 +164,7 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/itchyny/timefmt-go v0.1.5 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
-	github.com/klauspost/compress v1.16.7
+	github.com/klauspost/compress v1.17.8
 	github.com/logrusorgru/aurora v2.0.3+incompatible // indirect
 	github.com/lucasb-eyer/go-colorful v1.2.0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect

go.sum

@@ -360,8 +360,8 @@ github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/X
 github.com/juju/ansiterm v0.0.0-20180109212912-720a0952cc2a/go.mod h1:UJSiEoRfvx3hP73CvoARgeLjaIOjybY9vj8PUPPFGeU=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGCFR9I=
-github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
+github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
@@ -411,6 +411,8 @@ github.com/mitchellh/hashstructure/v2 v2.0.2 h1:vGKWl0YJqUNxE8d+h8f6NJLcCJrgbhC4
 github.com/mitchellh/hashstructure/v2 v2.0.2/go.mod h1:MG3aRVU/N29oo/V/IhBX8GR/zz4kQkprJgF2EVszyDE=
 github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
 github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
+github.com/mostynb/go-grpc-compression v1.2.3 h1:42/BKWMy0KEJGSdWvzqIyOZ95YcR9mLPqKctH7Uo//I=
+github.com/mostynb/go-grpc-compression v1.2.3/go.mod h1:AghIxF3P57umzqM9yz795+y1Vjs47Km/Y2FE6ouQ7Lg=
 github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
 github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/mschoch/smat v0.2.0 h1:8imxQsjDm8yFEAVBe7azKmKSgzSkZXDuKkSq9374khM=
@@ -460,8 +462,8 @@ github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJ
 github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
 github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
-github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
-github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
+github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
+github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/rs/cors v1.10.0 h1:62NOS1h+r8p1mW6FM0FSB0exioXLhd/sh15KpjWBZ+8=
 github.com/rs/cors v1.10.0/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -490,8 +492,8 @@ github.com/spf13/viper v1.15.0 h1:js3yy885G8xwJa6iOISGFwd+qlUo5AvyXb7CiihdtiU=
 github.com/spf13/viper v1.15.0/go.mod h1:fFcTBJxvhhzSJiZy8n+PeW6t8l+KeT/uTARa0jHOQLA=
 github.com/spiffe/go-spiffe/v2 v2.5.0 h1:N2I01KCUkv1FAjZXJMwh95KK1ZIQLYbPfhaxw8WS0hE=
 github.com/spiffe/go-spiffe/v2 v2.5.0/go.mod h1:P+NxobPc6wXhVtINNtFjNWGBTreew1GBUCwT2wPmb7g=
-github.com/streamingfast/bstream v0.0.2-0.20250221181559-fb0809660f91 h1:n4Ws5vS8dF4J6WYP7mE0C9DsE1Inju7BKd6XWI4T18c=
-github.com/streamingfast/bstream v0.0.2-0.20250221181559-fb0809660f91/go.mod h1:n5wy+Vmwp4xbjXO7B81MAkAgjnf1vJ/lI2y6hWWyFbg=
+github.com/streamingfast/bstream v0.0.2-0.20250903174843-9c884c3356fd h1:CXp4kPAfdGyMJlitOipGLp0piCV+IRw9sIwzE4p0aU8=
+github.com/streamingfast/bstream v0.0.2-0.20250903174843-9c884c3356fd/go.mod h1:YXtnOZbqcqU4fzGqdi8B7rMcp2iQePQGf4oTIPBnzgc=
 github.com/streamingfast/cli v0.0.4-0.20241119021815-815afa473375 h1:nwuFSEJtQfqTuN62WvysfAtDT4qqwQ6ghFX0i2VY1fY=
 github.com/streamingfast/cli v0.0.4-0.20241119021815-815afa473375/go.mod h1:qOksW3DPhHVYBo8dcYxS7K3Q09wlcOChSdopeOjLWng=
 github.com/streamingfast/dauth v0.0.0-20250821195214-8e2a3c300f97 h1:Lcack+Yv9KWBeKfRrjQ+KUegwHbh6/aFLJuDyYzcXZ4=

metering/metering_test.go

@@ -70,7 +70,7 @@ func TestWithBytesReadMeteringOptionsZstd(t *testing.T) {
 	}
 
 	uncompressedSize := 1024
-	compressedSize := 24
+	compressedSize := 16
 
 	err = store.WriteObject(nil, "test", bytes.NewReader(bytes.Repeat([]byte("1"), uncompressedSize)))
 	if err != nil {

metrics/stats.go

@@ -57,8 +57,9 @@ type Stats struct {
 	// counter is used to get the next jobIdx
 	counter uint64
 
-	error  error
-	logger *zap.Logger
+	clientReadTime *dmetrics.AvgDurationCounter
+	error          error
+	logger         *zap.Logger
 }
 
 type runningJobs map[uint64]*extendedJob
@@ -157,6 +158,7 @@ func NewReqStats(config *Config, logger *zap.Logger) *Stats {
 	return &Stats{
 		config:             config,
 		blockRate:          dmetrics.MustNewAvgRateCounter(1*time.Second, 30*time.Second, "blocks"),
+		clientReadTime:     dmetrics.NewAvgDurationCounter(5*time.Minute, time.Second, "client_read_time"),
 		startTime:          time.Now(),
 		logger:             logger,
 		modulesStats:       make(map[string]*extendedStats),
@@ -462,6 +464,12 @@ func (s *Stats) RecordModuleWasmStoreDeletePrefix(moduleName string, sizeBytes u
 	mod.storeOperationTime += elapsed
 }
 
+func (s *Stats) RecordReadTime(since time.Time) {
+	s.Lock()
+	defer s.Unlock()
+	s.clientReadTime.AddElapsedTime(since)
+}
+
 func (s *Stats) RecordBlock(ref bstream.BlockRef) {
 	s.Lock()
 	defer s.Unlock()
@@ -726,6 +734,7 @@ func (s *Stats) getZapFields(meter dmetering.Meter) []zap.Field {
 		zap.Uint64("remote_blocks_processed", s.remoteProcessedBlockCount), // "estimated" from remote ranges
 		zap.Uint64("total_blocks_processed", s.processedBlocks),            // includes remote and local blocks processed in this request, multiplied by execution stages, excludes blocks that were skipped from indexes
 		zap.Uint64("uncompressed_egress_bytes", s.uncompressedEgressBytes),
+		zap.Duration("client_read_average_time_last_5_minutes", s.clientReadTime.Average()),
 		zap.String("error", errorText),
 	}
 

pipeline/pipeline.go

@@ -90,6 +90,7 @@ type Pipeline struct {
 	// (for chains with potential block skips)
 	lastFinalClock        *pbsubstreams.Clock
 	lastProcessedBlockRef bstream.BlockRef
+	lastCursor            *bstream.Cursor
 	sentBlocks            uint64
 
 	blockStepMap         map[bstream.StepType]uint64
@@ -293,6 +294,10 @@ func (p *Pipeline) GetStoreMap() store.Map {
 	return p.stores.StoreMap
 }
 
+func (p *Pipeline) LastCursor() *bstream.Cursor {
+	return p.lastCursor
+}
+
 func (p *Pipeline) setupProcessingModule(reqDetails *reqctx.RequestDetails) {
 	for _, module := range reqDetails.Modules.Modules {
 		if reqDetails.IsOutputModule(module.Name) {

pipeline/process_block.go

@@ -197,6 +197,7 @@ func (p *Pipeline) handleStepUndo(clock *pbsubstreams.Clock, cursor *bstream.Cur
 	targetClock := blockRefToPB(reorgJunctionBlock)
 
 	p.lastProcessedBlockRef = reorgJunctionBlock
+	p.lastCursor = targetCursor
 	return p.respFunc(
 		&pbsubstreamsrpc.Response{
 			Message: &pbsubstreamsrpc.Response_BlockUndoSignal{
@@ -308,6 +309,7 @@ func (p *Pipeline) handleStepNew(ctx context.Context, clock *pbsubstreams.Clock,
 	p.stores.resetStores()
 	logger.Debug("block processed", zap.Uint64("block_num", clock.Number))
 	p.lastProcessedBlockRef = bstream.NewBlockRef(clock.Id, clock.Number)
+	p.lastCursor = cursor
 	return nil
 }
 

service/live_back_filler.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 
@@ -71,6 +72,9 @@ func RequestBackProcessing(ctx context.Context, logger *zap.Logger, startBlock u
 		err := requestBackProcessing(ctx, logger, liveBackFillerRequest, clientFactory)
 		if err != nil {
 			logger.Debug("retryable error while live backprocessing", zap.Error(err))
+			if errors.Is(err, context.Canceled) {
+				return derr.NewFatalError(err)
+			}
 			return err
 		}
 

service/tier1.go

@@ -770,7 +770,7 @@ func (s *Tier1Service) blocks(ctx context.Context, request *pbsubstreamsrpc.Requ
 		zap.String("cursor", cursor),
 	)
 
-	var streamHandler bstream.Handler
+	var wrappedPipe bstream.Handler
 	if requestDetails.ProductionMode {
 		liveBackFiller := NewLiveBackFiller(ctx, pipe, logger, execGraph.OutputModuleStageIndex(), segmentSize, requestDetails.LinearHandoffBlockNum, s.runtimeConfig.ClientFactory, RequestBackProcessing)
 
@@ -781,14 +781,14 @@ func (s *Tier1Service) blocks(ctx context.Context, request *pbsubstreamsrpc.Requ
 		}
 
 		go liveBackFiller.Start(ctx)
-		streamHandler = liveBackFiller
+		wrappedPipe = liveBackFiller
 	} else {
-		streamHandler = pipe
+		wrappedPipe = pipe
 	}
 
 	blockStream, err := s.streamFactoryFunc(
 		ctx,
-		streamHandler,
+		wrappedPipe,
 		int64(requestDetails.LinearHandoffBlockNum),
 		request.StopBlockNum,
 		cursor,
@@ -803,7 +803,37 @@ func (s *Tier1Service) blocks(ctx context.Context, request *pbsubstreamsrpc.Requ
 	}
 
 	ctx, span := reqctx.WithSpan(ctx, "substreams/tier1/pipeline/blocks_stream")
-	streamErr = blockStream.Run(ctx)
+	for {
+		streamErr = blockStream.Run(ctx)
+		if errors.Is(streamErr, hub.ErrSubscriptionChannelFull) {
+			cur := pipe.LastCursor()
+			if cur == nil {
+				logger.Warn("subscription channel at max capacity, but no cursor was found to reconnect")
+				break
+			}
+
+			logger.Warn("subscription channel at max capacity, creating new stream", zap.String("last_block", cur.Block.String()))
+			blockStream, err = s.streamFactoryFunc(
+				ctx,
+				wrappedPipe,
+				int64(requestDetails.LinearHandoffBlockNum),
+				request.StopBlockNum,
+				cur.ToOpaque(),
+				request.FinalBlocksOnly,
+				false,
+				logger.Named("stream"),
+				bsstream.WithLiveSourceHandlerMiddleware(metering.LiveSourceMiddlewareHandlerFactory(ctx)),
+				bsstream.WithFileSourceHandlerMiddleware(metering.FileSourceMiddlewareHandlerFactory(ctx)),
+			)
+			if err != nil {
+				streamErr = fmt.Errorf("error getting stream: %w", err)
+				break
+			}
+			continue
+		}
+		break
+	}
+
 	span.EndWithErr(&streamErr)
 
 	return pipe.OnStreamTerminated(ctx, streamErr)
@@ -868,10 +898,12 @@ func tier1ResponseHandler(ctx context.Context, mut *sync.Mutex, logger *zap.Logg
 			}
 		}
 
+		begin := time.Now()
 		if err := streamSrv.Send(resp); err != nil {
 			logger.Info("unable to send block probably due to client disconnecting", zap.String("user_id", userID), zap.String("api_key_id", apiKeyID), zap.Error(err))
 			return connect.NewError(connect.CodeUnavailable, err)
 		}
+		stats.RecordReadTime(begin)
 
 		if isData {
 			stats.RecordDataSent()

tui/tui.go

@@ -155,6 +155,9 @@ func (ui *TUI) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrpc.
 	if debugSubstreamsRun {
 		time.Sleep(debugSubstreamsRunDelay)
 		debugSubstreamsRunDelay += time.Millisecond * 100
+		if debugSubstreamsRunDelay > 500*time.Millisecond {
+			debugSubstreamsRunDelay = 500 * time.Millisecond
+		}
 	}
 	_ = isLive
 	if ui.testRunner != nil {