Revert "PLEX-2473 LogPoller switch to batch inserts (Part 2) (#356)"

This reverts commit ffb14c0.

Author: dhaidashenko

pkg/logpoller/helper_test.go

@@ -182,7 +182,7 @@ func (th *TestHarness) assertHaveCanonical(t *testing.T, start, end int) {
 		require.NoError(t, err, "block %v", i)
 		chainBlk, err := th.Client.BlockByNumber(testutils.Context(t), big.NewInt(int64(i)))
 		require.NoError(t, err)
-		assert.Equal(t, chainBlk.Hash().String(), blk.BlockHash.String(), "block %v", i)
+		assert.Equal(t, chainBlk.Hash().Bytes(), blk.BlockHash.Bytes(), "block %v", i)
 	}
 }
 

pkg/logpoller/log_poller.go

@@ -965,7 +965,7 @@ func (lp *logPoller) backfill(ctx context.Context, start, end int64) error {
 		}
 
 		lp.lggr.Debugw("Inserting backfilled logs with batch endblock", "from", from, "to", to, "logs", len(gethLogs))
-		err = lp.orm.InsertLogsWithBlocks(ctx, convertLogs(gethLogs, blocks, lp.lggr, lp.ec.ConfiguredChainID()), []Block{endblock})
+		err = lp.orm.InsertLogsWithBlock(ctx, convertLogs(gethLogs, blocks, lp.lggr, lp.ec.ConfiguredChainID()), endblock)
 		if err != nil {
 			lp.lggr.Warnw("Unable to insert logs, retrying", "err", err, "from", from, "to", to)
 			return err
@@ -1187,118 +1187,52 @@ func (lp *logPoller) pollAndSaveLogs(ctx context.Context, currentBlockNumber int
 			return fmt.Errorf("failed to backfill finalized logs: %w", err)
 		}
 		currentBlockNumber = lastSafeBackfillBlock + 1
-		currentBlock, err = lp.getCurrentBlockMaybeHandleReorg(ctx, currentBlockNumber, nil, isReplay)
-		if err != nil {
-			// If there's an error handling the reorg, we can't be sure what state the db was left in.
-			// Resume from the latest block saved.
-			return fmt.Errorf("failed to get current block: %w", err)
-		}
-		currentBlockNumber = currentBlock.Number
 	}
 
 	for {
-		blocks, logs, err := lp.getUnfinalizedLogs(ctx, currentBlock, latestBlockNumber, safeBlockNumber, latestFinalizedBlockNumber, isReplay)
-		// even if we have an error, we may have partial logs and blocks that can be saved, so we save what we have and then retry.
-		if len(logs) > 0 || len(blocks) > 0 {
-			lp.lggr.Debugw("Saving logs", "logs", len(logs), "blocks", len(blocks), "currentBlockNumber", currentBlockNumber)
-			insertErr := lp.orm.InsertLogsWithBlocks(ctx, logs, blocks)
-			if insertErr != nil {
-				lp.lggr.Warnw("Unable to save logs, retrying later", "insertErr", insertErr, "block", currentBlockNumber, "err", err)
-				return nil
-			}
-		}
-
-		if err == nil {
-			if len(blocks) > 0 {
-				latestBlockNumber = blocks[len(blocks)-1].BlockNumber
+		if currentBlockNumber > currentBlock.Number {
+			currentBlock, err = lp.getCurrentBlockMaybeHandleReorg(ctx, currentBlockNumber, nil, isReplay)
+			if err != nil {
+				// If there's an error handling the reorg, we can't be sure what state the db was left in.
+				// Resume from the latest block saved.
+				return fmt.Errorf("failed to get current block: %w", err)
 			}
-			lp.lggr.Debugw("Finished processing unfinalized blocks", "from", currentBlockNumber, "to", latestBlockNumber)
-			return nil
-		}
-
-		var reorgErr *reorgError
-		if !errors.As(err, &reorgErr) {
-			return fmt.Errorf("failed to get unfinalized logs: %w", err)
+			currentBlockNumber = currentBlock.Number
 		}
 
-		lp.lggr.Warnw("Reorg detected during unfinalized log processing, handling reorg", "err", err, "currentBlockNumber", currentBlockNumber, "lastKnownMatchingHead", reorgErr.ReorgedAt.Number)
-		currentBlock, err = lp.handleReorg(ctx, reorgErr.ReorgedAt)
-		if err != nil {
-			return fmt.Errorf("failed to handle reorg: %w", err)
-		}
-		currentBlockNumber = currentBlock.Number
-		lp.lggr.Infow("Finished handling reorg, resuming log processing from new block after LCA", "currentBlockNumber", currentBlock.Number)
-	}
-}
-
-type reorgError struct {
-	ReorgedAt *evmtypes.Head
-}
-
-func newReorgError(reorgedAt *evmtypes.Head) error {
-	return &reorgError{ReorgedAt: reorgedAt}
-}
-
-func (e *reorgError) Error() string {
-	return fmt.Sprintf("reorg detected at block %d", e.ReorgedAt.Number)
-}
-
-func (lp *logPoller) getUnfinalizedLogs(ctx context.Context, currentBlock *evmtypes.Head, latest, safe, finalized int64, isReplay bool) ([]Block, []Log, error) {
-	const maxUnfinalizedBlocks = 2000
-	var logs []Log
-	var blocks []Block
-	for {
 		h := currentBlock.Hash
-		rpcLogs, err := lp.latencyMonitor.FilterLogs(ctx, lp.Filter(nil, nil, &h))
+		var logs []types.Log
+		logs, err = lp.latencyMonitor.FilterLogs(ctx, lp.Filter(nil, nil, &h))
 		if err != nil {
-			lp.lggr.Warnw("Unable to query for logs, retrying on next poll", "err", err, "block", currentBlock.Number)
-			return blocks, logs, fmt.Errorf("unable to query for logs: %w", err)
+			lp.lggr.Warnw("Unable to query for logs, retrying", "err", err, "block", currentBlockNumber)
+			return nil
 		}
-		lp.lggr.Debugw("Unfinalized log query", "logs", len(logs), "currentBlockNumber", currentBlock.Number, "blockHash", currentBlock.Hash, "timestamp", currentBlock.Timestamp)
+		lp.lggr.Debugw("Unfinalized log query", "logs", len(logs), "currentBlockNumber", currentBlockNumber, "blockHash", currentBlock.Hash, "timestamp", currentBlock.Timestamp)
 		block := Block{
 			BlockHash:            h,
-			BlockNumber:          currentBlock.Number,
+			BlockNumber:          currentBlockNumber,
 			BlockTimestamp:       currentBlock.Timestamp,
-			FinalizedBlockNumber: finalized,
-			SafeBlockNumber:      safe,
-		}
-		logs = append(logs, convertLogs(rpcLogs, []Block{block}, lp.lggr, lp.ec.ConfiguredChainID())...)
-		blocks = append(blocks, block)
-
-		if currentBlock.Number >= latest {
-			return blocks, logs, nil
-		}
-
-		if len(blocks) >= maxUnfinalizedBlocks {
-			lp.lggr.Warnw("Too many unfinalized blocks, stopping log retrieval to avoid OOM", "currentBlockNumber", currentBlock.Number, "latestBlockNumber", latest)
-			return blocks, logs, nil
-		}
-
-		nextBlock, err := lp.headerByNumber(ctx, currentBlock.Number+1)
+			FinalizedBlockNumber: latestFinalizedBlockNumber,
+			SafeBlockNumber:      safeBlockNumber,
+		}
+		err = lp.orm.InsertLogsWithBlock(
+			ctx,
+			convertLogs(logs, []Block{block}, lp.lggr, lp.ec.ConfiguredChainID()),
+			block,
+		)
 		if err != nil {
-			lp.lggr.Warnw("Unable to get next block header, retrying on next poll", "err", err, "block", currentBlock.Number)
-			return blocks, logs, fmt.Errorf("unable to get next block: %w", err)
-		}
-
-		if nextBlock.ParentHash != currentBlock.Hash {
-			return blocks, logs, newReorgError(nextBlock)
+			lp.lggr.Warnw("Unable to save logs resuming from last saved block + 1", "err", err, "block", currentBlockNumber)
+			return nil
 		}
-
-		if isReplay {
-			// During replay, we also check if the next block matches what we have in the DB to avoid false positives on reorgs due to finality violation.
-			nextBlockDB, err := lp.orm.SelectBlockByNumber(ctx, nextBlock.Number)
-			if err != nil && !pkgerrors.Is(err, sql.ErrNoRows) {
-				lp.lggr.Warnw("Unable to get next block from DB during replay, retrying on next poll", "err", err, "block", nextBlock.Number)
-				return blocks, logs, fmt.Errorf("failed to get next block from DB during replay: %w", err)
-			}
-
-			if nextBlockDB != nil && nextBlock.Hash != nextBlockDB.BlockHash {
-				return blocks, logs, newReorgError(nextBlock)
-			}
+		// Update current block.
+		// Same reorg detection on unfinalized blocks.
+		currentBlockNumber++
+		if currentBlockNumber > latestBlockNumber {
+			break
 		}
-
-		currentBlock = nextBlock
 	}
+
+	return nil
 }
 
 // Returns information about latestBlock, latestFinalizedBlockNumber provided by HeadTracker

pkg/logpoller/log_poller_test.go

@@ -690,9 +690,7 @@ func TestLogPoller_SynchronizedWithGeth(t *testing.T) {
 	owner.GasPrice = big.NewInt(10e9)
 	p.Property("synchronized with geth", prop.ForAll(func(mineOrReorg []uint64) bool {
 		// After the set of reorgs, we should have the same canonical blocks that geth does.
-		seed := time.Now().UnixNano()
-		localRand := rand.New(rand.NewSource(seed))
-		t.Log("Starting test", mineOrReorg, "seed", seed)
+		t.Log("Starting test", mineOrReorg)
 		chainID := testutils.NewRandomEVMChainID()
 		// Set up a test chain with a log emitting contract deployed.
 		orm := logpoller.NewORM(chainID, db, lggr)
@@ -744,7 +742,7 @@ func TestLogPoller_SynchronizedWithGeth(t *testing.T) {
 		}
 		// Randomly pick to mine or reorg
 		for i := 0; i < numChainInserts; i++ {
-			if localRand.Int63()%2 == 0 {
+			if rand.Int63()%2 == 0 {
 				// Mine blocks
 				for j := 0; j < int(mineOrReorg[i]); j++ {
 					backend.Commit()
@@ -772,7 +770,7 @@ func TestLogPoller_SynchronizedWithGeth(t *testing.T) {
 				require.NoError(t, err1)
 				t.Logf("New latest (%v, %x), latest parent %x)\n", latest.NumberU64(), latest.Hash(), latest.ParentHash())
 			}
-			lp.PollAndSaveLogs(testutils.Context(t), currentBlock.BlockNumber+1, false)
+			lp.PollAndSaveLogs(testutils.Context(t), currentBlock.BlockNumber, false)
 			currentBlock, err = lp.LatestBlock(testutils.Context(t))
 			require.NoError(t, err)
 		}

pkg/logpoller/observability.go

@@ -52,16 +52,16 @@ func (o *ObservedORM) InsertLogs(ctx context.Context, logs []Log) error {
 	err := withObservedExec(ctx, o, "InsertLogs", metrics.Create, func() error {
 		return o.ORM.InsertLogs(ctx, logs)
 	})
-	trackInsertedLogsAndBlocks(ctx, o, logs, nil, err)
+	trackInsertedLogsAndBlock(ctx, o, logs, nil, err)
 	trackInsertedBlockLatency(ctx, o, logs, err)
 	return err
 }
 
-func (o *ObservedORM) InsertLogsWithBlocks(ctx context.Context, logs []Log, blocks []Block) error {
+func (o *ObservedORM) InsertLogsWithBlock(ctx context.Context, logs []Log, block Block) error {
 	err := withObservedExec(ctx, o, "InsertLogsWithBlock", metrics.Create, func() error {
-		return o.ORM.InsertLogsWithBlocks(ctx, logs, blocks)
+		return o.ORM.InsertLogsWithBlock(ctx, logs, block)
 	})
-	trackInsertedLogsAndBlocks(ctx, o, logs, blocks, err)
+	trackInsertedLogsAndBlock(ctx, o, logs, &block, err)
 	trackInsertedBlockLatency(ctx, o, logs, err)
 	return err
 }
@@ -290,15 +290,15 @@ func withObservedExec(ctx context.Context, o *ObservedORM, query string, queryTy
 	return exec()
 }
 
-func trackInsertedLogsAndBlocks(ctx context.Context, o *ObservedORM, logs []Log, blocks []Block, err error) {
+func trackInsertedLogsAndBlock(ctx context.Context, o *ObservedORM, logs []Log, block *Block, err error) {
 	if err != nil {
 		return
 	}
 	ctx, cancel := context.WithTimeout(ctx, client.QueryTimeout)
 	defer cancel()
 	o.metrics.IncrementLogsInserted(ctx, int64(len(logs)))
-	if len(blocks) > 0 {
-		o.metrics.IncrementBlocksInserted(ctx, int64(len(blocks)))
+	if block != nil {
+		o.metrics.IncrementBlocksInserted(ctx, 1)
 	}
 }
 

pkg/logpoller/observability_test.go

@@ -41,11 +41,9 @@ func TestMultipleMetricsArePublished(t *testing.T) {
 	_, _ = orm.SelectLatestLogEventSigsAddrsWithConfs(ctx, 0, []common.Address{{}}, []common.Hash{{}}, 1)
 	_, _ = orm.SelectIndexedLogsCreatedAfter(ctx, common.Address{}, common.Hash{}, 1, []common.Hash{}, time.Now(), 0)
 	_ = orm.InsertLogs(ctx, []Log{})
-	_ = orm.InsertLogsWithBlocks(ctx, []Log{}, []Block{
-		{
-			BlockNumber:    1,
-			BlockTimestamp: time.Now(),
-		},
+	_ = orm.InsertLogsWithBlock(ctx, []Log{}, Block{
+		BlockNumber:    1,
+		BlockTimestamp: time.Now(),
 	})
 
 	require.Equal(t, 13, testutil.CollectAndCount(orm.queryDuration))
@@ -115,25 +113,21 @@ func TestCountersAreProperlyPopulatedForWrites(t *testing.T) {
 	assert.Equal(t, 10, int(testutil.ToFloat64(orm.logsInserted.WithLabelValues(network, "420"))))
 	assert.Equal(t, 1, testutil.CollectAndCount(orm.discoveryLatency))
 	// Insert 5 more logs with block
-	require.NoError(t, orm.InsertLogsWithBlocks(ctx, logs[10:15], []Block{
-		{
-			BlockHash:            utils.RandomBytes32(),
-			BlockNumber:          10,
-			BlockTimestamp:       time.Now(),
-			FinalizedBlockNumber: 5,
-		},
+	require.NoError(t, orm.InsertLogsWithBlock(ctx, logs[10:15], Block{
+		BlockHash:            utils.RandomBytes32(),
+		BlockNumber:          10,
+		BlockTimestamp:       time.Now(),
+		FinalizedBlockNumber: 5,
 	}))
 	assert.Equal(t, 15, int(testutil.ToFloat64(orm.logsInserted.WithLabelValues(network, "420"))))
 	assert.Equal(t, 1, int(testutil.ToFloat64(orm.blocksInserted.WithLabelValues(network, "420"))))
 
 	// Insert 5 more logs with block
-	require.NoError(t, orm.InsertLogsWithBlocks(ctx, logs[15:], []Block{
-		{
-			BlockHash:            utils.RandomBytes32(),
-			BlockNumber:          15,
-			BlockTimestamp:       time.Now(),
-			FinalizedBlockNumber: 5,
-		},
+	require.NoError(t, orm.InsertLogsWithBlock(ctx, logs[15:], Block{
+		BlockHash:            utils.RandomBytes32(),
+		BlockNumber:          15,
+		BlockTimestamp:       time.Now(),
+		FinalizedBlockNumber: 5,
 	}))
 	assert.Equal(t, 20, int(testutil.ToFloat64(orm.logsInserted.WithLabelValues(network, "420"))))
 	assert.Equal(t, 2, int(testutil.ToFloat64(orm.blocksInserted.WithLabelValues(network, "420"))))
@@ -149,11 +143,9 @@ func TestCountersAreProperlyPopulatedForWrites(t *testing.T) {
 	assert.Equal(t, 2, counterFromGaugeByLabels(orm.datasetSize, network, "420", "DeleteBlocksBefore", "delete"))
 
 	// Don't update counters in case of an error
-	require.Error(t, orm.InsertLogsWithBlocks(ctx, logs, []Block{
-		{
-			BlockHash:      utils.RandomBytes32(),
-			BlockTimestamp: time.Now(),
-		},
+	require.Error(t, orm.InsertLogsWithBlock(ctx, logs, Block{
+		BlockHash:      utils.RandomBytes32(),
+		BlockTimestamp: time.Now(),
 	}))
 	assert.Equal(t, 20, int(testutil.ToFloat64(orm.logsInserted.WithLabelValues(network, "420"))))
 	assert.Equal(t, 2, int(testutil.ToFloat64(orm.blocksInserted.WithLabelValues(network, "420"))))

pkg/logpoller/orm.go

@@ -24,7 +24,7 @@ import (
 // What is more, LogPoller should not be aware of the underlying database implementation and delegate all the queries to the ORM.
 type ORM interface {
 	InsertLogs(ctx context.Context, logs []Log) error
-	InsertLogsWithBlocks(ctx context.Context, logs []Log, blocks []Block) error
+	InsertLogsWithBlock(ctx context.Context, logs []Log, block Block) error
 	InsertFilter(ctx context.Context, filter Filter) error
 
 	LoadFilters(ctx context.Context) (map[string]Filter, error)
@@ -112,18 +112,6 @@ func (o *DSORM) InsertBlock(ctx context.Context, blockHash common.Hash, blockNum
 	return err
 }
 
-func (o *DSORM) InsertBlocks(ctx context.Context, blocks []Block) error {
-	const q = `INSERT INTO evm.log_poller_blocks
-				(evm_chain_id, block_hash, block_number, block_timestamp, finalized_block_number, created_at, safe_block_number)
-      		VALUES (:evm_chain_id, :block_hash, :block_number, :block_timestamp, :finalized_block_number, NOW(), :safe_block_number)
-			ON CONFLICT DO NOTHING`
-	// maintain behaviour of InsertBlock
-	for i := range blocks {
-		blocks[i].EVMChainID = sqlutil.New(o.chainID)
-	}
-	return batchInsert(ctx, o.ds, q, blocks, 1000)
-}
-
 // InsertFilter is idempotent.
 //
 // Each address/event pair must have a unique job id, so it may be removed when the job is deleted.
@@ -556,10 +544,10 @@ func (o *DSORM) InsertLogs(ctx context.Context, logs []Log) error {
 	})
 }
 
-func (o *DSORM) InsertLogsWithBlocks(ctx context.Context, logs []Log, blocks []Block) error {
+func (o *DSORM) InsertLogsWithBlock(ctx context.Context, logs []Log, block Block) error {
 	// Optimization, don't open TX when there is only a block to be persisted
 	if len(logs) == 0 {
-		return o.InsertBlocks(ctx, blocks)
+		return o.InsertBlock(ctx, block.BlockHash, block.BlockNumber, block.BlockTimestamp, block.FinalizedBlockNumber, block.SafeBlockNumber)
 	}
 
 	if err := o.validateLogs(logs); err != nil {
@@ -568,7 +556,7 @@ func (o *DSORM) InsertLogsWithBlocks(ctx context.Context, logs []Log, blocks []B
 
 	// Block and logs goes with the same TX to ensure atomicity
 	return o.Transact(ctx, func(orm *DSORM) error {
-		err := orm.InsertBlocks(ctx, blocks)
+		err := orm.InsertBlock(ctx, block.BlockHash, block.BlockNumber, block.BlockTimestamp, block.FinalizedBlockNumber, block.SafeBlockNumber)
 		if err != nil {
 			return err
 		}
@@ -577,22 +565,20 @@ func (o *DSORM) InsertLogsWithBlocks(ctx context.Context, logs []Log, blocks []B
 }
 
 func (o *DSORM) insertLogsWithinTx(ctx context.Context, logs []Log, tx sqlutil.DataSource) error {
-	const q = `INSERT INTO evm.logs
+	batchInsertSize := 4000
+	for i := 0; i < len(logs); i += batchInsertSize {
+		start, end := i, i+batchInsertSize
+		if end > len(logs) {
+			end = len(logs)
+		}
+
+		query := `INSERT INTO evm.logs
 					(evm_chain_id, log_index, block_hash, block_number, block_timestamp, address, event_sig, topics, tx_hash, data, created_at)
 				VALUES
 					(:evm_chain_id, :log_index, :block_hash, :block_number, :block_timestamp, :address, :event_sig, :topics, :tx_hash, :data, NOW())
 				ON CONFLICT DO NOTHING`
-	return batchInsert(ctx, tx, q, logs, 4000)
-}
-
-func batchInsert[T any](ctx context.Context, ds sqlutil.DataSource, query string, objs []T, batchInsertSize int) error {
-	for i := 0; i < len(objs); i += batchInsertSize {
-		start, end := i, i+batchInsertSize
-		if end > len(objs) {
-			end = len(objs)
-		}
 
-		_, err := ds.NamedExecContext(ctx, query, objs[start:end])
+		_, err := tx.NamedExecContext(ctx, query, logs[start:end])
 		if err != nil {
 			if pkgerrors.Is(err, context.DeadlineExceeded) && batchInsertSize > 500 {
 				// In case of DB timeouts, try to insert again with a smaller batch upto a limit