feat(taiko-client): enable proof aggregation (batch proofs)

.github/workflows/taiko-client--docker.yml

@@ -2,7 +2,7 @@ name: "Build and Push Multi-Arch Docker Image"
 
 on:
   push:
-    branches: [main]
+    branches: [feature/batch-proof]
     tags:
       - "taiko-client-v*"
     paths:

packages/taiko-client/bindings/encoding/input.go

@@ -294,6 +294,13 @@ var (
 		{Name: "TaikoData.Transition", Type: transitionComponentsType},
 		{Name: "TaikoData.TierProof", Type: tierProofComponentsType},
 	}
+	proveBlocksInputArgs = abi.Arguments{
+		{Name: "TaikoData.BlockMetadata", Type: blockMetadataV2ComponentsType},
+		{Name: "TaikoData.Transition", Type: transitionComponentsType},
+	}
+	proveBlocksBatchProofArgs = abi.Arguments{
+		{Name: "TaikoData.TierProof", Type: tierProofComponentsType},
+	}
 )
 
 // Contract ABIs.
@@ -423,6 +430,43 @@ func EncodeProveBlockInput(
 	return b, nil
 }
 
+// EncodeProveBlocksInput performs the solidity `abi.encode` for the given TaikoL1.proveBlocks input.
+func EncodeProveBlocksInput(
+	metas []metadata.TaikoBlockMetaData,
+	transitions []bindings.TaikoDataTransition,
+) ([][]byte, error) {
+	if len(metas) != len(transitions) {
+		return nil, fmt.Errorf("both arrays of TaikoBlockMetaData and TaikoDataTransition must be equal in length")
+	}
+	b := make([][]byte, 0, len(metas))
+	for i := range metas {
+		input, err := proveBlocksInputArgs.Pack(
+			metas[i].(*metadata.TaikoDataBlockMetadataOntake).InnerMetadata(),
+			transitions[i],
+		)
+		if err != nil {
+			return nil, fmt.Errorf("failed to abi.encode TaikoL1.proveBlocks input item after ontake fork, %w", err)
+		}
+
+		b = append(b, input)
+	}
+
+	return b, nil
+}
+
+// EncodeProveBlocksBatchProof performs the solidity `abi.encode` for the given TaikoL1.proveBlocks batchProof.
+func EncodeProveBlocksBatchProof(
+	tierProof *bindings.TaikoDataTierProof,
+) ([]byte, error) {
+	input, err := proveBlocksBatchProofArgs.Pack(
+		tierProof,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to abi.encode TaikoL1.proveBlocks input item after ontake fork, %w", err)
+	}
+	return input, nil
+}
+
 // UnpackTxListBytes unpacks the input data of a TaikoL1.proposeBlock transaction, and returns the txList bytes.
 func UnpackTxListBytes(txData []byte) ([]byte, error) {
 	method, err := TaikoL1ABI.MethodById(txData)

packages/taiko-client/cmd/flags/prover.go

@@ -195,6 +195,31 @@ var (
 		Category: proverCategory,
 		EnvVars:  []string{"PROVER_BLOCK_CONFIRMATIONS"},
 	}
+	// Batch proof related flag
+	SGXBatchSize = &cli.Uint64Flag{
+		Name: "prover.sgx.batchSize",
+		Usage: "The default size of batch sgx proofs, when it arrives, submit a batch of proof immediately, " +
+			"this flag only works post Ontake fork",
+		Value:    1,
+		Category: proverCategory,
+		EnvVars:  []string{"PROVER_SGX_BATCH_SIZE"},
+	}
+	ZKVMBatchSize = &cli.Uint64Flag{
+		Name: "prover.zkvm.batchSize",
+		Usage: "The size of batch ZKVM proof, when it arrives, submit a batch of proof immediately, " +
+			"this flag only works post Ontake fork",
+		Value:    1,
+		Category: proverCategory,
+		EnvVars:  []string{"PROVER_ZKVM_BATCH_SIZE"},
+	}
+	ForceProveInterval = &cli.DurationFlag{
+		Name: "prover.forceBatchProvingInterval",
+		Usage: "Time interval to prove blocks even the number of pending proof do not exceed prover.batchSize, " +
+			"this flag only works post Ontake fork",
+		Category: proverCategory,
+		Value:    30 * time.Minute,
+		EnvVars:  []string{"PROVER_FORCE_BATCH_PROVING_INTERVAL"},
+	}
 )
 
 // ProverFlags All prover flags.
@@ -227,4 +252,7 @@ var ProverFlags = MergeFlags(CommonFlags, []cli.Flag{
 	BlockConfirmations,
 	RaikoRequestTimeout,
 	RaikoZKVMHostEndpoint,
+	SGXBatchSize,
+	ZKVMBatchSize,
+	ForceProveInterval,
 }, TxmgrFlags)

packages/taiko-client/internal/metrics/metrics.go

@@ -47,15 +47,27 @@ var (
 	ProverSubmissionErrorCounter = factory.NewCounter(prometheus.CounterOpts{
 		Name: "prover_proof_submission_error",
 	})
+	ProverAggregationSubmissionErrorCounter = factory.NewCounter(prometheus.CounterOpts{
+		Name: "prover_proof_aggregation_submission_error",
+	})
 	ProverSgxProofGeneratedCounter = factory.NewCounter(prometheus.CounterOpts{
 		Name: "prover_proof_sgx_generated",
 	})
+	ProverSgxProofAggregationGeneratedCounter = factory.NewCounter(prometheus.CounterOpts{
+		Name: "prover_proof_sgx_aggregation_generated",
+	})
 	ProverR0ProofGeneratedCounter = factory.NewCounter(prometheus.CounterOpts{
 		Name: "prover_proof_r0_generated",
 	})
+	ProverR0ProofAggregationGeneratedCounter = factory.NewCounter(prometheus.CounterOpts{
+		Name: "prover_proof_r0_aggregation_generated",
+	})
 	ProverSp1ProofGeneratedCounter = factory.NewCounter(prometheus.CounterOpts{
 		Name: "prover_proof_sp1_generated",
 	})
+	ProverSp1ProofAggregationGeneratedCounter = factory.NewCounter(prometheus.CounterOpts{
+		Name: "prover_proof_sp1_aggregation_generated",
+	})
 	ProverSubmissionRevertedCounter = factory.NewCounter(prometheus.CounterOpts{
 		Name: "prover_proof_submission_reverted",
 	})

packages/taiko-client/pkg/rpc/utils.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/taikoxyz/taiko-mono/packages/taiko-client/bindings"
 	"github.com/taikoxyz/taiko-mono/packages/taiko-client/bindings/encoding"
@@ -62,7 +63,6 @@ func GetProtocolStateVariables(
 	}
 	opts.Context, cancel = CtxWithTimeoutOrDefault(opts.Context, defaultTimeout)
 	defer cancel()
-
 	// Notice: sloB.LastProposedIn and slotB.LastUnpausedAt are always 0
 	// before upgrading contract, but we can ignore it since we won't use it.
 	slotA, slotB, err := taikoL1Client.GetStateVariables(opts)
@@ -248,6 +248,126 @@ func GetBlockProofStatus(
 	}, nil
 }
 
+// BatchGetBlocksProofStatus checks whether the batch of L2 blocks still need new proofs or new contests.
+// Here are the possible status:
+// 1. No proof on chain at all.
+// 2. A valid proof has been submitted.
+// 3. An invalid proof has been submitted, and there is no valid contest.
+// 4. An invalid proof has been submitted, and there is a valid contest.
+func BatchGetBlocksProofStatus(
+	ctx context.Context,
+	cli *Client,
+	ids []*big.Int,
+	proverAddress common.Address,
+	proverSetAddress common.Address,
+) ([]*BlockProofStatus, error) {
+	ctxWithTimeout, cancel := CtxWithTimeoutOrDefault(ctx, defaultTimeout)
+	defer cancel()
+	var (
+		parentHashes = make([][32]byte, len(ids))
+		parents      = make([]*types.Header, len(ids))
+		blockIDs     = make([]uint64, len(ids))
+		result       = make([]*BlockProofStatus, len(ids))
+	)
+	// Get the local L2 parent header.
+	g, gCtx := errgroup.WithContext(ctxWithTimeout)
+	for i, id := range ids {
+		g.Go(func() error {
+			parent, err := cli.L2.HeaderByNumber(gCtx, new(big.Int).Sub(id, common.Big1))
+			if err != nil {
+				return err
+			}
+			parentHashes[i] = parent.Hash()
+			parents[i] = parent
+			blockIDs[i] = id.Uint64()
+			return nil
+		})
+	}
+	if gErr := g.Wait(); gErr != nil {
+		return nil, gErr
+	}
+
+	// Get the transition state from TaikoL1 contract.
+	transitions, err := cli.TaikoL1.GetTransitions(
+		&bind.CallOpts{Context: ctxWithTimeout},
+		blockIDs,
+		parentHashes,
+	)
+	if err != nil {
+		return nil, err
+	}
+	g, gCtx = errgroup.WithContext(ctxWithTimeout)
+	for i, transition := range transitions {
+		// No proof on chain
+		if transition.BlockHash == (common.Hash{}) {
+			result[i] = &BlockProofStatus{IsSubmitted: false, ParentHeader: parents[i]}
+			continue
+		}
+		g.Go(func() error {
+			header, err := cli.WaitL2Header(gCtx, ids[i])
+			if err != nil {
+				return err
+			}
+			if header.Hash() != transition.BlockHash ||
+				(transition.StateRoot != (common.Hash{}) && transition.StateRoot != header.Root) {
+				log.Info(
+					"Different block hash or state root detected, try submitting a contest",
+					"localBlockHash", header.Hash(),
+					"protocolTransitionBlockHash", common.BytesToHash(transition.BlockHash[:]),
+					"localStateRoot", header.Root,
+					"protocolTransitionStateRoot", common.BytesToHash(transition.StateRoot[:]),
+				)
+				result[i] = &BlockProofStatus{
+					IsSubmitted:            true,
+					Invalid:                true,
+					CurrentTransitionState: &transitions[i],
+					ParentHeader:           parents[i],
+				}
+				return nil
+			}
+
+			if proverAddress == transition.Prover ||
+				(proverSetAddress != ZeroAddress && transition.Prover == proverSetAddress) {
+				log.Info(
+					"📬 Block's proof has already been submitted by current prover",
+					"blockID", ids[i],
+					"parent", parents[i].Hash().Hex(),
+					"hash", common.Bytes2Hex(transition.BlockHash[:]),
+					"stateRoot", common.Bytes2Hex(transition.StateRoot[:]),
+					"timestamp", transition.Timestamp,
+					"contester", transition.Contester,
+				)
+				result[i] = &BlockProofStatus{
+					IsSubmitted:            true,
+					Invalid:                transition.Contester != ZeroAddress,
+					ParentHeader:           parents[i],
+					CurrentTransitionState: &transitions[i],
+				}
+				return nil
+			}
+			log.Info(
+				"📬 Block's proof has already been submitted by another prover",
+				"blockID", ids[i],
+				"prover", transition.Prover,
+				"parent", parents[i].Hash().Hex(),
+				"hash", common.Bytes2Hex(transition.BlockHash[:]),
+				"stateRoot", common.Bytes2Hex(transition.StateRoot[:]),
+				"timestamp", transition.Timestamp,
+				"contester", transition.Contester,
+			)
+
+			result[i] = &BlockProofStatus{
+				IsSubmitted:            true,
+				Invalid:                transition.Contester != ZeroAddress,
+				ParentHeader:           parents[i],
+				CurrentTransitionState: &transitions[i],
+			}
+			return nil
+		})
+	}
+	return result, g.Wait()
+}
+
 // SetHead makes a `debug_setHead` RPC call to set the chain's head, should only be used
 // for testing purpose.
 func SetHead(ctx context.Context, client *EthClient, headNum *big.Int) error {

packages/taiko-client/prover/config.go

@@ -60,6 +60,9 @@ type Config struct {
 	BlockConfirmations                      uint64
 	TxmgrConfigs                            *txmgr.CLIConfig
 	PrivateTxmgrConfigs                     *txmgr.CLIConfig
+	SGXProofBufferSize                      uint64
+	ZKVMProofBufferSize                     uint64
+	ForceProveInterval                      time.Duration
 }
 
 // NewConfigFromCliContext creates a new config instance from command line flags.
@@ -183,5 +186,8 @@ func NewConfigFromCliContext(c *cli.Context) (*Config, error) {
 			l1ProverPrivKey,
 			c,
 		),
+		SGXProofBufferSize:  c.Uint64(flags.SGXBatchSize.Name),
+		ZKVMProofBufferSize: c.Uint64(flags.ZKVMBatchSize.Name),
+		ForceProveInterval:  c.Duration(flags.ForceProveInterval.Name),
 	}, nil
 }

packages/taiko-client/prover/init.go

@@ -98,9 +98,10 @@ func (p *Prover) initProofSubmitters(
 ) error {
 	for _, tier := range p.sharedState.GetTiers() {
 		var (
-			producer  proofProducer.ProofProducer
-			submitter proofSubmitter.Submitter
-			err       error
+			bufferSize = p.cfg.SGXProofBufferSize
+			producer   proofProducer.ProofProducer
+			submitter  proofSubmitter.Submitter
+			err        error
 		)
 		switch tier.ID {
 		case encoding.TierOptimisticID:
@@ -121,6 +122,7 @@ func (p *Prover) initProofSubmitters(
 				Dummy:               p.cfg.Dummy,
 				RaikoRequestTimeout: p.cfg.RaikoRequestTimeout,
 			}
+			bufferSize = p.cfg.ZKVMProofBufferSize
 		case encoding.TierZkVMSp1ID:
 			producer = &proofProducer.ZKvmProofProducer{
 				ZKProofType:         proofProducer.ZKProofTypeSP1,
@@ -129,10 +131,13 @@ func (p *Prover) initProofSubmitters(
 				Dummy:               p.cfg.Dummy,
 				RaikoRequestTimeout: p.cfg.RaikoRequestTimeout,
 			}
+			bufferSize = p.cfg.ZKVMProofBufferSize
 		case encoding.TierGuardianMinorityID:
 			producer = proofProducer.NewGuardianProofProducer(encoding.TierGuardianMinorityID, p.cfg.EnableLivenessBondProof)
+			bufferSize = 0
 		case encoding.TierGuardianMajorityID:
 			producer = proofProducer.NewGuardianProofProducer(encoding.TierGuardianMajorityID, p.cfg.EnableLivenessBondProof)
+			bufferSize = 0
 		default:
 			return fmt.Errorf("unsupported tier: %d", tier.ID)
 		}
@@ -141,6 +146,8 @@ func (p *Prover) initProofSubmitters(
 			p.rpc,
 			producer,
 			p.proofGenerationCh,
+			p.batchProofGenerationCh,
+			p.aggregationNotify,
 			p.cfg.ProverSetAddress,
 			p.cfg.TaikoL2Address,
 			p.cfg.Graffiti,
@@ -151,6 +158,7 @@ func (p *Prover) initProofSubmitters(
 			tiers,
 			p.IsGuardianProver(),
 			p.cfg.GuardianProofSubmissionDelay,
+			bufferSize,
 		); err != nil {
 			return err
 		}

packages/taiko-client/prover/proof_producer/dummy_producer.go

@@ -31,3 +31,15 @@ func (o *DummyProofProducer) RequestProof(
 		Tier:    tier,
 	}, nil
 }
+
+// RequestBatchProofs returns a dummy proof aggregation to the result channel.
+func (o *DummyProofProducer) RequestBatchProofs(
+	proofs []*ProofWithHeader,
+	tier uint16,
+) (*BatchProofs, error) {
+	return &BatchProofs{
+		Proofs:     proofs,
+		BatchProof: bytes.Repeat([]byte{0xbb}, 100),
+		Tier:       tier,
+	}, nil
+}

packages/taiko-client/prover/proof_producer/guardian_producer.go

@@ -61,13 +61,23 @@ func (g *GuardianProofProducer) RequestProof(
 	return g.DummyProofProducer.RequestProof(opts, blockID, meta, header, g.Tier(), requestAt)
 }
 
+// RequestCancel implements the ProofProducer interface to cancel the proof generating progress.
 func (g *GuardianProofProducer) RequestCancel(
 	_ context.Context,
 	_ *ProofRequestOptions,
 ) error {
 	return nil
 }
 
+// Aggregate implements the ProofProducer interface to aggregate a batch of proofs.
+func (g *GuardianProofProducer) Aggregate(
+	_ context.Context,
+	_ []*ProofWithHeader,
+	_ time.Time,
+) (*BatchProofs, error) {
+	return nil, nil
+}
+
 // Tier implements the ProofProducer interface.
 func (g *GuardianProofProducer) Tier() uint16 {
 	return g.tier