NubitDA integration

cmd/run_xlayer.go

@@ -9,6 +9,7 @@ import (
 	"github.com/0xPolygonHermez/zkevm-node/config/apollo"
 	"github.com/0xPolygonHermez/zkevm-node/dataavailability"
 	"github.com/0xPolygonHermez/zkevm-node/dataavailability/datacommittee"
+	"github.com/0xPolygonHermez/zkevm-node/dataavailability/nubit"
 	"github.com/0xPolygonHermez/zkevm-node/etherman"
 	"github.com/0xPolygonHermez/zkevm-node/ethtxmanager"
 	"github.com/0xPolygonHermez/zkevm-node/event"
@@ -123,6 +124,22 @@ func newDataAvailability(c config.Config, st *state.State, etherman *etherman.Cl
 		if err != nil {
 			return nil, err
 		}
+	case string(dataavailability.DataAvailabilityNubitDA):
+		var (
+			pk  *ecdsa.PrivateKey
+			err error
+		)
+		if isSequenceSender {
+			_, pk, err = etherman.LoadAuthFromKeyStoreXLayer(c.SequenceSender.DAPermitApiPrivateKey.Path, c.SequenceSender.DAPermitApiPrivateKey.Password)
+			if err != nil {
+				return nil, err
+			}
+			log.Infof("from pk %s", crypto.PubkeyToAddress(pk.PublicKey))
+		}
+		daBackend, err = nubit.NewNubitDABackend(&c.DataAvailability, pk)
+		if err != nil {
+			return nil, err
+		}
 	default:
 		return nil, fmt.Errorf("unexpected / unsupported DA protocol: %s", daProtocolName)
 	}

config/config.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/0xPolygonHermez/zkevm-node/aggregator"
 	"github.com/0xPolygonHermez/zkevm-node/config/types"
+	"github.com/0xPolygonHermez/zkevm-node/dataavailability/nubit"
 	"github.com/0xPolygonHermez/zkevm-node/db"
 	"github.com/0xPolygonHermez/zkevm-node/etherman"
 	"github.com/0xPolygonHermez/zkevm-node/ethtxmanager"
@@ -102,6 +103,8 @@ type Config struct {
 	SequenceSender sequencesender.Config
 	// Configuration of the aggregator service
 	Aggregator aggregator.Config
+	// Configuration of the NubitDA data availability service
+	DataAvailability nubit.Config
 	// Configuration of the genesis of the network. This is used to known the initial state of the network
 	NetworkConfig NetworkConfig
 	// Configuration of the gas price suggester service

config/default.go

@@ -245,6 +245,13 @@ AggLayerTxTimeout = "5m"
 AggLayerURL = ""
 SequencerPrivateKey = {}
 
+[DataAvailability]
+NubitRpcURL = "http://127.0.0.1:26658"
+NubitAuthKey = ""
+NubitNamespace = "xlayer"
+NubitGetProofMaxRetry = "10"
+NubitGetProofWaitPeriod = "5s"
+
 [L2GasPriceSuggester]
 Type = "follower"
 UpdatePeriod = "10s"

dataavailability/config.go

@@ -6,4 +6,6 @@ type DABackendType string
 const (
 	// DataAvailabilityCommittee is the DAC protocol backend
 	DataAvailabilityCommittee DABackendType = "DataAvailabilityCommittee"
+	// DataAvailabilityNubitDA is the NubitDA protocol backend
+	DataAvailabilityNubitDA DABackendType = "NubitDA"
 )

dataavailability/nubit/abi.go

@@ -0,0 +1,28 @@
+package nubit
+
+const blobDataABI = `[
+	{
+		"type": "function",
+		"name": "BlobData",
+		"inputs": [
+			{
+			"name": "blobData",
+			"type": "tuple",
+			"internalType": "struct NubitDAVerifier.BlobData",
+			"components": [
+				{
+				"name": "blobID",
+				"type": "bytes",
+				"internalType": "bytes"
+				},
+				{
+				"name": "signature",
+				"type": "bytes",
+				"internalType": "bytes"
+				}
+			]
+			}
+		],
+		"stateMutability": "pure"
+	}
+]`

dataavailability/nubit/backend.go

@@ -0,0 +1,142 @@
+package nubit
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"encoding/hex"
+	"strings"
+	"time"
+
+	daTypes "github.com/0xPolygon/cdk-data-availability/types"
+	"github.com/0xPolygonHermez/zkevm-node/log"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/rollkit/go-da"
+	"github.com/rollkit/go-da/proxy"
+)
+
+// NubitDABackend implements the DA integration with Nubit DA layer
+type NubitDABackend struct {
+	client     da.DA
+	config     *Config
+	namespace  da.Namespace
+	privKey    *ecdsa.PrivateKey
+	commitTime time.Time
+}
+
+// NewNubitDABackend is the factory method to create a new instance of NubitDABackend
+func NewNubitDABackend(
+	cfg *Config,
+	privKey *ecdsa.PrivateKey,
+) (*NubitDABackend, error) {
+	log.Infof("NubitDABackend config: %#v", cfg)
+	cn, err := proxy.NewClient(cfg.NubitRpcURL, cfg.NubitAuthKey)
+	if err != nil {
+		return nil, err
+	}
+
+	hexStr := hex.EncodeToString([]byte(cfg.NubitNamespace))
+	name, err := hex.DecodeString(strings.Repeat("0", NubitNamespaceBytesLength-len(hexStr)) + hexStr)
+	if err != nil {
+		log.Errorf("error decoding NubitDA namespace config: %+v", err)
+		return nil, err
+	}
+	if err != nil {
+		return nil, err
+	}
+	log.Infof("NubitDABackend namespace: %s", string(name))
+
+	return &NubitDABackend{
+		config:     cfg,
+		privKey:    privKey,
+		namespace:  name,
+		client:     cn,
+		commitTime: time.Now(),
+	}, nil
+}
+
+// Init initializes the NubitDA backend
+func (backend *NubitDABackend) Init() error {
+	return nil
+}
+
+// PostSequence sends the sequence data to the data availability backend, and returns the dataAvailabilityMessage
+// as expected by the contract
+func (backend *NubitDABackend) PostSequence(ctx context.Context, batchesData [][]byte) ([]byte, error) {
+	// Check commit time interval validation
+	lastCommitTime := time.Since(backend.commitTime)
+	if lastCommitTime < NubitMinCommitTime {
+		time.Sleep(NubitMinCommitTime - lastCommitTime)
+	}
+
+	// Encode NubitDA blob data
+	data := EncodeSequence(batchesData)
+	ids, err := backend.client.Submit(ctx, [][]byte{data}, -1, backend.namespace)
+	// Ensure only a single blob ID returned
+	if err != nil || len(ids) != 1 {
+		log.Errorf("Submit batch data with NubitDA client failed: %s", err)
+		return nil, err
+	}
+	blobID := ids[0]
+	backend.commitTime = time.Now()
+	log.Infof("Data submitted to Nubit DA: %d bytes against namespace %v sent with id %#x", len(data), backend.namespace, blobID)
+
+	// Get proof of batches data on NubitDA layer
+	tries := uint64(0)
+	posted := false
+	for tries < backend.config.NubitGetProofMaxRetry {
+		dataProof, err := backend.client.GetProofs(ctx, [][]byte{blobID}, backend.namespace)
+		if err != nil {
+			log.Infof("Proof not available: %s", err)
+		}
+		if len(dataProof) == 1 {
+			// TODO: add data proof to DA message
+			log.Infof("Data proof from Nubit DA received: %+v", dataProof)
+			posted = true
+			break
+		}
+
+		// Retries
+		tries += 1
+		time.Sleep(backend.config.NubitGetProofWaitPeriod.Duration)
+	}
+	if !posted {
+		log.Errorf("Get blob proof on Nubit DA failed: %s", err)
+		return nil, err
+	}
+
+	// Get abi-encoded data availability message
+	sequence := daTypes.Sequence{}
+	for _, seq := range batchesData {
+		sequence = append(sequence, seq)
+	}
+	signedSequence, err := sequence.Sign(backend.privKey)
+	if err != nil {
+		log.Errorf("Failed to sign sequence with pk: %v", err)
+		return nil, err
+	}
+	signature := append(sequence.HashToSign(), signedSequence.Signature...)
+	blobData := BlobData{
+		BlobID:    blobID,
+		Signature: signature,
+	}
+
+	return TryEncodeToDataAvailabilityMessage(blobData)
+}
+
+// GetSequence gets the sequence data from NubitDA layer
+func (backend *NubitDABackend) GetSequence(ctx context.Context, batchHashes []common.Hash, dataAvailabilityMessage []byte) ([][]byte, error) {
+	blobData, err := TryDecodeFromDataAvailabilityMessage(dataAvailabilityMessage)
+	if err != nil {
+		log.Error("Error decoding from da message: ", err)
+		return nil, err
+	}
+
+	reply, err := backend.client.Get(ctx, [][]byte{blobData.BlobID}, backend.namespace)
+	if err != nil || len(reply) != 1 {
+		log.Error("Error retrieving blob from NubitDA client: ", err)
+		return nil, err
+	}
+
+	batchesData, _ := DecodeSequence(reply[0])
+	return batchesData, nil
+}

dataavailability/nubit/backend_test.go

@@ -0,0 +1,141 @@
+package nubit
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	crand "crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/0xPolygonHermez/zkevm-node/config/types"
+	"github.com/0xPolygonHermez/zkevm-node/log"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/rollkit/go-da/proxy"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestOffchainPipeline(t *testing.T) {
+	cfg := Config{
+		NubitRpcURL:             "http://127.0.0.1:26658",
+		NubitAuthKey:            "",
+		NubitNamespace:          "xlayer",
+		NubitGetProofMaxRetry:   10,
+		NubitGetProofWaitPeriod: types.NewDuration(5 * time.Second),
+	}
+	pk, err := ecdsa.GenerateKey(elliptic.P256(), crand.Reader)
+	require.NoError(t, err)
+
+	backend, err := NewNubitDABackend(&cfg, pk)
+	require.NoError(t, err)
+
+	// Generate mock string batch data
+	stringData := "hihihihihihihihihihihihihihihihihihi"
+	data := []byte(stringData)
+
+	// Generate mock string sequence
+	mockBatches := [][]byte{}
+	for i := 0; i < 1; i++ {
+		mockBatches = append(mockBatches, data)
+	}
+
+	msg, err := backend.PostSequence(context.Background(), mockBatches)
+	fmt.Println("DA msg: ", msg)
+	require.NoError(t, err)
+	time.Sleep(600 * time.Millisecond)
+
+	blobData, err := TryDecodeFromDataAvailabilityMessage(msg)
+	require.NoError(t, err)
+	require.NotNil(t, blobData.BlobID)
+	require.NotNil(t, blobData.Signature)
+	require.NotZero(t, len(blobData.BlobID))
+	require.NotZero(t, len(blobData.Signature))
+	fmt.Println("Decoding DA msg successful")
+
+	// Retrieve sequence with provider
+	returnData, err := backend.GetSequence(context.Background(), []common.Hash{}, msg)
+
+	// Validate retrieved data
+	require.NoError(t, err)
+	require.Equal(t, 10, len(returnData))
+	for _, batchData := range returnData {
+		assert.Equal(t, stringData, string(batchData))
+	}
+}
+
+func TestOffchainPipelineWithRandomData(t *testing.T) {
+	cfg := Config{
+		NubitRpcURL:             "http://127.0.0.1:26658",
+		NubitAuthKey:            "",
+		NubitNamespace:          "xlayer",
+		NubitGetProofMaxRetry:   10,
+		NubitGetProofWaitPeriod: types.NewDuration(5 * time.Second),
+	}
+	pk, err := ecdsa.GenerateKey(elliptic.P256(), crand.Reader)
+	require.NoError(t, err)
+
+	backend, err := NewNubitDABackend(&cfg, pk)
+	require.NoError(t, err)
+
+	// Define Different DataSizes
+	dataSize := []int{100000, 200000, 1000, 80, 30000}
+
+	// Disperse Blob with different DataSizes
+	rand.Seed(time.Now().UnixNano())                         //nolint:gosec,staticcheck
+	data := make([]byte, dataSize[rand.Intn(len(dataSize))]) //nolint:gosec,staticcheck
+	_, err = rand.Read(data)                                 //nolint:gosec,staticcheck
+	assert.NoError(t, err)
+
+	// Generate mock string sequence
+	mockBatches := [][]byte{}
+	for i := 0; i < 10; i++ {
+		mockBatches = append(mockBatches, data)
+	}
+
+	msg, err := backend.PostSequence(context.Background(), mockBatches)
+	fmt.Println("DA msg: ", msg)
+	require.NoError(t, err)
+	time.Sleep(600 * time.Millisecond)
+
+	blobData, err := TryDecodeFromDataAvailabilityMessage(msg)
+	require.NoError(t, err)
+	require.NotNil(t, blobData.BlobID)
+	require.NotNil(t, blobData.Signature)
+	require.NotZero(t, len(blobData.BlobID))
+	require.NotZero(t, len(blobData.Signature))
+	fmt.Println("Decoding DA msg successful")
+
+	// Retrieve sequence with provider
+	returnData, err := backend.GetSequence(context.Background(), []common.Hash{}, msg)
+
+	// Validate retrieved data
+	require.NoError(t, err)
+	require.Equal(t, 10, len(returnData))
+	for idx, batchData := range returnData {
+		assert.Equal(t, mockBatches[idx], batchData)
+	}
+}
+
+func NewMockNubitDABackend(url string, authKey string, pk *ecdsa.PrivateKey) (*NubitDABackend, error) {
+	cn, err := proxy.NewClient(url, authKey)
+	if err != nil || cn == nil {
+		return nil, err
+	}
+
+	name, err := hex.DecodeString("xlayer")
+	if err != nil {
+		return nil, err
+	}
+
+	log.Infof("Nubit Namespace: %s ", string(name))
+	return &NubitDABackend{
+		namespace:  name,
+		client:     cn,
+		privKey:    pk,
+		commitTime: time.Now(),
+	}, nil
+}