Merge branch 'main' into security-update

Author: Stefan-Ethernal

client/client.go

@@ -17,8 +17,10 @@ type Factory interface {
 
 // Client is the interface that defines the implementation of all the endpoints
 type Client interface {
+	GetStatus(ctx context.Context) (*types.DACStatus, error)
 	GetOffChainData(ctx context.Context, hash common.Hash) ([]byte, error)
-	SignSequence(signedSequence types.SignedSequence) ([]byte, error)
+	ListOffChainData(ctx context.Context, hashes []common.Hash) (map[common.Hash][]byte, error)
+	SignSequence(ctx context.Context, signedSequence types.SignedSequence) ([]byte, error)
 }
 
 // factory is the implementation of the data committee client factory
@@ -46,10 +48,29 @@ func New(url string) Client {
 	}
 }
 
+// GetStatus returns DAC status
+func (c *client) GetStatus(ctx context.Context) (*types.DACStatus, error) {
+	response, err := rpc.JSONRPCCallWithContext(ctx, c.url, "status_getStatus")
+	if err != nil {
+		return nil, err
+	}
+
+	if response.Error != nil {
+		return nil, fmt.Errorf("%v %v", response.Error.Code, response.Error.Message)
+	}
+
+	var result types.DACStatus
+	if err = json.Unmarshal(response.Result, &result); err != nil {
+		return nil, err
+	}
+
+	return &result, nil
+}
+
 // SignSequence sends a request to sign the given sequence by the data committee member
 // if successful returns the signature. The signature should be validated after using this method!
-func (c *client) SignSequence(signedSequence types.SignedSequence) ([]byte, error) {
-	response, err := rpc.JSONRPCCall(c.url, "datacom_signSequence", signedSequence)
+func (c *client) SignSequence(ctx context.Context, signedSequence types.SignedSequence) ([]byte, error) {
+	response, err := rpc.JSONRPCCallWithContext(ctx, c.url, "datacom_signSequence", signedSequence)
 	if err != nil {
 		return nil, err
 	}
@@ -84,3 +105,27 @@ func (c *client) GetOffChainData(ctx context.Context, hash common.Hash) ([]byte,
 
 	return result, nil
 }
+
+// ListOffChainData returns data based on the given hashes
+func (c *client) ListOffChainData(ctx context.Context, hashes []common.Hash) (map[common.Hash][]byte, error) {
+	response, err := rpc.JSONRPCCallWithContext(ctx, c.url, "sync_listOffChainData", hashes)
+	if err != nil {
+		return nil, err
+	}
+
+	if response.Error != nil {
+		return nil, fmt.Errorf("%v %v", response.Error.Code, response.Error.Message)
+	}
+
+	result := make(map[common.Hash]types.ArgBytes)
+	if err = json.Unmarshal(response.Result, &result); err != nil {
+		return nil, err
+	}
+
+	preparedResult := make(map[common.Hash][]byte)
+	for key, val := range result {
+		preparedResult[key] = val
+	}
+
+	return preparedResult, nil
+}

client/client_test.go

@@ -16,6 +16,69 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestClient_GetStatus(t *testing.T) {
+	tests := []struct {
+		name       string
+		result     string
+		status     *types.DACStatus
+		statusCode int
+		err        error
+	}{
+		{
+			name:   "successfully got status",
+			result: `{"result":{"version":"v1.0.0","uptime":"123","key_count":2,"backfill_progress":5}}`,
+			status: &types.DACStatus{
+				Uptime:           "123",
+				Version:          "v1.0.0",
+				KeyCount:         2,
+				BackfillProgress: 5,
+			},
+		},
+		{
+			name:   "error returned by server",
+			result: `{"error":{"code":123,"message":"test error"}}`,
+			err:    errors.New("123 test error"),
+		},
+		{
+			name:       "unsuccessful status code returned by server",
+			statusCode: http.StatusInternalServerError,
+			err:        errors.New("invalid status code, expected: 200, found: 500"),
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				var res rpc.Request
+				require.NoError(t, json.NewDecoder(r.Body).Decode(&res))
+				require.Equal(t, "status_getStatus", res.Method)
+
+				if tt.statusCode > 0 {
+					w.WriteHeader(tt.statusCode)
+				}
+
+				_, err := fmt.Fprint(w, tt.result)
+				require.NoError(t, err)
+			}))
+			defer srv.Close()
+
+			client := New(srv.URL)
+
+			got, err := client.GetStatus(context.Background())
+			if tt.err != nil {
+				require.Error(t, err)
+				require.EqualError(t, tt.err, err.Error())
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tt.status, got)
+			}
+		})
+	}
+}
+
 func TestClient_SignSequence(t *testing.T) {
 	tests := []struct {
 		name       string
@@ -87,7 +150,7 @@ func TestClient_SignSequence(t *testing.T) {
 
 			client := New(srv.URL)
 
-			got, err := client.SignSequence(tt.ss)
+			got, err := client.SignSequence(context.Background(), tt.ss)
 			if tt.err != nil {
 				require.Error(t, err)
 				require.EqualError(t, tt.err, err.Error())
@@ -169,3 +232,81 @@ func TestClient_GetOffChainData(t *testing.T) {
 		})
 	}
 }
+
+func TestClient_ListOffChainData(t *testing.T) {
+	tests := []struct {
+		name       string
+		hashes     []common.Hash
+		result     string
+		data       map[common.Hash][]byte
+		statusCode int
+		err        error
+	}{
+		{
+			name:   "successfully got offhcain data",
+			hashes: []common.Hash{common.BytesToHash([]byte("hash"))},
+			result: fmt.Sprintf(`{"result":{"%s":"%s"}}`,
+				common.BytesToHash([]byte("hash")).Hex(), hex.EncodeToString([]byte("offchaindata"))),
+			data: map[common.Hash][]byte{
+				common.BytesToHash([]byte("hash")): []byte("offchaindata"),
+			},
+		},
+		{
+			name:   "error returned by server",
+			hashes: []common.Hash{common.BytesToHash([]byte("hash"))},
+			result: `{"error":{"code":123,"message":"test error"}}`,
+			err:    errors.New("123 test error"),
+		},
+		{
+			name:   "invalid offchain data returned by server",
+			hashes: []common.Hash{common.BytesToHash([]byte("hash"))},
+			result: fmt.Sprintf(`{"result":{"%s":"invalid-signature"}}`,
+				common.BytesToHash([]byte("hash")).Hex()),
+			data: map[common.Hash][]byte{
+				common.BytesToHash([]byte("hash")): nil,
+			},
+		},
+		{
+			name:       "unsuccessful status code returned by server",
+			hashes:     []common.Hash{common.BytesToHash([]byte("hash"))},
+			statusCode: http.StatusUnauthorized,
+			err:        errors.New("invalid status code, expected: 200, found: 401"),
+		},
+	}
+	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 rpc.Request
+				require.NoError(t, json.NewDecoder(r.Body).Decode(&res))
+				require.Equal(t, "sync_listOffChainData", res.Method)
+
+				var params [][]common.Hash
+				require.NoError(t, json.Unmarshal(res.Params, &params))
+				require.Equal(t, tt.hashes, params[0])
+
+				if tt.statusCode > 0 {
+					w.WriteHeader(tt.statusCode)
+				}
+
+				_, err := fmt.Fprint(w, tt.result)
+				require.NoError(t, err)
+			}))
+			defer svr.Close()
+
+			c := &client{url: svr.URL}
+
+			got, err := c.ListOffChainData(context.Background(), tt.hashes)
+			if tt.err != nil {
+				require.Error(t, err)
+				require.EqualError(t, tt.err, err.Error())
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tt.data, got)
+			}
+		})
+	}
+}

cmd/main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/0xPolygon/cdk-data-availability/rpc"
 	"github.com/0xPolygon/cdk-data-availability/sequencer"
 	"github.com/0xPolygon/cdk-data-availability/services/datacom"
+	"github.com/0xPolygon/cdk-data-availability/services/status"
 	"github.com/0xPolygon/cdk-data-availability/services/sync"
 	"github.com/0xPolygon/cdk-data-availability/synchronizer"
 	"github.com/0xPolygon/cdk-data-availability/types"
@@ -140,17 +141,17 @@ func start(cliCtx *cli.Context) error {
 	server := rpc.NewServer(
 		c.RPC,
 		[]rpc.Service{
+			{
+				Name:    status.APISTATUS,
+				Service: status.NewEndpoints(storage),
+			},
 			{
 				Name:    sync.APISYNC,
-				Service: sync.NewSyncEndpoints(storage),
+				Service: sync.NewEndpoints(storage),
 			},
 			{
-				Name: datacom.APIDATACOM,
-				Service: datacom.NewDataComEndpoints(
-					storage,
-					pk,
-					sequencerTracker,
-				),
+				Name:    datacom.APIDATACOM,
+				Service: datacom.NewEndpoints(storage, pk, sequencerTracker),
 			},
 		},
 	)

config/config.go

@@ -39,6 +39,7 @@ type L1Config struct {
 	Timeout                types.Duration `mapstructure:"Timeout"`
 	RetryPeriod            types.Duration `mapstructure:"RetryPeriod"`
 	BlockBatchSize         uint           `mapstructure:"BlockBatchSize"`
+	TrackSequencer         bool           `mapstructure:"TrackSequencer"`
 
 	// GenesisBlock represents the block number where PolygonValidium contract is deployed on L1
 	GenesisBlock uint64 `mapstructure:"GenesisBlock"`

config/default.go

@@ -20,6 +20,7 @@ Timeout = "1m"
 RetryPeriod = "5s"
 BlockBatchSize = "64"
 GenesisBlock = "0"
+TrackSequencer = true
 
 [Log]
 Environment = "development" # "production" or "development"

db/db.go

@@ -29,7 +29,10 @@ type DB interface {
 
 	Exists(ctx context.Context, key common.Hash) bool
 	GetOffChainData(ctx context.Context, key common.Hash, dbTx sqlx.QueryerContext) (types.ArgBytes, error)
+	ListOffChainData(ctx context.Context, keys []common.Hash, dbTx sqlx.QueryerContext) (map[common.Hash]types.ArgBytes, error)
 	StoreOffChainData(ctx context.Context, od []types.OffChainData, dbTx sqlx.ExecerContext) error
+
+	CountOffchainData(ctx context.Context) (uint64, error)
 }
 
 // Tx is the interface that defines functions a db tx has to implement
@@ -218,6 +221,66 @@ func (db *pgDB) GetOffChainData(ctx context.Context, key common.Hash, dbTx sqlx.
 	return common.FromHex(hexValue), nil
 }
 
+// ListOffChainData returns values identified by the given keys
+func (db *pgDB) ListOffChainData(ctx context.Context, keys []common.Hash, dbTx sqlx.QueryerContext) (map[common.Hash]types.ArgBytes, error) {
+	if len(keys) == 0 {
+		return nil, nil
+	}
+
+	const listOffchainDataSQL = `
+		SELECT key, value
+		FROM data_node.offchain_data 
+		WHERE key IN (?);
+	`
+
+	preparedKeys := make([]string, len(keys))
+	for i, key := range keys {
+		preparedKeys[i] = key.Hex()
+	}
+
+	query, args, err := sqlx.In(listOffchainDataSQL, preparedKeys)
+	if err != nil {
+		return nil, err
+	}
+
+	// sqlx.In returns queries with the `?` bindvar, we can rebind it for our backend
+	query = db.pg.Rebind(query)
+
+	rows, err := db.querier(dbTx).QueryxContext(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+
+	defer rows.Close()
+
+	list := make(map[common.Hash]types.ArgBytes)
+	for rows.Next() {
+		data := struct {
+			Key   string `db:"key"`
+			Value string `db:"value"`
+		}{}
+		if err = rows.StructScan(&data); err != nil {
+			return nil, err
+		}
+
+		list[common.HexToHash(data.Key)] = common.FromHex(data.Value)
+	}
+
+	return list, nil
+}
+
+// CountOffchainData returns the count of rows in the offchain_data table
+func (db *pgDB) CountOffchainData(ctx context.Context) (uint64, error) {
+	const countQuery = "SELECT COUNT(*) FROM data_node.offchain_data;"
+
+	var count uint64
+	if err := db.pg.QueryRowContext(ctx, countQuery).Scan(&count); err != nil {
+		return 0, err
+	}
+
+	return count, nil
+}
+
 func (db *pgDB) execer(dbTx sqlx.ExecerContext) sqlx.ExecerContext {
 	if dbTx != nil {
 		return dbTx