fix(da): switch to jsonrpc da client

cmd/rollkit/commands/run_node.go

@@ -4,9 +4,7 @@ import (
 	"context"
 	"fmt"
 	"math/rand"
-	"net"
 	"os"
-	"strconv"
 
 	cmtcmd "github.com/cometbft/cometbft/cmd/cometbft/commands"
 	cometconf "github.com/cometbft/cometbft/config"
@@ -20,8 +18,6 @@ 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"
 	goDATest "github.com/rollkit/go-da/test"
@@ -176,7 +172,7 @@ func NewRunNodeCmd() *cobra.Command {
 
 	// use mock da server by default
 	if !cmd.Flags().Lookup("rollkit.da_address").Changed {
-		rollkitConfig.DAAddress = ":7980"
+		rollkitConfig.DAAddress = "http://localhost:7980"
 	}
 	return cmd
 }
@@ -194,16 +190,13 @@ func addNodeFlags(cmd *cobra.Command) {
 }
 
 // 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))
+func startMockGRPCServ() *proxy.Server {
+	srv := proxy.NewServer("localhost", "7980", goDATest.NewDummyDA())
+	err := srv.Start(context.Background())
 	if err != nil {
 		fmt.Println(err)
 		return nil
 	}
-	go func() {
-		_ = srv.Serve(lis)
-	}()
 	return srv
 }
 

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")

da/client.go

@@ -0,0 +1,106 @@
+package da
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/filecoin-project/go-jsonrpc"
+
+	"github.com/rollkit/go-da"
+)
+
+type API struct {
+	Internal struct {
+		MaxBlobSize func(ctx context.Context) (uint64, error)                                            `perm:"read"`
+		Get         func(ctx context.Context, ids []da.ID, ns da.Namespace) ([]da.Blob, error)           `perm:"read"`
+		GetIDs      func(ctx context.Context, height uint64, ns da.Namespace) ([]da.ID, error)           `perm:"read"`
+		GetProofs   func(ctx context.Context, ids []da.ID, ns da.Namespace) ([]da.Proof, error)          `perm:"read"`
+		Commit      func(ctx context.Context, blobs []da.Blob, ns da.Namespace) ([]da.Commitment, error) `perm:"read"`
+		Validate    func(context.Context, []da.ID, []da.Proof, da.Namespace) ([]bool, error)             `perm:"read"`
+		Submit      func(context.Context, []da.Blob, float64, da.Namespace) ([]da.ID, error)             `perm:"write"`
+	}
+}
+
+func (api *API) MaxBlobSize(ctx context.Context) (uint64, error) {
+	return api.Internal.MaxBlobSize(ctx)
+}
+
+func (api *API) Get(ctx context.Context, ids []da.ID, ns da.Namespace) ([]da.Blob, error) {
+	return api.Internal.Get(ctx, ids, ns)
+}
+
+func (api *API) GetIDs(ctx context.Context, height uint64, ns da.Namespace) ([]da.ID, error) {
+	return api.Internal.GetIDs(ctx, height, ns)
+}
+
+func (api *API) GetProofs(ctx context.Context, ids []da.ID, ns da.Namespace) ([]da.Proof, error) {
+	return api.Internal.GetProofs(ctx, ids, ns)
+}
+
+func (api *API) Commit(ctx context.Context, blobs []da.Blob, ns da.Namespace) ([]da.Commitment, error) {
+	return api.Internal.Commit(ctx, blobs, ns)
+}
+
+func (api *API) Validate(ctx context.Context, ids []da.ID, proofs []da.Proof, ns da.Namespace) ([]bool, error) {
+	return api.Internal.Validate(ctx, ids, proofs, ns)
+}
+
+func (api *API) Submit(ctx context.Context, blobs []da.Blob, gasPrice float64, ns da.Namespace) ([]da.ID, error) {
+	return api.Internal.Submit(ctx, blobs, gasPrice, ns)
+}
+
+type Client struct {
+	DA     API
+	closer multiClientCloser
+}
+
+// multiClientCloser is a wrapper struct to close clients across multiple namespaces.
+type multiClientCloser struct {
+	closers []jsonrpc.ClientCloser
+}
+
+// register adds a new closer to the multiClientCloser
+func (m *multiClientCloser) register(closer jsonrpc.ClientCloser) {
+	m.closers = append(m.closers, closer)
+}
+
+// closeAll closes all saved clients.
+func (m *multiClientCloser) closeAll() {
+	for _, closer := range m.closers {
+		closer()
+	}
+}
+
+// Close closes the connections to all namespaces registered on the staticClient.
+func (c *Client) Close() {
+	c.closer.closeAll()
+}
+
+// NewClient creates a new Client with one connection per namespace with the
+// given token as the authorization token.
+func NewClient(ctx context.Context, addr string, token string) (*Client, error) {
+	authHeader := http.Header{"Authorization": []string{fmt.Sprintf("Bearer %s", token)}}
+	return newClient(ctx, addr, authHeader)
+}
+
+func newClient(ctx context.Context, addr string, authHeader http.Header) (*Client, error) {
+	var multiCloser multiClientCloser
+	var client Client
+	for name, module := range moduleMap(&client) {
+		closer, err := jsonrpc.NewClient(ctx, addr, name, module, authHeader)
+		if err != nil {
+			return nil, err
+		}
+		multiCloser.register(closer)
+	}
+
+	return &client, nil
+}
+
+func moduleMap(client *Client) map[string]interface{} {
+	// TODO: this duplication of strings many times across the codebase can be avoided with issue #1176
+	return map[string]interface{}{
+		"da": &client.DA.Internal,
+	}
+}

da/da_test.go

@@ -6,18 +6,13 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
-	"net"
 	"os"
-	"strconv"
 	"testing"
 	"time"
 
 	"github.com/cometbft/cometbft/libs/log"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"google.golang.org/grpc"
-
-	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/rollkit/go-da"
 	"github.com/rollkit/go-da/proxy"
@@ -36,7 +31,10 @@ func TestMain(m *testing.M) {
 	exitCode := m.Run()
 
 	// teardown servers
-	srv.GracefulStop()
+	err := srv.Stop(context.TODO())
+	if err != nil {
+		fmt.Println(err)
+	}
 
 	os.Exit(exitCode)
 }
@@ -97,26 +95,22 @@ func TestSubmitRetrieve(t *testing.T) {
 	}
 }
 
-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))
+func startMockGRPCServ() *proxy.Server {
+	srv := proxy.NewServer("localhost", "7980", goDATest.NewDummyDA())
+	err := srv.Start(context.TODO())
 	if err != nil {
 		fmt.Println(err)
 		return nil
 	}
-	go func() {
-		_ = srv.Serve(lis)
-	}()
 	return srv
 }
 
 func startMockGRPCClient() (*DAClient, error) {
-	client := proxy.NewClient()
-	err := client.Start("127.0.0.1:7980", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	client, err := proxy.NewClient(context.TODO(), "http://localhost:7980", "")
 	if err != nil {
 		return nil, err
 	}
-	return &DAClient{DA: client, GasPrice: -1, GasMultiplier: -1, Logger: log.TestingLogger()}, nil
+	return &DAClient{DA: &client.DA, GasPrice: -1, GasMultiplier: -1, Logger: log.TestingLogger()}, nil
 }
 
 func doTestSubmitTimeout(t *testing.T, dalc *DAClient, blocks []*types.Block) {

da/mock/cmd/main.go

@@ -1,14 +1,11 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"log"
-	"net"
 	"strconv"
 
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-
 	"github.com/rollkit/go-da/proxy"
 	goDATest "github.com/rollkit/go-da/test"
 )
@@ -22,13 +19,9 @@ func main() {
 	flag.StringVar(&host, "host", "0.0.0.0", "listening address")
 	flag.Parse()
 
-	lis, err := net.Listen("tcp", host+":"+strconv.Itoa(port))
-	if err != nil {
-		log.Panic(err)
-	}
-	log.Println("Listening on:", lis.Addr())
-	srv := proxy.NewServer(goDATest.NewDummyDA(), grpc.Creds(insecure.NewCredentials()))
-	if err := srv.Serve(lis); err != nil {
+	srv := proxy.NewServer(host, strconv.Itoa(port), goDATest.NewDummyDA())
+	log.Printf("Listening on: %s:%d", host, port)
+	if err := srv.Start(context.Background()); err != nil {
 		log.Fatal("error while serving:", err)
 	}
 }

go.mod

@@ -34,6 +34,7 @@ require (
 
 require (
 	github.com/celestiaorg/go-header v0.5.3
+	github.com/filecoin-project/go-jsonrpc v0.3.1
 	github.com/ipfs/go-ds-badger4 v0.1.5
 )
 
@@ -179,9 +180,12 @@ require (
 	golang.org/x/sys v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/tools v0.16.0 // indirect
+	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	gonum.org/v1/gonum v0.12.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	lukechampine.com/blake3 v1.2.1 // indirect
 )
+
+replace github.com/rollkit/go-da => github.com/rollkit/go-da v0.4.1-0.20240305232117-d39d77c7a6c3

go.sum

@@ -356,6 +356,8 @@ github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGE
 github.com/fatih/color v1.12.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/fatih/structtag v1.2.0/go.mod h1:mBJUNpUnHmRKrKlQQlmCrh5PuhftFbNv8Ys4/aAZl94=
+github.com/filecoin-project/go-jsonrpc v0.3.1 h1:qwvAUc5VwAkooquKJmfz9R2+F8znhiqcNHYjEp/NM10=
+github.com/filecoin-project/go-jsonrpc v0.3.1/go.mod h1:jBSvPTl8V1N7gSTuCR4bis8wnQnIjHbRPpROol6iQKM=
 github.com/firefart/nonamedreturns v1.0.4/go.mod h1:TDhe/tjI1BXo48CmYbUduTV7BdIga8MAO/xbKdcVsGI=
 github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:xEzjJPgXI435gkrCt3MPfRiAkVrwSbHsst4LCFVfpJc=
 github.com/flynn/noise v1.0.0 h1:DlTHqmzmvcEiKj+4RYo/imoswx/4r6iBlCMfVtrMXpQ=
@@ -1362,8 +1364,8 @@ github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTE
 github.com/rogpeppe/go-internal v1.8.1/go.mod h1:JeRgkft04UBgHMgCIwADu4Pn6Mtm5d4nPKWu0nJ5d+o=
 github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
 github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
-github.com/rollkit/go-da v0.4.0 h1:/s7ZrVq7DC2aK8UXIvB7rsXrZ2mVGRw7zrexcxRvhlw=
-github.com/rollkit/go-da v0.4.0/go.mod h1:Kef0XI5ecEKd3TXzI8S+9knAUJnZg0svh2DuXoCsPlM=
+github.com/rollkit/go-da v0.4.1-0.20240305232117-d39d77c7a6c3 h1:a/5BwBCFSssxLmkjjrbVeYi8kUzly4kt+qqfBXPExkM=
+github.com/rollkit/go-da v0.4.1-0.20240305232117-d39d77c7a6c3/go.mod h1:+SNuI00o3NiQCXCWK1iwJV8aUM7E2DN6WLPlN7nswp0=
 github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
 github.com/rs/cors v1.8.2/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
 github.com/rs/cors v1.10.1 h1:L0uuZVXIKlI1SShY2nhFfo44TYvDPQ1w4oFkUJNfhyo=
@@ -2153,6 +2155,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
+golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 h1:H2TDz8ibqkAF6YGhCdN3jS9O0/s90v0rJh3X/OLHEUk=
+golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2/go.mod h1:K8+ghG5WaK9qNqU5K3HdILfMLy1f3aNYFI/wnl100a8=
 gonum.org/v1/gonum v0.0.0-20180816165407-929014505bf4/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
 gonum.org/v1/gonum v0.8.2/go.mod h1:oe/vMfY3deqTw+1EZJhuvEW2iwGF1bW9wwu7XCu0+v0=
 gonum.org/v1/gonum v0.12.0 h1:xKuo6hzt+gMav00meVPUlXwSdoEJP46BR+wdxQEFK2o=

node/full.go

@@ -14,8 +14,6 @@ import (
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
 
 	abci "github.com/cometbft/cometbft/abci/types"
 	llcfg "github.com/cometbft/cometbft/config"
@@ -26,7 +24,6 @@ import (
 	rpcclient "github.com/cometbft/cometbft/rpc/client"
 	cmtypes "github.com/cometbft/cometbft/types"
 
-	goDAProxy "github.com/rollkit/go-da/proxy"
 	"github.com/rollkit/rollkit/block"
 	"github.com/rollkit/rollkit/config"
 	"github.com/rollkit/rollkit/da"
@@ -228,12 +225,11 @@ func initDALC(nodeConfig config.NodeConfig, dalcKV ds.TxnDatastore, logger log.L
 	if err != nil {
 		return nil, fmt.Errorf("error decoding namespace: %w", err)
 	}
-	daClient := goDAProxy.NewClient()
-	err = daClient.Start(nodeConfig.DAAddress, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	daClient, err := da.NewClient(context.Background(), nodeConfig.DAAddress, nodeConfig.DAAuthToken)
 	if err != nil {
-		return nil, fmt.Errorf("error while establishing GRPC connection to DA layer: %w", err)
+		return nil, fmt.Errorf("error while establishing connection to DA layer: %w", err)
 	}
-	return &da.DAClient{DA: daClient, Namespace: namespace, GasPrice: nodeConfig.DAGasPrice, GasMultiplier: nodeConfig.DAGasMultiplier, Logger: logger.With("module", "da_client")}, nil
+	return &da.DAClient{DA: &daClient.DA, Namespace: namespace, GasPrice: nodeConfig.DAGasPrice, Logger: logger.With("module", "da_client")}, nil
 }
 
 func initMempool(logger log.Logger, proxyApp proxy.AppConns, memplMetrics *mempool.Metrics) *mempool.CListMempool {

node/node_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/rollkit/rollkit/types"
 )
 
-var MockServerAddr = ":7980"
+var MockServerAddr = "http://localhost:7980"
 
 var MockNamespace = "00000000000000000000000000000000000000000000000000deadbeef"
 

rpc/json/helpers_test.go

@@ -30,7 +30,7 @@ func prepareProposalResponse(_ context.Context, req *abci.RequestPrepareProposal
 	}, nil
 }
 
-var MockServerAddr = ":7980"
+var MockServerAddr = "http://localhost:7980"
 
 var MockNamespace = "deadbeef"