From c6a7fa8ee3f305212081663816b050fef20bc07e Mon Sep 17 00:00:00 2001 From: "Thomas P." Date: Tue, 17 Feb 2026 21:29:32 +0100 Subject: [PATCH 1/2] enh(logs): log an error when a client request fails also try to add client details if we can cast it as an rpc client --- rpc/clients.go | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/rpc/clients.go b/rpc/clients.go index 78bf49c..14f2fa2 100644 --- a/rpc/clients.go +++ b/rpc/clients.go @@ -7,6 +7,7 @@ import ( "time" "github.com/hashicorp/go-multierror" + "github.com/streamingfast/eth-go/rpc" "go.uber.org/zap" ) @@ -75,6 +76,15 @@ func WithClientsContext[C any, V any](clients *Clients[C], ctx context.Context, cancel() if err != nil { + clientDetails := "" + + rpc, ok := any(client).(*rpc.Client) + if (ok) { + clientDetails = rpc.String() + } + + clients.logger.Error("failed client request", zap.Error(err), zap.String("client", clientDetails)) + errs = multierror.Append(errs, err) client, err = clients.rollingStrategy.next(clients) if err != nil { From e1afd9fbea0338f7ae690e6813fd955d9b434329 Mon Sep 17 00:00:00 2001 From: "Thomas P." Date: Tue, 17 Feb 2026 21:35:15 +0100 Subject: [PATCH 2/2] feat(blockpoller): poll blocks in parallel Previously, blockpoller was polling blocks one by one due to the lock on the clients object (tracking failures and moving to the next rpc client) We now duplicate the client list for every block poll, enabling us to poll multiple blocks in parallel from multiple RPCs. Performance impact is negligible as the main bottleneck here is the RPC call - launching multiple calls in parallel outweighs the penalty of duplicating our clients list. We also try to start with a different rpc for each block poll to avoid spamming a single endpoint with too many requests at once --- blockpoller/poller.go | 34 ++++++++++++++++++++++++++++++---- rpc/clients.go | 16 ++++++++++++++++ 2 files changed, 46 insertions(+), 4 deletions(-) diff --git a/blockpoller/poller.go b/blockpoller/poller.go index 84e1e12..b36ae36 100644 --- a/blockpoller/poller.go +++ b/blockpoller/poller.go @@ -76,6 +76,8 @@ func New[C any]( func (p *BlockPoller[C]) Run(firstStreamableBlockNum uint64, stopBlock *uint64, blockFetchBatchSize int) error { p.startBlockNumGate = firstStreamableBlockNum + p.optimisticallyPolledBlocks = map[uint64]*BlockItem{} + p.logger.Info("starting poller", zap.Uint64("first_streamable_block", firstStreamableBlockNum), zap.Uint64("block_fetch_batch_size", uint64(blockFetchBatchSize)), @@ -221,14 +223,13 @@ type BlockItem struct { } func (p *BlockPoller[C]) loadNextBlocks(requestedBlock uint64, numberOfBlockToFetch int) error { - p.optimisticallyPolledBlocks = map[uint64]*BlockItem{} p.fetching = true nailer := dhammer.NewNailer(numberOfBlockToFetch, func(ctx context.Context, blockToFetch uint64) (*BlockItem, error) { var blockItem *BlockItem err := derr.Retry(p.fetchBlockRetryCount, func(ctx context.Context) error { - - bi, err := rpc.WithClients(p.clients, func(ctx context.Context, client C) (*BlockItem, error) { + clients := p.clients.DuplicateAndStartAt(int(blockToFetch % uint64(numberOfBlockToFetch))) + bi, err := rpc.WithClients(clients, func(ctx context.Context, client C) (*BlockItem, error) { b, skipped, err := p.blockFetcher.Fetch(ctx, client, blockToFetch) if err != nil { return nil, fmt.Errorf("fetching block %d: %w", blockToFetch, err) @@ -341,12 +342,37 @@ func (p *BlockPoller[C]) requestBlock(blockNumber uint64, numberOfBlockToFetch i time.Sleep(20 * time.Millisecond) continue + } else if !p.fetching { + // Optimistically anticipate next iterations + max := blockNumber + + p.optimisticallyPolledBlocksLock.Lock() + for key := range p.optimisticallyPolledBlocks { + if key > max { + max = key + } + // Cleanup old blocks + if key < blockNumber { + delete(p.optimisticallyPolledBlocks, key) + } + } + p.optimisticallyPolledBlocksLock.Unlock() + + if (max < blockNumber + uint64(numberOfBlockToFetch)) { + go func() { + p.logger.Info("anticipating future block polls", zap.Uint64("block_num", blockNumber), zap.Uint64("max", max)) + err := p.loadNextBlocks(max + 1, numberOfBlockToFetch) + if err != nil { + p.Shutdown(err) + return + } + }() + } } p.logger.Info("block was optimistically polled", zap.Uint64("block_num", blockNumber), zap.Bool("keep", false)) return blockItem, nil } - } type FetchResponse struct { diff --git a/rpc/clients.go b/rpc/clients.go index 14f2fa2..cdb7087 100644 --- a/rpc/clients.go +++ b/rpc/clients.go @@ -57,6 +57,22 @@ func (c *Clients[C]) Add(client C) { defer c.lock.Unlock() c.clients = append(c.clients, client) } + +func (c *Clients[C]) DuplicateAndStartAt(start int) *Clients[C]{ + size := len(c.clients) + clients := Clients[C]{ + clients: make([]C, size), + maxBlockFetchDuration: c.maxBlockFetchDuration, + rollingStrategy: c.rollingStrategy, + lock: sync.Mutex{}, + logger: c.logger, + } + for i, v := range c.clients { + clients.clients[(start + i) % size] = v + } + return &clients +} + func WithClientsContext[C any, V any](clients *Clients[C], ctx context.Context, f func(context.Context, C) (v V, err error)) (v V, err error) { clients.lock.Lock() defer clients.lock.Unlock()