Bundle submission rework (#1966)

* Contract changes.

* Synchronization rework.

* clear linter issues.

* Abi bindings regenerated.

* Log level change.

* Addressed review comments.

* Removed local network test.

---------

Co-authored-by: StefanIliev545 <[email protected]>

contracts/src/management/ManagementContract.sol

@@ -61,6 +61,7 @@ contract ManagementContract is Initializable, OwnableUpgradeable {
     function initialize() public initializer {
         __Ownable_init(msg.sender);
         lastBatchSeqNo = 0;
+        rollups.nextFreeSequenceNumber = 0; //redundant as the default is 0, but for clarity
         merkleMessageBus = new MerkleTreeMessageBus.MerkleTreeMessageBus();
         messageBus = MessageBus.IMessageBus(address(merkleMessageBus));
 
@@ -72,14 +73,36 @@ contract ManagementContract is Initializable, OwnableUpgradeable {
         return (rol.Hash == rollupHash , rol);
     }
 
+    function GetRollupByNumber(uint256 number) view public returns(bool, Structs.MetaRollup memory) {
+        bytes32 hash = rollups.byOrder[number];
+        if (hash == 0x0) { // ensure we don't try to get rollup for hash zero as that would not pull anything, but the hash would match and return true
+            return (false, Structs.MetaRollup(0x0, "", 0));
+        }
+
+        return GetRollupByHash(hash);
+    }
+
+    function GetUniqueForkID(uint256 number) view public returns(bytes32, Structs.MetaRollup memory) {
+        (bool success, Structs.MetaRollup memory rollup) = GetRollupByNumber(number);
+        if (!success) {
+            return (0x0, rollup);
+        }
+
+        return (rollups.toUniqueForkID[number], rollup);
+    }
+
     function AppendRollup(Structs.MetaRollup calldata _r) internal {
         rollups.byHash[_r.Hash] = _r;
+        rollups.byOrder[rollups.nextFreeSequenceNumber] = _r.Hash;
+        rollups.toUniqueForkID[rollups.nextFreeSequenceNumber] = keccak256(abi.encode(_r.Hash, blockhash(block.number-1)));
+        rollups.nextFreeSequenceNumber++;
+
         if (_r.LastSequenceNumber > lastBatchSeqNo) {
             lastBatchSeqNo = _r.LastSequenceNumber;
         }
     }
 
-    function addCrossChainMessagesRoot(bytes32 _lastBatchHash, bytes32 blockHash, uint256 blockNum, bytes[] memory crossChainHashes, bytes calldata signature) external {
+    function addCrossChainMessagesRoot(bytes32 _lastBatchHash, bytes32 blockHash, uint256 blockNum, bytes[] memory crossChainHashes, bytes calldata signature, uint256 rollupNumber, bytes32 forkID) external {
         if (block.number > blockNum + 255) {
             revert("Block binding too old");
         }
@@ -88,6 +111,10 @@ contract ManagementContract is Initializable, OwnableUpgradeable {
             revert(string(abi.encodePacked("Invalid block binding:", Strings.toString(block.number),":", Strings.toString(uint256(blockHash)), ":", Strings.toString(uint256(blockhash(blockNum))))));
         }
 
+        if (rollups.toUniqueForkID[rollupNumber] != forkID) {
+            revert("Invalid forkID");
+        }
+
         address enclaveID = ECDSA.recover(keccak256(abi.encode(_lastBatchHash, blockHash, blockNum, crossChainHashes)), signature);
         require(attested[enclaveID], "enclaveID not attested"); //todo: only sequencer, rather than everyone who has attested.
 

contracts/src/management/Structs.sol

@@ -12,6 +12,9 @@ interface Structs {
 
     struct RollupStorage {
         mapping(bytes32=>MetaRollup) byHash;
+        mapping(uint256=>bytes32) byOrder;
+        mapping(uint256=>bytes32) toUniqueForkID;
+        uint256 nextFreeSequenceNumber;
     }
 
     struct HeaderCrossChainData {

go/common/errutil/errors_util.go

@@ -25,6 +25,9 @@ var (
 	ErrAncestorBatchNotFound       = errors.New("parent for batch not found")
 	ErrCrossChainBundleRepublished = errors.New("root already added to the message bus")
 	ErrCrossChainBundleNoBatches   = errors.New("no batches for cross chain bundle")
+	ErrNoNextRollup                = errors.New("no next rollup")
+	ErrRollupForkMismatch          = errors.New("rollup fork mismatch")
+	ErrNoBundleToPublish           = errors.New("no bundle to publish")
 )
 
 // BlockRejectError is used as a standard format for error response from enclave for block submission errors

go/common/host/services.go

@@ -22,6 +22,7 @@ const (
 	EnclaveServiceName         = "enclaves"
 	LogSubscriptionServiceName = "log-subs"
 	FilterAPIServiceName       = "filter-api"
+	CrossChainServiceName      = "cross-chain"
 )
 
 // The host has a number of services that encapsulate the various responsibilities of the host.
@@ -109,7 +110,7 @@ type L1Publisher interface {
 	PublishSecretResponse(secretResponse *common.ProducedSecretResponse) error
 
 	// PublishCrossChainBundle will create and publish a cross-chain bundle tx to the management contract
-	PublishCrossChainBundle(bundle *common.ExtCrossChainBundle) error
+	PublishCrossChainBundle(*common.ExtCrossChainBundle, *big.Int, gethcommon.Hash) error
 
 	FetchLatestSeqNo() (*big.Int, error)
 
@@ -119,7 +120,7 @@ type L1Publisher interface {
 	ResyncImportantContracts() error
 
 	// GetBundleRangeFromManagementContract returns the range of batches for which to build a bundle
-	GetBundleRangeFromManagementContract() (*big.Int, *big.Int, error)
+	GetBundleRangeFromManagementContract(lastRollupNumber *big.Int, lastRollupUID gethcommon.Hash) (*gethcommon.Hash, *big.Int, *big.Int, error)
 }
 
 // L2BatchRepository provides an interface for the host to request L2 batch data (live-streaming and historical)

go/host/enclave/guardian.go

@@ -46,6 +46,7 @@ type guardianServiceLocator interface {
 	L1Repo() host.L1BlockRepository
 	L2Repo() host.L2BatchRepository
 	LogSubs() host.LogSubscriptionManager
+	CrossChainMachine() l1.CrossChainStateMachine
 }
 
 // Guardian is a host service which monitors an enclave, it's responsibilities include:
@@ -634,48 +635,24 @@ func (g *Guardian) periodicBundleSubmission() {
 
 	bundleSubmissionTicker := time.NewTicker(interval)
 
-	fromSequenceNumber, _, err := g.sl.L1Publisher().GetBundleRangeFromManagementContract()
-	if err != nil {
-		g.logger.Error(`Unable to get bundle range from management contract and initialize cross chain publishing`, log.ErrKey, err)
-		return
-	}
-
 	for {
 		select {
 		case <-bundleSubmissionTicker.C:
-			from, to, err := g.sl.L1Publisher().GetBundleRangeFromManagementContract()
+			err := g.sl.CrossChainMachine().Synchronize()
 			if err != nil {
-				g.logger.Error("Unable to get bundle range from management contract", log.ErrKey, err)
+				g.logger.Error("Failed to synchronize cross chain state machine", log.ErrKey, err)
 				continue
 			}
 
-			if from.Uint64() > fromSequenceNumber.Uint64() {
-				fromSequenceNumber.Set(from)
-			}
-
-			bundle, err := g.enclaveClient.ExportCrossChainData(context.Background(), fromSequenceNumber.Uint64(), to.Uint64())
+			err = g.sl.CrossChainMachine().PublishNextBundle()
 			if err != nil {
-				if !errors.Is(err, errutil.ErrCrossChainBundleNoBatches) {
-					g.logger.Error("Unable to export cross chain bundle from enclave", log.ErrKey, err)
-				}
-				if errors.Is(err, context.DeadlineExceeded) {
-					g.logger.Error(`Cross chain bundle export timed out.`, log.ErrKey, err)
-					return // stop the process - if we are timing out we are not going to catch up
+				if errors.Is(err, errutil.ErrCrossChainBundleNoBatches) {
+					g.logger.Debug("No batches to publish")
+				} else {
+					g.logger.Error("Failed to publish next bundle", log.ErrKey, err)
 				}
 				continue
 			}
-
-			if len(bundle.CrossChainRootHashes) == 0 {
-				g.logger.Debug("No cross chain data to submit")
-				fromSequenceNumber.SetUint64(to.Uint64() + 1)
-				continue
-			}
-
-			err = g.sl.L1Publisher().PublishCrossChainBundle(bundle)
-			if err != nil {
-				g.logger.Error("Unable to publish cross chain bundle", log.ErrKey, err)
-				continue
-			}
 		case <-g.hostInterrupter.Done():
 			bundleSubmissionTicker.Stop()
 			return

go/host/host.go

@@ -114,10 +114,13 @@ func NewHost(config *config.HostConfig, hostServices *ServicesRegistry, p2p host
 		retryIntervalForL1Receipt,
 		hostStorage,
 	)
+
 	hostServices.RegisterService(hostcommon.L1PublisherName, l1Publisher)
 	hostServices.RegisterService(hostcommon.L2BatchRepositoryName, l2Repo)
 	hostServices.RegisterService(hostcommon.EnclaveServiceName, enclService)
 	hostServices.RegisterService(hostcommon.LogSubscriptionServiceName, subsService)
+	l1StateMachine := l1.NewCrossChainStateMachine(l1Publisher, mgmtContractLib, ethClient, hostServices.Enclaves().GetEnclaveClient(), logger, host.stopControl)
+	hostServices.RegisterService(hostcommon.CrossChainServiceName, l1StateMachine)
 
 	var prof *profiler.Profiler
 	if config.ProfilerEnabled {

go/host/l1/publisher.go

@@ -99,43 +99,46 @@ func (p *Publisher) Start() error {
 	return nil
 }
 
-func (p *Publisher) GetBundleRangeFromManagementContract() (*big.Int, *big.Int, error) {
+func (p *Publisher) GetBundleRangeFromManagementContract(lastRollupNumber *big.Int, lastRollupUID gethcommon.Hash) (*gethcommon.Hash, *big.Int, *big.Int, error) {
 	if p.mgmtContractLib.IsMock() {
-		return nil, nil, fmt.Errorf("bundle publishing unavailable for mocked environments")
+		return nil, nil, nil, fmt.Errorf("bundle publishing unavailable for mocked environments")
 	}
 
 	managementCtr, err := ManagementContract.NewManagementContract(*p.mgmtContractLib.GetContractAddr(), p.ethClient.EthClient())
 	if err != nil {
 		p.logger.Error("Unable to instantiate management contract client")
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 
-	lastBatchHash, err := managementCtr.LastBatchHash(&bind.CallOpts{})
+	hashBytes, rollup, err := managementCtr.GetUniqueForkID(&bind.CallOpts{}, lastRollupNumber)
 	if err != nil {
-		p.logger.Error("Unable to fetch last batch hash from management contract", log.ErrKey, err)
-		return nil, nil, err
+		p.logger.Error("Unable to get unique fork ID from management contract")
+		return nil, nil, nil, err
 	}
 
-	var fromSeqNo *big.Int
-	if lastBatchHash == [32]byte{} {
-		fromSeqNo = big.NewInt(0)
-	} else {
-		batch, err := p.storage.FetchBatch(lastBatchHash)
-		if err != nil {
-			p.logger.Error("Unable to fetch last batch from host db", log.ErrKey, err)
-			return nil, nil, err
-		}
-		fromSeqNo = batch.SeqNo()
-		fromSeqNo = batch.SeqNo().Add(fromSeqNo, big.NewInt(1))
+	rollupUid := gethcommon.BytesToHash(hashBytes[:])
+	if rollupUid != lastRollupUID {
+		return nil, nil, nil, errutil.ErrRollupForkMismatch
 	}
 
-	lastBatchRolledUpSeqNo, err := managementCtr.LastBatchSeqNo(&bind.CallOpts{})
+	fromSeqNo := big.NewInt(0)
+	if lastRollupNumber.Cmp(big.NewInt(0)) != 0 {
+		fromSeqNo = big.NewInt(0).SetUint64(rollup.LastSequenceNumber.Uint64() + 1)
+	}
+
+	nextRollupNumber := big.NewInt(0).SetUint64(lastRollupNumber.Uint64() + 1)
+	nextHashBytes, nextRollup, err := managementCtr.GetUniqueForkID(&bind.CallOpts{}, nextRollupNumber)
 	if err != nil {
-		p.logger.Error("Unable to fetch last batch seq no from management contract", log.ErrKey, err)
-		return nil, nil, err
+		p.logger.Error("Unable to get unique fork ID from management contract")
+		return nil, nil, nil, err
+	}
+
+	nextRollupUID := gethcommon.BytesToHash(nextHashBytes[:])
+	if nextRollupUID.Big().Cmp(gethcommon.Big0) == 0 {
+		return nil, nil, nil, errutil.ErrNoNextRollup
 	}
 
-	return fromSeqNo, lastBatchRolledUpSeqNo, nil
+	return &nextRollupUID, fromSeqNo, nextRollup.LastSequenceNumber, nil
 }
 
 func (p *Publisher) Stop() error {
@@ -281,7 +284,7 @@ func (p *Publisher) PublishRollup(producedRollup *common.ExtRollup) {
 	}
 }
 
-func (p *Publisher) PublishCrossChainBundle(bundle *common.ExtCrossChainBundle) error {
+func (p *Publisher) PublishCrossChainBundle(bundle *common.ExtCrossChainBundle, rollupNum *big.Int, forkID gethcommon.Hash) error {
 	if p.mgmtContractLib.IsMock() {
 		return nil
 	}
@@ -310,7 +313,7 @@ func (p *Publisher) PublishCrossChainBundle(bundle *common.ExtCrossChainBundle)
 
 	transactor.Nonce = big.NewInt(0).SetUint64(nonce)
 
-	tx, err := managementCtr.AddCrossChainMessagesRoot(transactor, [32]byte(bundle.LastBatchHash.Bytes()), bundle.L1BlockHash, bundle.L1BlockNum, bundle.CrossChainRootHashes, bundle.Signature)
+	tx, err := managementCtr.AddCrossChainMessagesRoot(transactor, [32]byte(bundle.LastBatchHash.Bytes()), bundle.L1BlockHash, bundle.L1BlockNum, bundle.CrossChainRootHashes, bundle.Signature, rollupNum, forkID)
 	if err != nil {
 		if !errors.Is(err, errutil.ErrCrossChainBundleRepublished) {
 			p.logger.Error("Error with submitting cross chain bundle transaction.", log.ErrKey, err, log.BundleHashKey, bundle.LastBatchHash)