parlia: implement IsDataAvailable function;

consensus/parlia/blob_sidecar.go

@@ -0,0 +1,39 @@
+package parlia
+
+import (
+	"crypto/sha256"
+	"fmt"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
+)
+
+// validateBlobSidecar it is same as validateBlobSidecar in core/txpool/validation.go
+func validateBlobSidecar(hashes []common.Hash, sidecar *types.BlobTxSidecar) error {
+	if len(sidecar.Blobs) != len(hashes) {
+		return fmt.Errorf("invalid number of %d blobs compared to %d blob hashes", len(sidecar.Blobs), len(hashes))
+	}
+	if len(sidecar.Commitments) != len(hashes) {
+		return fmt.Errorf("invalid number of %d blob commitments compared to %d blob hashes", len(sidecar.Commitments), len(hashes))
+	}
+	if len(sidecar.Proofs) != len(hashes) {
+		return fmt.Errorf("invalid number of %d blob proofs compared to %d blob hashes", len(sidecar.Proofs), len(hashes))
+	}
+	// Blob quantities match up, validate that the provers match with the
+	// transaction hash before getting to the cryptography
+	hasher := sha256.New()
+	for i, vhash := range hashes {
+		computed := kzg4844.CalcBlobHashV1(hasher, &sidecar.Commitments[i])
+		if vhash != computed {
+			return fmt.Errorf("blob %d: computed hash %#x mismatches transaction one %#x", i, computed, vhash)
+		}
+	}
+	// Blob commitments match with the hashes in the transaction, verify the
+	// blobs themselves via KZG
+	for i := range sidecar.Blobs {
+		if err := kzg4844.VerifyBlobProof(sidecar.Blobs[i], sidecar.Commitments[i], sidecar.Proofs[i]); err != nil {
+			return fmt.Errorf("invalid blob %d: %v", i, err)
+		}
+	}
+	return nil
+}

consensus/parlia/parlia.go

@@ -351,8 +351,41 @@ func (p *Parlia) VerifyHeaders(chain consensus.ChainHeaderReader, headers []*typ
 }
 
 // IsDataAvailable it checks that the blobTx block has available blob data
-// TODO(GalaIO): implement it later
 func (p *Parlia) IsDataAvailable(chain consensus.ChainHeaderReader, block *types.Block, blobs types.BlobTxSidecars) error {
+	if !p.chainConfig.IsCancun(block.Number(), block.Time()) {
+		return nil
+	}
+	// only required to check within BlobLocalAvailableThreshold block's DA
+	currentHeader := chain.CurrentHeader()
+	if block.NumberU64() < currentHeader.Number.Uint64()-params.BlobLocalAvailableThreshold {
+		return nil
+	}
+
+	// alloc block's versionedHashes
+	versionedHashes := make([][]common.Hash, 0, len(block.Transactions()))
+	for _, tx := range block.Transactions() {
+		versionedHashes = append(versionedHashes, tx.BlobHashes())
+	}
+	if len(versionedHashes) != len(blobs) {
+		return errors.New("blobs do not match the versionedHashes length")
+	}
+
+	// check blob amount
+	blobCnt := 0
+	for _, h := range versionedHashes {
+		blobCnt += len(h)
+	}
+	if blobCnt > params.MaxBlobGasPerBlock/params.BlobTxBlobGasPerBlob {
+		return fmt.Errorf("too many blobs in block: have %d, permitted %d", blobCnt, params.MaxBlobGasPerBlock/params.BlobTxBlobGasPerBlob)
+	}
+
+	// check blob and versioned hash
+	for i := range versionedHashes {
+		if err := validateBlobSidecar(versionedHashes[i], blobs[i]); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

core/blockchain.go

@@ -1965,19 +1965,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 	}
 
 	for ; block != nil && err == nil || errors.Is(err, ErrKnownBlock); block, err = it.next() {
-		// TODO(GalaIO): check blob data available first
-		//if bc.chainConfig.IsCancun(block.Number(), block.Time()) {
-		//	if posa, ok := bc.engine.(consensus.PoSA); ok {
-		//		blobs, exist := bc.receivedBlobsCache.Load(block.Hash())
-		//		if !exist {
-		//			return it.index, fmt.Errorf("cannot find the target block's blob info, block: %v, hash: %v", block.NumberU64(), block.Hash())
-		//		}
-		//		if err = posa.IsDataAvailable(bc, block, blobs.(types.BlobTxSidecars)); err != nil {
-		//			return it.index, err
-		//		}
-		//	}
-		//}
-
 		// If the chain is terminating, stop processing blocks
 		if bc.insertStopped() {
 			log.Debug("Abort during block processing")
@@ -2034,6 +2021,20 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 		if parent == nil {
 			parent = bc.GetHeader(block.ParentHash(), block.NumberU64()-1)
 		}
+
+		// check blob data available first
+		// TODO(GalaIO): move IsDataAvailable combine into verifyHeaders?
+		if bc.chainConfig.IsCancun(block.Number(), block.Time()) {
+			if posa, ok := bc.engine.(consensus.PoSA); ok {
+				blobs, exist := bc.receivedBlobsCache.Load(block.Hash())
+				if !exist {
+					return it.index, fmt.Errorf("cannot find the target block's blob info, block: %v, hash: %v", block.NumberU64(), block.Hash())
+				}
+				if err = posa.IsDataAvailable(bc, block, blobs.(types.BlobTxSidecars)); err != nil {
+					return it.index, err
+				}
+			}
+		}
 		statedb, err := state.NewWithSharedPool(parent.Root, bc.stateCache, bc.snaps)
 		if err != nil {
 			return it.index, err

core/rawdb/accessors_chain.go

@@ -832,10 +832,10 @@ func WriteAncientBlocksWithBlobs(db ethdb.AncientStore, blocks []*types.Block, r
 
 	// do some sanity check
 	if len(blocks) != len(blobs) {
-		return 0, fmt.Errorf("the blobs len is different with blobks, %v:%v", len(blobs), len(blocks))
+		return 0, fmt.Errorf("the blobs len is different with blocks, %v:%v", len(blobs), len(blocks))
 	}
 	if len(blocks) != len(receipts) {
-		return 0, fmt.Errorf("the receipts len is different with blobks, %v:%v", len(receipts), len(blocks))
+		return 0, fmt.Errorf("the receipts len is different with blocks, %v:%v", len(receipts), len(blocks))
 	}
 	// try reset empty blob ancient table
 	if err := ResetEmptyBlobAncientTable(db, blocks[0].NumberU64()); err != nil {