feat(prover): improve onBlockProposed listener (taikoxyz#266)

pkg/rpc/methods.go

@@ -25,8 +25,10 @@ var (
 	errSyncing = errors.New("syncing")
 	// syncProgressRecheckDelay is the time delay of rechecking the L2 execution engine's sync progress again,
 	// if the previous check failed.
-	syncProgressRecheckDelay = 12 * time.Second
-	minTxGasLimit            = 21000
+	syncProgressRecheckDelay    = 12 * time.Second
+	waitL1OriginPollingInterval = 6 * time.Second
+	defaultWaitL1OriginTimeout  = 3 * time.Minute
+	minTxGasLimit               = 21000
 )
 
 // ensureGenesisMatched fetches the L2 genesis block from TaikoL1 contract,
@@ -74,8 +76,8 @@ func (c *Client) ensureGenesisMatched(ctx context.Context) error {
 	return nil
 }
 
-// WaitTillL2Synced keeps waiting until the L2 execution engine is fully synced.
-func (c *Client) WaitTillL2Synced(ctx context.Context) error {
+// WaitTillL2ExecutionEngineSynced keeps waiting until the L2 execution engine is fully synced.
+func (c *Client) WaitTillL2ExecutionEngineSynced(ctx context.Context) error {
 	return backoff.Retry(
 		func() error {
 			if ctx.Err() != nil {
@@ -172,17 +174,26 @@ func (c *Client) WaitL1Origin(ctx context.Context, blockID *big.Int) (*rawdb.L1O
 		err      error
 	)
 
-	ticker := time.NewTicker(time.Second)
+	ticker := time.NewTicker(waitL1OriginPollingInterval)
 	defer ticker.Stop()
 
+	var (
+		ctxWithTimeout = ctx
+		cancel         context.CancelFunc
+	)
+	if _, ok := ctx.Deadline(); !ok {
+		ctxWithTimeout, cancel = context.WithTimeout(ctx, defaultWaitL1OriginTimeout)
+		defer cancel()
+	}
+
 	log.Debug("Start fetching L1Origin from L2 execution engine", "blockID", blockID)
 
 	for {
 		select {
-		case <-ctx.Done():
+		case <-ctxWithTimeout.Done():
 			return nil, ctx.Err()
 		case <-ticker.C:
-			l1Origin, err = c.L2.L1OriginByID(ctx, blockID)
+			l1Origin, err = c.L2.L1OriginByID(ctxWithTimeout, blockID)
 			if err != nil {
 				log.Warn("Failed to fetch L1Origin from L2 execution engine", "blockID", blockID, "error", err)
 				continue
@@ -345,6 +356,12 @@ func (c *Client) CheckL1Reorg(ctx context.Context, blockID *big.Int) (bool, *typ
 
 		l1Origin, err := c.L2.L1OriginByID(ctx, blockID)
 		if err != nil {
+			// If the L2 EE is just synced through P2P, there is a chance that the EE do not have
+			// the chain head L1Origin information recorded.
+			if errors.Is(err, ethereum.NotFound) {
+				log.Info("L1Origin not found", "blockID", blockID)
+				return false, nil, nil, nil
+			}
 			return false, nil, nil, err
 		}
 

proposer/proposer.go

@@ -186,7 +186,7 @@ func (p *Proposer) ProposeOp(ctx context.Context) error {
 	}
 
 	// Wait until L2 execution engine is synced at first.
-	if err := p.rpc.WaitTillL2Synced(ctx); err != nil {
+	if err := p.rpc.WaitTillL2ExecutionEngineSynced(ctx); err != nil {
 		return fmt.Errorf("failed to wait until L2 execution engine synced: %w", err)
 	}
 

prover/prover.go

@@ -316,13 +316,17 @@ func (p *Prover) onBlockProposed(
 		return nil
 	}
 
+	if _, err := p.rpc.WaitL1Origin(ctx, event.Id); err != nil {
+		return err
+	}
+
 	// Check whteher the L1 chain has been reorged.
 	reorged, l1CurrentToReset, lastHandledBlockIDToReset, err := p.rpc.CheckL1Reorg(
 		ctx,
 		new(big.Int).Sub(event.Id, common.Big1),
 	)
 	if err != nil {
-		return fmt.Errorf("failed to check whether L1 chain was reorged: %w", err)
+		return fmt.Errorf("failed to check whether L1 chain was reorged (eventID %d): %w", event.Id, err)
 	}
 
 	if reorged {
@@ -489,7 +493,7 @@ func (p *Prover) Name() string {
 
 // initL1Current initializes prover's L1Current cursor.
 func (p *Prover) initL1Current(startingBlockID *big.Int) error {
-	if err := p.rpc.WaitTillL2Synced(p.ctx); err != nil {
+	if err := p.rpc.WaitTillL2ExecutionEngineSynced(p.ctx); err != nil {
 		return err
 	}