feat: Celestia DA Layer Client implementation

da/celestia/celestia.go

@@ -0,0 +1,147 @@
+package celestia
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+
+	"github.com/celestiaorg/optimint/da"
+	"github.com/celestiaorg/optimint/libs/cnrc"
+	"github.com/celestiaorg/optimint/log"
+	"github.com/celestiaorg/optimint/store"
+	"github.com/celestiaorg/optimint/types"
+	pb "github.com/celestiaorg/optimint/types/pb/optimint"
+)
+
+// DataAvailabilityLayerClient use celestia-node public API.
+type DataAvailabilityLayerClient struct {
+	client *cnrc.Client
+
+	config Config
+	logger log.Logger
+}
+
+var _ da.DataAvailabilityLayerClient = &DataAvailabilityLayerClient{}
+var _ da.BlockRetriever = &DataAvailabilityLayerClient{}
+
+type Config struct {
+	BaseURL string `json:"base_url"`
+	Timeout time.Duration `json:"timeout"`
+	GasLimit uint64 `json:"gas_limit"`
+	NamespaceID [8]byte `json:"namespace_id"`
+}
+
+func (c *DataAvailabilityLayerClient) Init(config []byte, kvStore store.KVStore, logger log.Logger) error {
+	c.logger = logger
+
+	if len(config) > 0 {
+ return json.Unmarshal(config, &c.config)
+	}
+
+	return nil
+}
+
+func (c *DataAvailabilityLayerClient) Start() error {
+	c.logger.Info("starting Celestia Data Availability Layer Client", "baseURL", c.config.BaseURL)
+	var err error
+	c.client, err = cnrc.NewClient(c.config.BaseURL, cnrc.WithTimeout(c.config.Timeout))
+	return err
+}
+
+func (c *DataAvailabilityLayerClient) Stop() error {
+	c.logger.Info("stopping Celestia Data Availability Layer Client")
+	return nil
+}
+
+func (c *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultSubmitBlock {
+	blob, err := block.MarshalBinary()
+	if err != nil {
+ return da.ResultSubmitBlock{
+ DAResult: da.DAResult{
+ Code: da.StatusError,
+ Message: err.Error(),
+ },
+ }
+	}
+
+	txResponse, err := c.client.SubmitPFD(context.TODO(), c.config.NamespaceID, blob, c.config.GasLimit)
+
+	if err != nil {
+ return da.ResultSubmitBlock{
+ DAResult: da.DAResult{
+ Code: da.StatusError,
+ Message: err.Error(),
+ },
+ }
+	}
+
+	return da.ResultSubmitBlock{
+ DAResult: da.DAResult{
+ Code: da.StatusSuccess,
+ Message: "tx hash: " + txResponse.TxHash,
+ DAHeight: uint64(txResponse.Height),
+ },
+	}
+}
+
+func (c *DataAvailabilityLayerClient) CheckBlockAvailability(dataLayerHeight uint64) da.ResultCheckBlock {
+	shares, err := c.client.NamespacedShares(context.TODO(), c.config.NamespaceID, dataLayerHeight)
+	if err != nil {
+ return da.ResultCheckBlock{
+ DAResult: da.DAResult{
+ Code: da.StatusError,
+ Message: err.Error(),
+ },
+ }
+	}
+
+	return da.ResultCheckBlock{
+ DAResult: da.DAResult{
+ Code: da.StatusSuccess,
+ DAHeight: dataLayerHeight,
+ },
+ DataAvailable: len(shares) > 0,
+	}
+}
+
+func (c *DataAvailabilityLayerClient) RetrieveBlocks(dataLayerHeight uint64) da.ResultRetrieveBlocks {
+	data, err := c.client.NamespacedData(context.TODO(), c.config.NamespaceID, dataLayerHeight)
+	if err != nil {
+ return da.ResultRetrieveBlocks{
+ DAResult: da.DAResult{
+ Code: da.StatusError,
+ Message: err.Error(),
+ },
+ }
+	}
+
+	blocks := make([]*types.Block, len(data))
+	for i, msg := range data {
+ var block pb.Block
+ err = proto.Unmarshal(msg, &block)
+ if err != nil {
+ c.logger.Error("failed to unmarshal block", "daHeight", dataLayerHeight, "position", i, "error", err)
+ continue
+ }
+ blocks[i] = new(types.Block)
+ err := blocks[i].FromProto(&block)
+ if err != nil {
+ return da.ResultRetrieveBlocks{
+ DAResult: da.DAResult{
+ Code: da.StatusError,
+ Message: err.Error(),
+ },
+ }
+ }
+	}
+
+	return da.ResultRetrieveBlocks{
+ DAResult: da.DAResult{
+ Code: da.StatusSuccess,
+ DAHeight: dataLayerHeight,
+ },
+ Blocks: blocks,
+	}
+}

da/celestia/mock/messages.go

@@ -0,0 +1,106 @@
+package mock
+
+import (
+	"bytes"
+	"encoding/binary"
+)
+
+// This code is extracted from celestia-app. It's here to build shares from messages (serialized blocks).
+// TODO(tzdybal): if we stop using `/namespaced_shares` we can get rid of this file.
+
+const (
+	ShareSize = 256
+	NamespaceSize = 8
+	MsgShareSize = ShareSize - NamespaceSize
+)
+
+// splitMessage breaks the data in a message into the minimum number of
+// namespaced shares
+func splitMessage(rawData []byte, nid []byte) []NamespacedShare {
+	shares := make([]NamespacedShare, 0)
+	firstRawShare := append(append(
+ make([]byte, 0, ShareSize),
+ nid...),
+ rawData[:MsgShareSize]...,
+	)
+	shares = append(shares, NamespacedShare{firstRawShare, nid})
+	rawData = rawData[MsgShareSize:]
+	for len(rawData) > 0 {
+ shareSizeOrLen := min(MsgShareSize, len(rawData))
+ rawShare := append(append(
+ make([]byte, 0, ShareSize),
+ nid...),
+ rawData[:shareSizeOrLen]...,
+ )
+ paddedShare := zeroPadIfNecessary(rawShare, ShareSize)
+ share := NamespacedShare{paddedShare, nid}
+ shares = append(shares, share)
+ rawData = rawData[shareSizeOrLen:]
+	}
+	return shares
+}
+
+// Share contains the raw share data without the corresponding namespace.
+type Share []byte
+
+// NamespacedShare extends a Share with the corresponding namespace.
+type NamespacedShare struct {
+	Share
+	ID []byte
+}
+
+func min(a, b int) int {
+	if a <= b {
+ return a
+	}
+	return b
+}
+
+func zeroPadIfNecessary(share []byte, width int) []byte {
+	oldLen := len(share)
+	if oldLen < width {
+ missingBytes := width - oldLen
+ padByte := []byte{0}
+ padding := bytes.Repeat(padByte, missingBytes)
+ share = append(share, padding...)
+ return share
+	}
+	return share
+}
+
+// marshalDelimited marshals the raw data (excluding the namespace) of this
+// message and prefixes it with the length of that encoding.
+func marshalDelimited(data []byte) ([]byte, error) {
+	lenBuf := make([]byte, binary.MaxVarintLen64)
+	length := uint64(len(data))
+	n := binary.PutUvarint(lenBuf, length)
+	return append(lenBuf[:n], data...), nil
+}
+
+// appendToShares appends raw data as shares.
+// Used to build shares from blocks/messages.
+func appendToShares(shares []NamespacedShare, nid []byte, rawData []byte) []NamespacedShare {
+	if len(rawData) <= MsgShareSize {
+ rawShare := append(append(
+ make([]byte, 0, len(nid)+len(rawData)),
+ nid...),
+ rawData...,
+ )
+ paddedShare := zeroPadIfNecessary(rawShare, ShareSize)
+ share := NamespacedShare{paddedShare, nid}
+ shares = append(shares, share)
+	} else { // len(rawData) > MsgShareSize
+ shares = append(shares, splitMessage(rawData, nid)...)
+	}
+	return shares
+}
+
+type namespacedSharesResponse struct {
+	Shares []Share `json:"shares"`
+	Height uint64 `json:"height"`
+}
+
+type namespacedDataResponse struct {
+	Data [][]byte `json:"data"`
+	Height uint64 `json:"height"`
+}