fix(relayer): chunk eth log queries and honor configured start block

The deposit-event poller issued a single eth_getLogs spanning
currentBlock+1..head, so a large catch-up gap tripped the provider's
range limit ("Block range limit exceeded") and retried the same
oversized range forever. The documented MaxBlockRange chunking existed
(chunkRange) but was never wired in. Walk the range in MaxBlockRange
slices via a new scanDepositRange helper, advancing scan progress only
through the last successful slice.

Also, loadEthereumOffset unconditionally used the stored chain_state
block and ignored eth_start_block once any progress was persisted, so
operators could only fast-forward by editing the DB. Take the later of
the stored block and the configured start_block; config can fast-forward
but never rewinds the relayer.

Author: sadiq1971

pkg/ethereum/client.go

@@ -275,55 +275,74 @@ func (c *Client) WatchDepositEvents(ctx context.Context, fromBlock uint64, handl
 					return
 				}
 
-				// Query for events from currentBlock+1 to latestBlock
-				opts := &bind.FilterOpts{
-					Start:   currentBlock + 1,
-					End:     &latestBlock,
-					Context: ctx,
-				}
-
-				filterStart := time.Now()
-				iter, err := c.bridge.FilterDepositToCanton(opts, nil, nil, nil)
-				c.observeRPC("filter_deposit_events", filterStart, err)
-				if err != nil {
-					c.metrics.EventPollFailuresTotal.WithLabelValues("filter_events").Inc()
-					c.logger.Warn("Failed to filter deposit events", zap.Error(err))
-					return
-				}
-
-				for iter.Next() {
-					c.metrics.EventsFetchedTotal.Inc()
-					event := iter.Event
-					depositEvent := &DepositEvent{
-						Token:           event.Token,
-						Sender:          event.Sender,
-						CantonRecipient: event.CantonRecipient,
-						Amount:          event.Amount,
-						Nonce:           event.Nonce,
-						BlockNumber:     event.Raw.BlockNumber,
-						TxHash:          event.Raw.TxHash,
-						LogIndex:        event.Raw.Index,
-					}
-
-					if err := handler(depositEvent); err != nil {
-						c.logger.Error("Failed to handle deposit event",
-							zap.Error(err),
-							zap.String("tx_hash", event.Raw.TxHash.Hex()))
+				// Walk [currentBlock+1, latestBlock] in slices of at most
+				// MaxBlockRange blocks so each eth_getLogs request stays under
+				// the provider's per-call range cap. Progress advances only
+				// through the last successful slice; a failing slice (and
+				// everything after it) is retried on the next tick.
+				for _, r := range chunkRange(currentBlock+1, latestBlock, c.config.MaxBlockRange) {
+					if err := c.scanDepositRange(ctx, r, handler); err != nil {
+						return
 					}
+					currentBlock = r.end
+					c.setLastScannedBlock(currentBlock)
 				}
+			}()
+		}
+	}
+}
 
-				if err := iter.Error(); err != nil {
-					c.metrics.EventPollFailuresTotal.WithLabelValues("iterator").Inc()
-					c.logger.Warn("Iterator error", zap.Error(err))
-				}
-				iter.Close()
+// scanDepositRange filters DepositToCanton events for a single inclusive block
+// range and dispatches each to handler. A filter or iterator error is returned
+// so the caller can stop advancing scan progress and retry the range on the
+// next tick. Handler errors are logged but do not abort the range.
+func (c *Client) scanDepositRange(ctx context.Context, r blockRange, handler func(*DepositEvent) error) error {
+	opts := &bind.FilterOpts{
+		Start:   r.start,
+		End:     &r.end,
+		Context: ctx,
+	}
 
-				// Update scan progress even if there were no events
-				currentBlock = latestBlock
-				c.setLastScannedBlock(currentBlock)
-			}()
+	filterStart := time.Now()
+	iter, err := c.bridge.FilterDepositToCanton(opts, nil, nil, nil)
+	c.observeRPC("filter_deposit_events", filterStart, err)
+	if err != nil {
+		c.metrics.EventPollFailuresTotal.WithLabelValues("filter_events").Inc()
+		c.logger.Warn("Failed to filter deposit events",
+			zap.Uint64("from_block", r.start),
+			zap.Uint64("to_block", r.end),
+			zap.Error(err))
+		return err
+	}
+	defer iter.Close()
+
+	for iter.Next() {
+		c.metrics.EventsFetchedTotal.Inc()
+		event := iter.Event
+		depositEvent := &DepositEvent{
+			Token:           event.Token,
+			Sender:          event.Sender,
+			CantonRecipient: event.CantonRecipient,
+			Amount:          event.Amount,
+			Nonce:           event.Nonce,
+			BlockNumber:     event.Raw.BlockNumber,
+			TxHash:          event.Raw.TxHash,
+			LogIndex:        event.Raw.Index,
+		}
+
+		if err := handler(depositEvent); err != nil {
+			c.logger.Error("Failed to handle deposit event",
+				zap.Error(err),
+				zap.String("tx_hash", event.Raw.TxHash.Hex()))
 		}
 	}
+
+	if err := iter.Error(); err != nil {
+		c.metrics.EventPollFailuresTotal.WithLabelValues("iterator").Inc()
+		c.logger.Warn("Iterator error", zap.Error(err))
+		return err
+	}
+	return nil
 }
 
 // WithdrawFromCanton submits a withdrawal transaction

pkg/relayer/engine/engine.go

@@ -365,6 +365,15 @@ func (e *Engine) loadEthereumOffset(ctx context.Context) error {
 
 	if state != nil {
 		e.ethLastBlock = state.LastBlock
+		// Honor a configured start_block that is ahead of the stored progress so
+		// operators can fast-forward past a problematic range without editing the
+		// DB. The later of the two always wins; config never rewinds the relayer.
+		if e.config.EthStartBlock > e.ethLastBlock {
+			e.logger.Info("Configured start_block is ahead of stored block, fast-forwarding",
+				zap.Uint64("stored_block", e.ethLastBlock),
+				zap.Uint64("configured_start_block", e.config.EthStartBlock))
+			e.ethLastBlock = e.config.EthStartBlock
+		}
 		e.logger.Info("Loaded Ethereum last block", zap.Uint64("block", e.ethLastBlock))
 		return nil
 	}