feat(prover): cancel proof if it becomes verified

metrics/metrics.go

@@ -40,6 +40,7 @@ var (
 	ProverSentValidProofCounter       = metrics.NewRegisteredCounter("prover/proof/valid/sent", nil)
 	ProverSentInvalidProofCounter     = metrics.NewRegisteredCounter("prover/proof/invalid/sent", nil)
 	ProverReceivedProposedBlockGauge  = metrics.NewRegisteredGauge("prover/proposed/received", nil)
+	ProverReceivedProvenBlockGauge    = metrics.NewRegisteredGauge("prover/proven/received", nil)
 )
 
 // Serve starts the metrics server on the given address, will be closed when the given

prover/proof_producer/dummy_producer.go

@@ -61,3 +61,8 @@ func (d *DummyProofProducer) proofDelay() time.Duration {
 
 	return delay
 }
+
+// Cancel cancels an existing proof generation.
+func (d *DummyProofProducer) Cancel(ctx context.Context, blockID *big.Int) error {
+	return nil
+}

prover/proof_producer/oracle_proof_producer.go

@@ -3,6 +3,7 @@ package producer
 import (
 	"context"
 	"crypto/ecdsa"
+	"errors"
 	"fmt"
 	"math/big"
 	"time"
@@ -17,6 +18,10 @@ import (
 	anchorTxValidator "github.com/taikoxyz/taiko-client/prover/anchor_tx_validator"
 )
 
+var (
+	errProtocolAddressMismatch = errors.New("oracle prover private key does not match protocol setting")
+)
+
 // OracleProducer is responsible for generating a fake "zkproof" consisting
 // of a signature of the evidence.
 type OracleProducer struct {
@@ -32,7 +37,13 @@ func NewOracleProducer(
 	proverPrivKey *ecdsa.PrivateKey,
 	taikoL2Address common.Address,
 	proofTimeTarget time.Duration,
+	protocolOracleProverAddress common.Address,
 ) (*OracleProducer, error) {
+	proverAddress := crypto.PubkeyToAddress(proverPrivKey.PublicKey)
+	if proverAddress != protocolOracleProverAddress {
+		return nil, errProtocolAddressMismatch
+	}
+
 	anchorValidator, err := anchorTxValidator.New(taikoL2Address, rpc.L2ChainID, rpc)
 	if err != nil {
 		return nil, err
@@ -138,3 +149,9 @@ func hashAndSignForOracleProof(
 
 	return sig, v, nil
 }
+
+// Cancel cancels an existing proof generation.
+// Since Oracle proofs are not "real" proofs, there is nothing to cancel.
+func (d *OracleProducer) Cancel(ctx context.Context, blockID *big.Int) error {
+	return nil
+}

prover/proof_producer/proof_producer.go

@@ -42,6 +42,7 @@ type ProofProducer interface {
 		header *types.Header,
 		resultCh chan *ProofWithHeader,
 	) error
+	Cancel(ctx context.Context, blockID *big.Int) error
 }
 
 func DegreeToCircuitsIdx(degree uint64) (uint16, error) {

prover/proof_producer/zkevm_cmd_producer.go

@@ -154,3 +154,11 @@ func (d *ZkevmCmdProducer) outputToCalldata(output *ProverCmdOutput) []byte {
 
 	return calldata[:bufLen]
 }
+
+// Cancel cancels an existing proof generation.
+// Right now, it is just a stub that does nothing, because it is not possible to cnacel the proof
+// with the current zkevm software.
+func (d *ZkevmCmdProducer) Cancel(ctx context.Context, blockID *big.Int) error {
+	log.Info("Cancel proof generation for block ", "blockId", blockID)
+	return nil
+}

prover/proof_producer/zkevm_rpcd_producer.go

@@ -210,3 +210,11 @@ func (d *ZkevmRpcdProducer) requestProof(opts *ProofRequestOptions) (*RpcdOutput
 
 	return output.Result, nil
 }
+
+// Cancel cancels an existing proof generation.
+// Right now, it is just a stub that does nothing, because it is not possible to cnacel the proof
+// with the current zkevm software.
+func (d *ZkevmRpcdProducer) Cancel(ctx context.Context, blockID *big.Int) error {
+	log.Info("Cancel proof generation for block ", "blockId", blockID)
+	return nil
+}

prover/proof_submitter/interface.go

@@ -2,6 +2,7 @@ package submitter
 
 import (
 	"context"
+	"math/big"
 
 	"github.com/taikoxyz/taiko-client/bindings"
 	proofProducer "github.com/taikoxyz/taiko-client/prover/proof_producer"
@@ -10,4 +11,5 @@ import (
 type ProofSubmitter interface {
 	RequestProof(ctx context.Context, event *bindings.TaikoL1ClientBlockProposed) error
 	SubmitProof(ctx context.Context, proofWithHeader *proofProducer.ProofWithHeader) error
+	CancelProof(ctx context.Context, blockID *big.Int) error
 }

prover/proof_submitter/valid_proof_submitter.go

@@ -5,6 +5,7 @@ import (
 	"crypto/ecdsa"
 	"errors"
 	"fmt"
+	"math/big"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -231,3 +232,10 @@ func (s *ValidProofSubmitter) SubmitProof(
 
 	return nil
 }
+
+// CancelProof cancels an existing proof generation.
+// Right now, it is just a stub that does nothing, because it is not possible to cnacel the proof
+// with the current zkevm software.
+func (s *ValidProofSubmitter) CancelProof(ctx context.Context, blockID *big.Int) error {
+	return s.proofProducer.Cancel(ctx, blockID)
+}

prover/proof_submitter/valid_proof_submitter_test.go

@@ -116,6 +116,20 @@ func (s *ProofSubmitterTestSuite) TestValidSubmitProofs() {
 	}
 }
 
+func (s *ProofSubmitterTestSuite) TestValidProofSubmitterRequestProofCancelled() {
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		time.AfterFunc(2*time.Second, func() {
+			cancel()
+		})
+	}()
+
+	s.ErrorContains(
+		s.validProofSubmitter.RequestProof(
+			ctx, &bindings.TaikoL1ClientBlockProposed{Id: common.Big256}), "context canceled",
+	)
+}
+
 func TestProofSubmitterTestSuite(t *testing.T) {
 	suite.Run(t, new(ProofSubmitterTestSuite))
 }

prover/prover.go

@@ -24,11 +24,14 @@ import (
 	"github.com/urfave/cli/v2"
 )
 
+type cancelFunc func()
+
 // Prover keep trying to prove new proposed blocks valid/invalid.
 type Prover struct {
 	// Configurations
-	cfg           *Config
-	proverAddress common.Address
+	cfg                 *Config
+	proverAddress       common.Address
+	oracleProverAddress common.Address
 
 	// Clients
 	rpc *rpc.Client
@@ -48,6 +51,8 @@ type Prover struct {
 	// Subscriptions
 	blockProposedCh  chan *bindings.TaikoL1ClientBlockProposed
 	blockProposedSub event.Subscription
+	blockProvenCh    chan *bindings.TaikoL1ClientBlockProven
+	blockProvenSub   event.Subscription
 	blockVerifiedCh  chan *bindings.TaikoL1ClientBlockVerified
 	blockVerifiedSub event.Subscription
 	proveNotify      chan struct{}
@@ -61,6 +66,9 @@ type Prover struct {
 	submitProofConcurrencyGuard chan struct{}
 	submitProofTxMutex          *sync.Mutex
 
+	currentBlocksBeingProven      map[uint64]cancelFunc
+	currentBlocksBeingProvenMutex *sync.Mutex
+
 	ctx context.Context
 	wg  sync.WaitGroup
 }
@@ -79,6 +87,8 @@ func (p *Prover) InitFromCli(ctx context.Context, c *cli.Context) error {
 func InitFromConfig(ctx context.Context, p *Prover, cfg *Config) (err error) {
 	p.cfg = cfg
 	p.ctx = ctx
+	p.currentBlocksBeingProven = make(map[uint64]cancelFunc)
+	p.currentBlocksBeingProvenMutex = &sync.Mutex{}
 
 	// Clients
 	if p.rpc, err = rpc.NewClient(p.ctx, &rpc.ClientConfig{
@@ -112,6 +122,7 @@ func InitFromConfig(ctx context.Context, p *Prover, cfg *Config) (err error) {
 	chBufferSize := p.protocolConfigs.MaxNumProposedBlocks.Uint64()
 	p.blockProposedCh = make(chan *bindings.TaikoL1ClientBlockProposed, chBufferSize)
 	p.blockVerifiedCh = make(chan *bindings.TaikoL1ClientBlockVerified, chBufferSize)
+	p.blockProvenCh = make(chan *bindings.TaikoL1ClientBlockProven, chBufferSize)
 	p.proveValidProofCh = make(chan *proofProducer.ProofWithHeader, chBufferSize)
 	p.proveInvalidProofCh = make(chan *proofProducer.ProofWithHeader, chBufferSize)
 	p.proveNotify = make(chan struct{}, 1)
@@ -123,6 +134,13 @@ func InitFromConfig(ctx context.Context, p *Prover, cfg *Config) (err error) {
 	p.proposeConcurrencyGuard = make(chan struct{}, cfg.MaxConcurrentProvingJobs)
 	p.submitProofConcurrencyGuard = make(chan struct{}, cfg.MaxConcurrentProvingJobs)
 
+	oracleProverAddress, err := p.rpc.TaikoL1.GetOracleProver(nil)
+	if err != nil {
+		return err
+	}
+
+	p.oracleProverAddress = oracleProverAddress
+
 	var producer proofProducer.ProofProducer
 
 	if cfg.OracleProver {
@@ -131,6 +149,7 @@ func InitFromConfig(ctx context.Context, p *Prover, cfg *Config) (err error) {
 			p.cfg.L1ProverPrivKey,
 			p.cfg.TaikoL2Address,
 			time.Duration(p.protocolConfigs.ProofTimeTarget)*time.Second,
+			oracleProverAddress,
 		); err != nil {
 			return err
 		}
@@ -219,6 +238,10 @@ func (p *Prover) eventLoop() {
 			if err := p.onBlockVerified(p.ctx, e); err != nil {
 				log.Error("Handle BlockVerified event error", "error", err)
 			}
+		case e := <-p.blockProvenCh:
+			if err := p.onBlockProven(p.ctx, e); err != nil {
+				log.Error("Handle BlockProven event error", "error", err)
+			}
 		case <-forceProvingTicker.C:
 			reqProving()
 		}
@@ -287,6 +310,16 @@ func (p *Prover) onBlockProposed(
 			return nil
 		}
 
+		ctx, cancelCtx := context.WithCancel(ctx)
+		p.currentBlocksBeingProvenMutex.Lock()
+		p.currentBlocksBeingProven[event.Id.Uint64()] = cancelFunc(func() {
+			defer cancelCtx()
+			if err := p.validProofSubmitter.CancelProof(ctx, event.Id); err != nil {
+				log.Error("error cancelling proof", "error", err, "blockID", event.Id)
+			}
+		})
+		p.currentBlocksBeingProvenMutex.Unlock()
+
 		return p.validProofSubmitter.RequestProof(ctx, event)
 	}
 
@@ -297,6 +330,9 @@ func (p *Prover) onBlockProposed(
 
 	go func() {
 		if err := handleBlockProposedEvent(); err != nil {
+			p.currentBlocksBeingProvenMutex.Lock()
+			delete(p.currentBlocksBeingProven, event.Id.Uint64())
+			p.currentBlocksBeingProvenMutex.Unlock()
 			log.Error("Handle new BlockProposed event error", "error", err)
 		}
 	}()
@@ -308,16 +344,21 @@ func (p *Prover) onBlockProposed(
 func (p *Prover) submitProofOp(ctx context.Context, proofWithHeader *proofProducer.ProofWithHeader, isValidProof bool) {
 	p.submitProofConcurrencyGuard <- struct{}{}
 	go func() {
-		defer func() { <-p.submitProofConcurrencyGuard }()
+		defer func() {
+			<-p.submitProofConcurrencyGuard
+			p.currentBlocksBeingProvenMutex.Lock()
+			delete(p.currentBlocksBeingProven, proofWithHeader.Meta.Id)
+			p.currentBlocksBeingProvenMutex.Unlock()
+		}()
 
 		if err := p.validProofSubmitter.SubmitProof(p.ctx, proofWithHeader); err != nil {
 			log.Error("Submit proof error", "isValidProof", isValidProof, "error", err)
 		}
 	}()
 }
 
-// onBlockVerified update the latestVerified block in current state.
-// TODO: cancel the corresponding block's proof generation, if requested before.
+// onBlockVerified update the latestVerified block in current state, and cancels
+// the block being proven if it's verified.
 func (p *Prover) onBlockVerified(ctx context.Context, event *bindings.TaikoL1ClientBlockVerified) error {
 	metrics.ProverLatestVerifiedIDGauge.Update(event.Id.Int64())
 	p.latestVerifiedL1Height = event.Raw.BlockNumber
@@ -328,6 +369,27 @@ func (p *Prover) onBlockVerified(ctx context.Context, event *bindings.TaikoL1Cli
 	}
 
 	log.Info("New verified valid block", "blockID", event.Id, "hash", common.BytesToHash(event.BlockHash[:]))
+
+	// cancel any proofs being generated for this block
+	p.cancelProof(ctx, event.Id.Uint64())
+
+	return nil
+}
+
+// onBlockProven cancels proof generation if the proof is being generated by this prover,
+// and the proof is not the oracle proof address.
+func (p *Prover) onBlockProven(ctx context.Context, event *bindings.TaikoL1ClientBlockProven) error {
+	metrics.ProverReceivedProvenBlockGauge.Update(event.Id.Int64())
+	// if oracle prover, dont cancel proof.
+	if event.Prover == p.oracleProverAddress {
+		return nil
+	}
+
+	// cancel any proofs being generated for this block
+	if err := p.cancelProofIfValid(ctx, event.Id.Uint64(), uint64(event.ParentGasUsed), event.ParentHash); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -430,10 +492,43 @@ func (p *Prover) NeedNewProof(id *big.Int) (bool, error) {
 func (p *Prover) initSubscription() {
 	p.blockProposedSub = rpc.SubscribeBlockProposed(p.rpc.TaikoL1, p.blockProposedCh)
 	p.blockVerifiedSub = rpc.SubscribeBlockVerified(p.rpc.TaikoL1, p.blockVerifiedCh)
+	p.blockProvenSub = rpc.SubscribeBlockProven(p.rpc.TaikoL1, p.blockProvenCh)
 }
 
 // closeSubscription closes all subscriptions.
 func (p *Prover) closeSubscription() {
 	p.blockVerifiedSub.Unsubscribe()
 	p.blockProposedSub.Unsubscribe()
 }
+
+// cancelProofIfValid cancels proof only if the parentGasUsed and parentHash in the proof match what
+// is expected
+func (p *Prover) cancelProofIfValid(
+	ctx context.Context,
+	blockID uint64,
+	parentGasUsed uint64,
+	parentHash common.Hash,
+) error {
+	parent, err := p.rpc.L2ParentByBlockId(ctx, new(big.Int).SetUint64(blockID))
+	if err != nil {
+		return err
+	}
+
+	if parent.GasUsed == parentGasUsed && parent.Hash() == parentHash {
+		p.cancelProof(ctx, blockID)
+	}
+
+	return nil
+}
+
+// cancelProof cancels local proof generation
+func (p *Prover) cancelProof(ctx context.Context, blockID uint64) {
+	p.currentBlocksBeingProvenMutex.Lock()
+	defer p.currentBlocksBeingProvenMutex.Unlock()
+
+	if cancel, ok := p.currentBlocksBeingProven[blockID]; ok {
+		cancel()
+		delete(p.currentBlocksBeingProven, blockID)
+		log.Info("Cancelled proof for ", "blockID", blockID)
+	}
+}