fix(da): add jsonrpc da client

block/manager_test.go

@@ -30,7 +30,7 @@ import (
 func getManager(t *testing.T, backend goDA.DA) *Manager {
 	logger := test.NewFileLoggerCustom(t, test.TempLogFileName(t, t.Name()))
 	return &Manager{
-		dalc:       &da.DAClient{DA: backend, GasPrice: -1, GasMultiplier: -1, Logger: logger},
+		dalc:       da.NewDAClient(backend, -1, -1, nil, logger),
 		blockCache: NewBlockCache(),
 		logger:     logger,
 	}

cmd/rollkit/commands/run_node.go

@@ -4,9 +4,8 @@ import (
 	"context"
 	"fmt"
 	"math/rand"
-	"net"
+	"net/url"
 	"os"
-	"strconv"
 
 	cmtcmd "github.com/cometbft/cometbft/cmd/cometbft/commands"
 	cometconf "github.com/cometbft/cometbft/config"
@@ -20,10 +19,8 @@ import (
 	cometproxy "github.com/cometbft/cometbft/proxy"
 	comettypes "github.com/cometbft/cometbft/types"
 	comettime "github.com/cometbft/cometbft/types/time"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 
-	"github.com/rollkit/go-da/proxy"
+	proxy "github.com/rollkit/go-da/proxy/jsonrpc"
 	goDATest "github.com/rollkit/go-da/test"
 
 	"github.com/spf13/cobra"
@@ -115,6 +112,16 @@ func NewRunNodeCmd() *cobra.Command {
 			// initialize the metrics
 			metrics := rollnode.DefaultMetricsProvider(cometconf.DefaultInstrumentationConfig())
 
+			// use mock jsonrpc da server by default
+			if !cmd.Flags().Lookup("rollkit.da_address").Changed {
+				srv, err := startMockDAServJSONRPC(cmd.Context())
+				if err != nil {
+					return fmt.Errorf("failed to launch mock da server: %w", err)
+				}
+				// nolint:errcheck,gosec
+				defer func() { srv.Stop(cmd.Context()) }()
+			}
+
 			// create the rollkit node
 			rollnode, err := rollnode.NewNode(
 				context.Background(),
@@ -130,9 +137,6 @@ func NewRunNodeCmd() *cobra.Command {
 				return fmt.Errorf("failed to create new rollkit node: %w", err)
 			}
 
-			// start mock da server
-			startMockGRPCServ()
-
 			// Launch the RPC server
 			server := rollrpc.NewServer(rollnode, config.RPC, logger)
 			err = server.Start()
@@ -173,11 +177,6 @@ func NewRunNodeCmd() *cobra.Command {
 	if !cmd.Flags().Lookup("rollkit.aggregator").Changed {
 		rollkitConfig.Aggregator = true
 	}
-
-	// use mock da server by default
-	if !cmd.Flags().Lookup("rollkit.da_address").Changed {
-		rollkitConfig.DAAddress = ":7980"
-	}
 	return cmd
 }
 
@@ -193,18 +192,15 @@ func addNodeFlags(cmd *cobra.Command) {
 	rollconf.AddFlags(cmd)
 }
 
-// startMockGRPCServ starts a mock gRPC server for the dummy DA
-func startMockGRPCServ() *grpc.Server {
-	srv := proxy.NewServer(goDATest.NewDummyDA(), grpc.Creds(insecure.NewCredentials()))
-	lis, err := net.Listen("tcp", "127.0.0.1"+":"+strconv.Itoa(7980))
+// startMockDAServJSONRPC starts a mock JSONRPC server
+func startMockDAServJSONRPC(ctx context.Context) (*proxy.Server, error) {
+	addr, _ := url.Parse(rollkitConfig.DAAddress)
+	srv := proxy.NewServer(addr.Hostname(), addr.Port(), goDATest.NewDummyDA())
+	err := srv.Start(ctx)
 	if err != nil {
-		fmt.Println(err)
-		return nil
+		return nil, err
 	}
-	go func() {
-		_ = srv.Serve(lis)
-	}()
-	return srv
+	return srv, nil
 }
 
 // TODO (Ferret-san): modify so that it initiates files with rollkit configurations by default

cmd/rollkit/docs/rollkit_start.md

@@ -30,7 +30,8 @@ rollkit start [flags]
       --proxy_app string                                proxy app address, or one of: 'kvstore', 'persistent_kvstore' or 'noop' for local testing. (default "tcp://127.0.0.1:26658")
       --rollkit.aggregator                              run node in aggregator mode
       --rollkit.block_time duration                     block time (for aggregator mode) (default 1s)
-      --rollkit.da_address string                       DA address (host:port) (default ":26650")
+      --rollkit.da_address string                       DA address (host:port) (default "http://localhost:26658")
+      --rollkit.da_auth_token string                    DA auth token
       --rollkit.da_block_time duration                  DA chain block time (for syncing) (default 15s)
       --rollkit.da_gas_multiplier float                 DA gas price multiplier for retrying blob transactions (default -1)
       --rollkit.da_gas_price float                      DA gas price for blob transactions (default -1)

config/config.go

@@ -14,6 +14,8 @@ const (
 	FlagAggregator = "rollkit.aggregator"
 	// FlagDAAddress is a flag for specifying the data availability layer address
 	FlagDAAddress = "rollkit.da_address"
+	// FlagDAAuthToken is a flag for specifying the data availability layer auth token
+	FlagDAAuthToken = "rollkit.da_auth_token" // #nosec G101
 	// FlagBlockTime is a flag for specifying the block time
 	FlagBlockTime = "rollkit.block_time"
 	// FlagDABlockTime is a flag for specifying the data availability layer block time
@@ -45,6 +47,7 @@ type NodeConfig struct {
 	Aggregator         bool `mapstructure:"aggregator"`
 	BlockManagerConfig `mapstructure:",squash"`
 	DAAddress          string `mapstructure:"da_address"`
+	DAAuthToken        string `mapstructure:"da_auth_token"`
 	Light              bool   `mapstructure:"light"`
 	HeaderConfig       `mapstructure:",squash"`
 	LazyAggregator     bool                         `mapstructure:"lazy_aggregator"`
@@ -106,6 +109,7 @@ func GetNodeConfig(nodeConf *NodeConfig, cmConf *cmcfg.Config) {
 func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error {
 	nc.Aggregator = v.GetBool(FlagAggregator)
 	nc.DAAddress = v.GetString(FlagDAAddress)
+	nc.DAAuthToken = v.GetString(FlagDAAuthToken)
 	nc.DAGasPrice = v.GetFloat64(FlagDAGasPrice)
 	nc.DAGasMultiplier = v.GetFloat64(FlagDAGasMultiplier)
 	nc.DANamespace = v.GetString(FlagDANamespace)
@@ -127,6 +131,7 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().Bool(FlagAggregator, def.Aggregator, "run node in aggregator mode")
 	cmd.Flags().Bool(FlagLazyAggregator, def.LazyAggregator, "wait for transactions, don't build empty blocks")
 	cmd.Flags().String(FlagDAAddress, def.DAAddress, "DA address (host:port)")
+	cmd.Flags().String(FlagDAAuthToken, def.DAAuthToken, "DA auth token")
 	cmd.Flags().Duration(FlagBlockTime, def.BlockTime, "block time (for aggregator mode)")
 	cmd.Flags().Duration(FlagDABlockTime, def.DABlockTime, "DA chain block time (for syncing)")
 	cmd.Flags().Float64(FlagDAGasPrice, def.DAGasPrice, "DA gas price for blob transactions")

config/defaults.go

@@ -26,7 +26,7 @@ var DefaultNodeConfig = NodeConfig{
 		BlockTime:   1 * time.Second,
 		DABlockTime: 15 * time.Second,
 	},
-	DAAddress:       ":26650",
+	DAAddress:       "http://localhost:26658",
 	DAGasPrice:      -1,
 	DAGasMultiplier: -1,
 	Light:           false,

da/da.go

@@ -16,12 +16,12 @@ import (
 	pb "github.com/rollkit/rollkit/types/pb/rollkit"
 )
 
-var (
-	// submitTimeout is the timeout for block submission
-	submitTimeout = 60 * time.Second
+const (
+	// defaultSubmitTimeout is the timeout for block submission
+	defaultSubmitTimeout = 60 * time.Second
 
-	// retrieveTimeout is the timeout for block retrieval
-	retrieveTimeout = 60 * time.Second
+	// defaultRetrieveTimeout is the timeout for block retrieval
+	defaultRetrieveTimeout = 60 * time.Second
 )
 
 var (
@@ -95,11 +95,26 @@ type ResultRetrieveBlocks struct {
 
 // DAClient is a new DA implementation.
 type DAClient struct {
-	DA            goDA.DA
-	GasPrice      float64
-	GasMultiplier float64
-	Namespace     goDA.Namespace
-	Logger        log.Logger
+	DA              goDA.DA
+	GasPrice        float64
+	GasMultiplier   float64
+	Namespace       goDA.Namespace
+	SubmitTimeout   time.Duration
+	RetrieveTimeout time.Duration
+	Logger          log.Logger
+}
+
+// NewDAClient returns a new DA client.
+func NewDAClient(da goDA.DA, gasPrice, gasMultiplier float64, ns goDA.Namespace, logger log.Logger) *DAClient {
+	return &DAClient{
+		DA:              da,
+		GasPrice:        gasPrice,
+		GasMultiplier:   gasMultiplier,
+		Namespace:       ns,
+		SubmitTimeout:   defaultSubmitTimeout,
+		RetrieveTimeout: defaultRetrieveTimeout,
+		Logger:          logger,
+	}
 }
 
 // SubmitBlocks submits blocks to DA.
@@ -133,7 +148,7 @@ func (dac *DAClient) SubmitBlocks(ctx context.Context, blocks []*types.Block, ma
 			},
 		}
 	}
-	ctx, cancel := context.WithTimeout(ctx, submitTimeout)
+	ctx, cancel := context.WithTimeout(ctx, dac.SubmitTimeout)
 	defer cancel()
 	ids, err := dac.DA.Submit(ctx, blobs, gasPrice, dac.Namespace)
 	if err != nil {
@@ -200,7 +215,7 @@ func (dac *DAClient) RetrieveBlocks(ctx context.Context, dataLayerHeight uint64)
 		}
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, retrieveTimeout)
+	ctx, cancel := context.WithTimeout(ctx, dac.RetrieveTimeout)
 	defer cancel()
 	blobs, err := dac.DA.Get(ctx, ids, dac.Namespace)
 	if err != nil {

da/da.md

@@ -4,7 +4,11 @@ Rollkit provides a wrapper for [go-da][go-da], a generic data availability inter
 
 ## Details
 
-`DAClient` under the hood uses a GRPC implementation of the [go-da][go-da] DA interface. Using the `DAAddress` specified in the node's config, node creates a GRPC connection to it using go-da's gprc implementation [grpc-proxy][grpc-proxy] which is then used under the hood of `DAClient` to communicate with the underlying DA.
+`DAClient` can connect via either gRPC or JSON-RPC transports using the [go-da][go-da] [proxy/grpc][proxy/grpc] or [proxy/jsonrpc][proxy/jsonrpc] implementations. The connection can be configured using the following cli flags:
+
+* `--rollkit.da_address`: url address of the DA service (default: "grpc://localhost:26650")
+* `--rollkit.da_auth_token`: authentication token of the DA service
+* `--rollkit.da_namespace`: namespace to use when submitting blobs to the DA service
 
 Given a set of blocks to be submitted to DA by the block manager, the `SubmitBlocks` first encodes the blocks using protobuf (the encoded data are called blobs) and invokes the `Submit` method on the underlying DA implementation. On successful submission (`StatusSuccess`), the DA block height which included in the rollup blocks is returned.
 
@@ -29,9 +33,12 @@ See [da implementation]
 
 [2] [celestia-da][celestia-da]
 
-[3] [grpc-proxy][grpc-proxy]
+[3] [proxy/grpc][proxy/grpc]
+
+[4] [proxy/jsonrpc][proxy/jsonrpc]
 
 [da implementation]: https://github.com/rollkit/rollkit/blob/main/da/da.go
 [go-da]: https://github.com/rollkit/go-da
 [celestia-da]: https://github.com/rollkit/celestia-da
-[grpc-proxy]: https://github.com/rollkit/go-da/tree/main/proxy
+[proxy/grpc]: https://github.com/rollkit/go-da/tree/main/proxy/grpc
+[proxy/jsonrpc]: https://github.com/rollkit/go-da/tree/main/proxy/jsonrpc