Prevent multiple blobs/ rollups in the same tx

go/enclave/components/interfaces.go

@@ -153,7 +153,9 @@ type RollupProducer interface {
 }
 
 type RollupConsumer interface {
-	// ProcessBlobsInBlock - extracts the blob hashes from the block's transactions and builds the blob hashes from the blobs,
+	// ProcessRollups - extracts the blob hashes from the block's transactions and builds the blob hashes from the blobs,
 	// compares this with the hashes seen in the block.
-	ProcessBlobsInBlock(ctx context.Context, processed *common.ProcessedL1Data) error
+	ProcessRollups(ctx context.Context, rollups []*common.ExtRollup) error
+	// GetRollupsFromL1Data -
+	GetRollupsFromL1Data(processed *common.ProcessedL1Data) ([]*common.ExtRollup, error)
 }

go/enclave/components/rollup_consumer.go

@@ -48,30 +48,9 @@ func NewRollupConsumer(
 	}
 }
 
-// ProcessBlobsInBlock - processes the blobs in a block, extracts the rollups, verifies the rollups and stores them
-func (rc *rollupConsumerImpl) ProcessBlobsInBlock(ctx context.Context, processed *common.ProcessedL1Data) error {
-	defer core.LogMethodDuration(rc.logger, measure.NewStopwatch(), "Rollup consumer processed blobs", log.BlockHashKey, processed.BlockHeader.Hash())
-
-	block := processed.BlockHeader
-	rollups, err := rc.extractAndVerifyRollups(processed)
-	if err != nil {
-		rc.logger.Error("Failed to extract rollups from block", log.BlockHashKey, block.Hash(), log.ErrKey, err)
-		return err
-	}
-	if len(rollups) == 0 {
-		rc.logger.Trace("No rollups found in block", log.BlockHashKey, block.Hash())
-		return nil
-	}
-
-	rollups, err = rc.getSignedRollup(rollups)
-	if err != nil {
-		return err
-	}
-
-	if len(rollups) > 1 {
-		// todo - we need to sort this out
-		rc.logger.Warn(fmt.Sprintf("Multiple rollups %d in block %s", len(rollups), block.Hash()))
-	}
+// ProcessRollups - processes the rollups found in the block, verifies the rollups and stores them
+func (rc *rollupConsumerImpl) ProcessRollups(ctx context.Context, rollups []*common.ExtRollup) error {
+	defer core.LogMethodDuration(rc.logger, measure.NewStopwatch(), "Rollup consumer processed blobs")
 
 	for _, rollup := range rollups {
 		l1CompressionBlock, err := rc.storage.FetchBlock(ctx, rollup.Header.CompressionL1Head)
@@ -103,6 +82,33 @@ func (rc *rollupConsumerImpl) ProcessBlobsInBlock(ctx context.Context, processed
 	return nil
 }
 
+// GetRollupsFromL1Data - extracts the rollups from the processed L1 data and checks sequencer signature on them
+func (rc *rollupConsumerImpl) GetRollupsFromL1Data(processed *common.ProcessedL1Data) ([]*common.ExtRollup, error) {
+	defer core.LogMethodDuration(rc.logger, measure.NewStopwatch(), "Rollup consumer get rollups from L1 data", log.BlockHashKey, processed.BlockHeader.Hash())
+
+	block := processed.BlockHeader
+	rollups, err := rc.extractAndVerifyRollups(processed)
+	if err != nil {
+		rc.logger.Error("Failed to extract rollups from block", log.BlockHashKey, block.Hash(), log.ErrKey, err)
+		return nil, err // if multiple rollups are found with the same tx hash we will return here
+	}
+	if len(rollups) == 0 {
+		rc.logger.Trace("No rollups found in block", log.BlockHashKey, block.Hash())
+		return nil, nil
+	}
+
+	rollups, err = rc.getSignedRollup(rollups)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(rollups) > 0 {
+		// this is allowed as long as they come from unique transactions
+		rc.logger.Trace(fmt.Sprintf("Multiple rollups %d in block %s", len(rollups), block.Hash()))
+	}
+	return rollups, nil
+}
+
 func (rc *rollupConsumerImpl) getSignedRollup(rollups []*common.ExtRollup) ([]*common.ExtRollup, error) {
 	signedRollup := make([]*common.ExtRollup, 0)
 
@@ -117,11 +123,9 @@ func (rc *rollupConsumerImpl) getSignedRollup(rollups []*common.ExtRollup) ([]*c
 	return signedRollup, nil
 }
 
-// todo - when processing the rollup, instead of looking up batches one by one, compare the last sequence number from the db with the ones in the rollup
-// extractAndVerifyRollups returns a list of the rollups published in this block
-// It processes each transaction, attempting to extract and verify rollups
-// If a transaction is not a rollup or fails verification, it's skipped
-// The function only returns an error if there's a critical failure in rollup reconstruction
+// extractAndVerifyRollups extracts rollups from L1 transactions in the processed block.
+// It verifies blob hashes match the rollup hashes and ensures each transaction only contains one rollup.
+// Returns an error if multiple rollups are found in the same transaction or if rollup reconstruction fails.
 func (rc *rollupConsumerImpl) extractAndVerifyRollups(processed *common.ProcessedL1Data) ([]*common.ExtRollup, error) {
 	rollupTxs := processed.GetEvents(common.RollupTx)
 	rollups := make([]*common.ExtRollup, 0, len(rollupTxs))
@@ -131,6 +135,8 @@ func (rc *rollupConsumerImpl) extractAndVerifyRollups(processed *common.Processe
 		return nil, err
 	}
 
+	txsSeen := make(map[gethcommon.Hash]bool)
+
 	for i, tx := range rollupTxs {
 		t := rc.MgmtContractLib.DecodeTx(tx.Transaction)
 		if t == nil {
@@ -142,6 +148,12 @@ func (rc *rollupConsumerImpl) extractAndVerifyRollups(processed *common.Processe
 			continue
 		}
 
+		// prevent the case where someone pushes a blob to the same slot. multiple rollups can be found in a block,
+		// but they must come from unique transactions
+		if txsSeen[tx.Transaction.Hash()] {
+			return nil, fmt.Errorf("multiple rollups from same transaction: %s", tx.Transaction.Hash())
+		}
+
 		if err := verifyBlobHashes(rollupHashes, blobHashes); err != nil {
 			rc.logger.Warn(fmt.Sprintf("blob hashes in rollup at index %d do not match the rollup blob hashes. Cause: %s", i, err))
 			continue // Blob hashes don't match, skip this rollup
@@ -155,6 +167,7 @@ func (rc *rollupConsumerImpl) extractAndVerifyRollups(processed *common.Processe
 		}
 
 		rollups = append(rollups, r)
+		txsSeen[tx.Transaction.Hash()] = true
 
 		rc.logger.Info("Extracted rollup from block", log.RollupHashKey, r.Hash(), log.BlockHashKey, processed.BlockHeader.Hash())
 	}

go/enclave/enclave_admin_service.go

@@ -467,6 +467,11 @@ func (e *enclaveAdminService) streamEventsForNewHeadBatch(ctx context.Context, b
 
 func (e *enclaveAdminService) ingestL1Block(ctx context.Context, processed *common.ProcessedL1Data) (*components.BlockIngestionType, error) {
 	e.logger.Info("Start ingesting block", log.BlockHashKey, processed.BlockHeader.Hash())
+	rollups, err := e.rollupConsumer.GetRollupsFromL1Data(processed)
+	if err != nil {
+		// early return before storing block if multiple rollups are found in the block
+		return nil, err
+	}
 	ingestion, err := e.l1BlockProcessor.Process(ctx, processed)
 	if err != nil {
 		// only warn for unexpected errors
@@ -478,7 +483,7 @@ func (e *enclaveAdminService) ingestL1Block(ctx context.Context, processed *comm
 		return nil, err
 	}
 
-	err = e.rollupConsumer.ProcessBlobsInBlock(ctx, processed)
+	err = e.rollupConsumer.ProcessRollups(ctx, rollups)
 	if err != nil && !errors.Is(err, components.ErrDuplicateRollup) {
 		e.logger.Error("Encountered error while processing l1 block rollups", log.ErrKey, err)
 		// Unsure what to do here; block has been stored