Add validium support for L1 recovery (DAC) (#560)

* Support validium mode in L1 recovery

* Retry data fetching from DA when the response error is 429 (too many requests)

* Unit tests and minor fixes

* Test decode validium batch data

* Revert unrelated changes

* Address CR feedbacks

* Remove unused functions

* Update zk/da/client.go

Co-authored-by: Rachit Sonthalia <[email protected]>

---------

Co-authored-by: Igor Mandrigin <[email protected]>
Co-authored-by: Rachit Sonthalia <[email protected]>

cmd/utils/flags.go

@@ -591,6 +591,11 @@ var (
 		Usage: "Output the payload of the executor, serialised requests stored to disk by batch number",
 		Value: "",
 	}
+	DAUrl = cli.StringFlag{
+		Name:  "zkevm.da-url",
+		Usage: "The URL of the data availability service",
+		Value: "",
+	}
 	DebugNoSync = cli.BoolFlag{
 		Name:  "debug.no-sync",
 		Usage: "Disable syncing",

eth/backend.go

@@ -775,6 +775,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 				contracts.SequencedBatchTopicEtrog,
 				contracts.VerificationTopicPreEtrog,
 				contracts.VerificationTopicEtrog,
+				contracts.VerificationValidiumTopicEtrog,
 			}}
 			l1Contracts = []libcommon.Address{cfg.AddressRollup, cfg.AddressAdmin, cfg.AddressZkevm}
 		}

eth/ethconfig/config_zkevm.go

@@ -45,6 +45,7 @@ type Zk struct {
 	DefaultGasPrice                        uint64
 	MaxGasPrice                            uint64
 	GasPriceFactor                         float64
+	DAUrl                                  string
 
 	RebuildTreeAfter    uint64
 	IncrementTreeAlways bool

turbo/cli/default_flags.go

@@ -216,4 +216,5 @@ var DefaultFlags = []cli.Flag{
 	&utils.DebugStepAfter,
 	&utils.PoolManagerUrl,
 	&utils.DisableVirtualCounters,
+	&utils.DAUrl,
 }

turbo/cli/flags_zkevm.go

@@ -121,6 +121,7 @@ func ApplyFlagsForZkConfig(ctx *cli.Context, cfg *ethconfig.Config) {
 		PoolManagerUrl:                         ctx.String(utils.PoolManagerUrl.Name),
 		DisableVirtualCounters:                 ctx.Bool(utils.DisableVirtualCounters.Name),
 		ExecutorPayloadOutput:                  ctx.String(utils.ExecutorPayloadOutput.Name),
+		DAUrl:                                  ctx.String(utils.DAUrl.Name),
 	}
 
 	checkFlag(utils.L2ChainIdFlag.Name, cfg.L2ChainId)

zk/contracts/l1_contracts.go

@@ -5,11 +5,12 @@ import (
 )
 
 var (
-	SequencedBatchTopicPreEtrog = common.HexToHash("0x303446e6a8cb73c83dff421c0b1d5e5ce0719dab1bff13660fc254e58cc17fce")
-	SequencedBatchTopicEtrog    = common.HexToHash("0x3e54d0825ed78523037d00a81759237eb436ce774bd546993ee67a1b67b6e766")
-	VerificationTopicPreEtrog   = common.HexToHash("0xcb339b570a7f0b25afa7333371ff11192092a0aeace12b671f4c212f2815c6fe")
-	VerificationTopicEtrog      = common.HexToHash("0xd1ec3a1216f08b6eff72e169ceb548b782db18a6614852618d86bb19f3f9b0d3")
-	UpdateL1InfoTreeTopic       = common.HexToHash("0xda61aa7823fcd807e37b95aabcbe17f03a6f3efd514176444dae191d27fd66b3")
-	InitialSequenceBatchesTopic = common.HexToHash("0x060116213bcbf54ca19fd649dc84b59ab2bbd200ab199770e4d923e222a28e7f")
-	SequenceBatchesTopic        = common.HexToHash("0x3e54d0825ed78523037d00a81759237eb436ce774bd546993ee67a1b67b6e766")
+	SequencedBatchTopicPreEtrog    = common.HexToHash("0x303446e6a8cb73c83dff421c0b1d5e5ce0719dab1bff13660fc254e58cc17fce")
+	SequencedBatchTopicEtrog       = common.HexToHash("0x3e54d0825ed78523037d00a81759237eb436ce774bd546993ee67a1b67b6e766")
+	VerificationValidiumTopicEtrog = common.HexToHash("0x9c72852172521097ba7e1482e6b44b351323df0155f97f4ea18fcec28e1f5966")
+	VerificationTopicPreEtrog      = common.HexToHash("0xcb339b570a7f0b25afa7333371ff11192092a0aeace12b671f4c212f2815c6fe")
+	VerificationTopicEtrog         = common.HexToHash("0xd1ec3a1216f08b6eff72e169ceb548b782db18a6614852618d86bb19f3f9b0d3")
+	UpdateL1InfoTreeTopic          = common.HexToHash("0xda61aa7823fcd807e37b95aabcbe17f03a6f3efd514176444dae191d27fd66b3")
+	InitialSequenceBatchesTopic    = common.HexToHash("0x060116213bcbf54ca19fd649dc84b59ab2bbd200ab199770e4d923e222a28e7f")
+	SequenceBatchesTopic           = common.HexToHash("0x3e54d0825ed78523037d00a81759237eb436ce774bd546993ee67a1b67b6e766")
 )

zk/da/client.go

@@ -0,0 +1,43 @@
+package da
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/gateway-fm/cdk-erigon-lib/common"
+	"github.com/ledgerwatch/erigon/common/hexutil"
+
+	"github.com/ledgerwatch/erigon/zkevm/jsonrpc/client"
+)
+
+const maxAttempts = 10
+const retryDelay = 500 * time.Millisecond
+
+func GetOffChainData(ctx context.Context, url string, hash common.Hash) ([]byte, error) {
+	attemp := 0
+
+	for attemp < maxAttempts {
+		response, err := client.JSONRPCCall(url, "sync_getOffChainData", hash)
+
+		if httpErr, ok := err.(*client.HTTPError); ok && httpErr.StatusCode == http.StatusTooManyRequests {
+			time.Sleep(retryDelay)
+			attemp += 1
+			continue
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		if response.Error != nil {
+			return nil, fmt.Errorf("%v %v", response.Error.Code, response.Error.Message)
+		}
+
+		return hexutil.Decode(strings.Trim(string(response.Result), "\""))
+	}
+
+	return nil, fmt.Errorf("max attempts of data fetching reached, attempts: %v, DA url: %s", maxAttempts, url)
+}

zk/da/client_test.go

@@ -0,0 +1,91 @@
+package da
+
+import (
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gateway-fm/cdk-erigon-lib/common"
+	"github.com/ledgerwatch/erigon/zkevm/jsonrpc/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestClient_GetOffChainData(t *testing.T) {
+	tests := []struct {
+		name       string
+		hash       common.Hash
+		result     string
+		data       []byte
+		statusCode int
+		err        string
+	}{
+		{
+			name:   "successfully got offhcain data",
+			hash:   common.BytesToHash([]byte("hash")),
+			result: fmt.Sprintf(`{"result":"0x%s"}`, hex.EncodeToString([]byte("offchaindata"))),
+			data:   []byte("offchaindata"),
+		},
+		{
+			name:   "error returned by server",
+			hash:   common.BytesToHash([]byte("hash")),
+			result: `{"error":{"code":123,"message":"test error"}}`,
+			err:    "123 test error",
+		},
+		{
+			name:   "invalid offchain data returned by server",
+			hash:   common.BytesToHash([]byte("hash")),
+			result: `{"result":"invalid-signature"}`,
+			err:    "hex string without 0x prefix",
+		},
+		{
+			name:       "unsuccessful status code returned by server",
+			hash:       common.BytesToHash([]byte("hash")),
+			statusCode: http.StatusUnauthorized,
+			err:        "invalid status code, expected: 200, found: 401",
+		},
+		{
+			name:       "handle retry on 429",
+			hash:       common.BytesToHash([]byte("hash")),
+			statusCode: http.StatusTooManyRequests,
+			err:        "max attempts of data fetching reached",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				var res types.Request
+				require.NoError(t, json.NewDecoder(r.Body).Decode(&res))
+				require.Equal(t, "sync_getOffChainData", res.Method)
+
+				var params []common.Hash
+				require.NoError(t, json.Unmarshal(res.Params, &params))
+				require.Equal(t, tt.hash, params[0])
+
+				if tt.statusCode > 0 {
+					w.WriteHeader(tt.statusCode)
+				}
+
+				_, err := fmt.Fprint(w, tt.result)
+				require.NoError(t, err)
+			}))
+			defer svr.Close()
+
+			got, err := GetOffChainData(context.Background(), svr.URL, tt.hash)
+			if tt.err != "" {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tt.err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tt.data, got)
+			}
+		})
+	}
+}

zk/l1_data/l1_decoder.go

@@ -1,12 +1,16 @@
 package l1_data
 
 import (
-	"strings"
-	"github.com/ledgerwatch/erigon/zk/contracts"
+	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
+
 	"github.com/gateway-fm/cdk-erigon-lib/common"
 	"github.com/ledgerwatch/erigon/accounts/abi"
+	"github.com/ledgerwatch/erigon/crypto"
+	"github.com/ledgerwatch/erigon/zk/contracts"
+	"github.com/ledgerwatch/erigon/zk/da"
 )
 
 type RollupBaseEtrogBatchData struct {
@@ -16,7 +20,52 @@ type RollupBaseEtrogBatchData struct {
 	ForcedBlockHashL1    [32]byte
 }
 
-func DecodeL1BatchData(txData []byte) ([][]byte, common.Address, error) {
+type ValidiumBatchData struct {
+	TransactionsHash     [32]byte
+	ForcedGlobalExitRoot [32]byte
+	ForcedTimestamp      uint64
+	ForcedBlockHashL1    [32]byte
+}
+
+func BuildSequencesForRollup(data []byte) ([]RollupBaseEtrogBatchData, error) {
+	var sequences []RollupBaseEtrogBatchData
+	err := json.Unmarshal(data, &sequences)
+	return sequences, err
+}
+
+func BuildSequencesForValidium(data []byte, daUrl string) ([]RollupBaseEtrogBatchData, error) {
+	var sequences []RollupBaseEtrogBatchData
+	var validiumSequences []ValidiumBatchData
+	err := json.Unmarshal(data, &validiumSequences)
+
+	if err != nil {
+		return nil, err
+	}
+
+	for _, validiumSequence := range validiumSequences {
+		hash := common.BytesToHash(validiumSequence.TransactionsHash[:])
+		data, err := da.GetOffChainData(context.Background(), daUrl, hash)
+		if err != nil {
+			return nil, err
+		}
+
+		actualTransactionsHash := crypto.Keccak256Hash(data)
+		if actualTransactionsHash != hash {
+			return nil, fmt.Errorf("unable to fetch off chain data for hash %s, got %s intead", hash.String(), actualTransactionsHash.String())
+		}
+
+		sequences = append(sequences, RollupBaseEtrogBatchData{
+			Transactions:         data,
+			ForcedGlobalExitRoot: validiumSequence.ForcedGlobalExitRoot,
+			ForcedTimestamp:      validiumSequence.ForcedTimestamp,
+			ForcedBlockHashL1:    validiumSequence.ForcedBlockHashL1,
+		})
+	}
+
+	return sequences, nil
+}
+
+func DecodeL1BatchData(txData []byte, daUrl string) ([][]byte, common.Address, error) {
 	// we need to know which version of the ABI to use here so lets find it
 	idAsString := fmt.Sprintf("%x", txData[:4])
 	abiMapped, found := contracts.SequenceBatchesMapping[idAsString]
@@ -42,6 +91,8 @@ func DecodeL1BatchData(txData []byte) ([][]byte, common.Address, error) {
 
 	var coinbase common.Address
 
+	isValidium := false
+
 	switch idAsString {
 	case contracts.SequenceBatchesIdv5_0:
 		cb, ok := data[1].(common.Address)
@@ -55,6 +106,16 @@ func DecodeL1BatchData(txData []byte) ([][]byte, common.Address, error) {
 			return nil, common.Address{}, fmt.Errorf("expected position 3 in the l1 call data to be address")
 		}
 		coinbase = cb
+	case contracts.SequenceBatchesValidiumElderBerry:
+		if daUrl == "" {
+			return nil, common.Address{}, fmt.Errorf("data availability url is required for validium")
+		}
+		isValidium = true
+		cb, ok := data[3].(common.Address)
+		if !ok {
+			return nil, common.Address{}, fmt.Errorf("expected position 3 in the l1 call data to be address")
+		}
+		coinbase = cb
 	default:
 		return nil, common.Address{}, fmt.Errorf("unknown l1 call data")
 	}
@@ -65,7 +126,13 @@ func DecodeL1BatchData(txData []byte) ([][]byte, common.Address, error) {
 	if err != nil {
 		return nil, coinbase, err
 	}
-	err = json.Unmarshal(bytedata, &sequences)
+
+	if isValidium {
+		sequences, err = BuildSequencesForValidium(bytedata, daUrl)
+	} else {
+		sequences, err = BuildSequencesForRollup(bytedata)
+	}
+
 	if err != nil {
 		return nil, coinbase, err
 	}

zk/stages/stage_l1syncer.go

@@ -230,6 +230,14 @@ func parseLogType(l1RollupId uint64, log *ethTypes.Log) (l1BatchInfo types.L1Bat
 		batchLogType = logVerify
 		batchNum = new(big.Int).SetBytes(log.Topics[1].Bytes()).Uint64()
 		stateRoot = common.BytesToHash(log.Data[:32])
+	case contracts.VerificationValidiumTopicEtrog:
+		if isRollupIdMatching {
+			batchLogType = logVerify
+			batchNum = new(big.Int).SetBytes(log.Topics[1].Bytes()).Uint64()
+			stateRoot = common.BytesToHash(log.Data[:32])
+		} else {
+			batchLogType = logIncompatible
+		}
 	case contracts.VerificationTopicEtrog:
 		if isRollupIdMatching {
 			batchLogType = logVerify

zk/stages/stage_sequencer_l1_block_sync.go

@@ -5,7 +5,10 @@ import (
 	"fmt"
 	"time"
 
+	"errors"
+
 	"github.com/gateway-fm/cdk-erigon-lib/kv"
+	"github.com/ledgerwatch/erigon/core/types"
 	"github.com/ledgerwatch/erigon/eth/ethconfig"
 	"github.com/ledgerwatch/erigon/eth/stagedsync"
 	"github.com/ledgerwatch/erigon/eth/stagedsync/stages"
@@ -14,8 +17,6 @@ import (
 	"github.com/ledgerwatch/erigon/zk/syncer"
 	zktx "github.com/ledgerwatch/erigon/zk/tx"
 	"github.com/ledgerwatch/log/v3"
-	"github.com/ledgerwatch/erigon/core/types"
-	"errors"
 )
 
 type SequencerL1BlockSyncCfg struct {
@@ -158,7 +159,7 @@ LOOP:
 					return errors.New("l1 info root is not 32 bytes")
 				}
 
-				batches, coinbase, err := l1_data.DecodeL1BatchData(transaction.GetData())
+				batches, coinbase, err := l1_data.DecodeL1BatchData(transaction.GetData(), cfg.zkCfg.DAUrl)
 				if err != nil {
 					return err
 				}

zkevm/jsonrpc/client/client.go

@@ -22,6 +22,15 @@ func NewClient(url string) *Client {
 	}
 }
 
+// HTTPError custom error type for handling HTTP responses
+type HTTPError struct {
+	StatusCode int
+}
+
+func (e *HTTPError) Error() string {
+	return fmt.Sprintf("invalid status code, expected: %d, found: %d", http.StatusOK, e.StatusCode)
+}
+
 // JSONRPCCall executes a 2.0 JSON RPC HTTP Post Request to the provided URL with
 // the provided method and parameters, which is compatible with the Ethereum
 // JSON RPC Server.
@@ -59,7 +68,7 @@ func JSONRPCCall(url, method string, parameters ...interface{}) (types.Response,
 	}
 
 	if httpRes.StatusCode != http.StatusOK {
-		return types.Response{}, fmt.Errorf("Invalid status code, expected: %v, found: %v", http.StatusOK, httpRes.StatusCode)
+		return types.Response{}, &HTTPError{StatusCode: httpRes.StatusCode}
 	}
 
 	resBody, err := io.ReadAll(httpRes.Body)