diff --git a/ingester/mainloop_test.go b/ingester/mainloop_test.go index 8700035..d65e6a4 100644 --- a/ingester/mainloop_test.go +++ b/ingester/mainloop_test.go @@ -190,76 +190,6 @@ func TestSendBlocks(t *testing.T) { require.Equal(t, int64(10), sentBlockNumber) } -func TestRunBlocksUseBatching(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - maxBlockNumber := int64(1000) - sentBlockNumber := int64(0) - producedBlockNumber := int64(0) - duneapi := &duneapi_mock.BlockchainIngesterMock{ - SendBlocksFunc: func(_ context.Context, blocks []models.RPCBlock) error { - if len(blocks) == 0 { - return nil - } - - // Fail if we're not sending a batch of blocks - require.Greater(t, len(blocks), 1) - - next := sentBlockNumber + 1 - for _, block := range blocks { - // We cannot send blocks out of order to DuneAPI - require.Equalf(t, next, block.BlockNumber, "expected block %d, got %d", next, block.BlockNumber) - next++ - } - - lastBlockNumber := blocks[len(blocks)-1].BlockNumber - atomic.StoreInt64(&sentBlockNumber, lastBlockNumber) - if lastBlockNumber >= maxBlockNumber { - // cancel execution when we have sent the last block - cancel() - return context.Canceled - } - - return nil - }, - PostProgressReportFunc: func(_ context.Context, _ models.BlockchainIndexProgress) error { - return nil - }, - } - rpcClient := &jsonrpc_mock.BlockchainClientMock{ - LatestBlockNumberFunc: func() (int64, error) { - return maxBlockNumber + 1, nil - }, - BlockByNumberFunc: func(_ context.Context, blockNumber int64) (models.RPCBlock, error) { - // Get blocks out of order by sleeping for a random amount of time - atomic.StoreInt64(&producedBlockNumber, blockNumber) - return models.RPCBlock{BlockNumber: blockNumber, Payload: []byte("block")}, nil - }, - CloseFunc: func() error { - return nil - }, - } - // Swap these to see logs - // logOutput := os.Stderr - logOutput := io.Discard - ing := ingester.New( - slog.New(slog.NewTextHandler(logOutput, nil)), - rpcClient, - duneapi, - ingester.Config{ - MaxConcurrentRequests: 20, // fetch blocks in multiple goroutines - // big enough compared to the time spent in block by number to ensure batching. We panic - // in the mocked Dune client if we don't get a batch of blocks (more than one block). - BlockSubmitInterval: 50 * time.Millisecond, - SkipFailedBlocks: false, - }, - nil, // progress - ) - - err := ing.Run(ctx, 1, -1) // run until canceled - require.ErrorIs(t, err, context.Canceled) // this is expected - require.GreaterOrEqual(t, sentBlockNumber, maxBlockNumber) -} - // TestRunBlocksOutOfOrder asserts that we can fetch blocks concurrently and that we ingest them in order // even if they are produced out of order. We ensure they are produced out of order by sleeping a random amount of time. func TestRunBlocksOutOfOrder(t *testing.T) { diff --git a/ingester/send.go b/ingester/send.go index 8d22df7..ca5f5f7 100644 --- a/ingester/send.go +++ b/ingester/send.go @@ -70,10 +70,6 @@ func (i *ingester) trySendCompletedBlocks( nextBlockToSend int64, ) (int64, error) { for { - if len(collectedBlocks) < maxBatchSize/10 { - // if we have very little to send, wait for next tick to avoid tiny batches impacting throughput - return nextBlockToSend, nil - } nextBlock, err := i.trySendBlockBatch(ctx, collectedBlocks, nextBlockToSend, maxBatchSize) if err != nil || nextBlock == nextBlockToSend { return nextBlock, err