feat(modshare): implement GetRow (#4002)

Co-authored-by: Hlib Kanunnikov <[email protected]>

nodebuilder/share/share.go

@@ -50,6 +50,8 @@ type Module interface {
 	GetSamples(ctx context.Context, header *header.ExtendedHeader, indices []shwap.SampleCoords) ([]shwap.Sample, error)
 	// GetEDS gets the full EDS identified by the given extended header.
 	GetEDS(ctx context.Context, height uint64) (*rsmt2d.ExtendedDataSquare, error)
+	// GetRow gets all shares from specified row.
+	GetRow(context.Context, uint64, int) (shwap.Row, error)
 	// GetNamespaceData gets all shares from an EDS within the given namespace.
 	// Shares are returned in a row-by-row order if the namespace spans multiple rows.
 	GetNamespaceData(
@@ -77,6 +79,11 @@ type API struct {
 			ctx context.Context,
 			height uint64,
 		) (*rsmt2d.ExtendedDataSquare, error) `perm:"read"`
+		GetRow func(
+			context.Context,
+			uint64,
+			int,
+		) (shwap.Row, error) `perm:"read"`
 		GetNamespaceData func(
 			ctx context.Context,
 			height uint64,
@@ -108,6 +115,10 @@ func (api *API) GetEDS(ctx context.Context, height uint64) (*rsmt2d.ExtendedData
 	return api.Internal.GetEDS(ctx, height)
 }
 
+func (api *API) GetRow(ctx context.Context, height uint64, rowIdx int) (shwap.Row, error) {
+	return api.Internal.GetRow(ctx, height, rowIdx)
+}
+
 func (api *API) GetRange(ctx context.Context, height uint64, start, end int) (*GetRangeResult, error) {
 	return api.Internal.GetRange(ctx, height, start, end)
 }
@@ -196,3 +207,11 @@ func (m module) GetNamespaceData(
 	}
 	return m.getter.GetNamespaceData(ctx, header, namespace)
 }
+
+func (m module) GetRow(ctx context.Context, height uint64, rowIdx int) (shwap.Row, error) {
+	header, err := m.hs.GetByHeight(ctx, height)
+	if err != nil {
+		return shwap.Row{}, err
+	}
+	return m.getter.GetRow(ctx, header, rowIdx)
+}

share/availability/light/availability_test.go

@@ -290,7 +290,7 @@ func (g successGetter) checkOnce(t *testing.T) {
 	}
 }
 
-func (g successGetter) GetSamples(_ context.Context, hdr *header.ExtendedHeader,
+func (g successGetter) GetSamples(_ context.Context, _ *header.ExtendedHeader,
 	indices []shwap.SampleCoords,
 ) ([]shwap.Sample, error) {
 	g.Lock()
@@ -305,6 +305,10 @@ func (g successGetter) GetSamples(_ context.Context, hdr *header.ExtendedHeader,
 	return smpls, nil
 }
 
+func (g successGetter) GetRow(_ context.Context, _ *header.ExtendedHeader, _ int) (shwap.Row, error) {
+	panic("not implemented")
+}
+
 func (g successGetter) GetEDS(_ context.Context, _ *header.ExtendedHeader) (*rsmt2d.ExtendedDataSquare, error) {
 	panic("not implemented")
 }

share/shwap/getter.go

@@ -39,6 +39,8 @@ type Getter interface {
 	// GetEDS gets the full EDS identified by the given extended header.
 	GetEDS(context.Context, *header.ExtendedHeader) (*rsmt2d.ExtendedDataSquare, error)
 
+	// GetRow gets Row by its index committed to the given extended header.
+	GetRow(ctx context.Context, header *header.ExtendedHeader, rowIdx int) (Row, error)
 	// GetNamespaceData gets all shares from an EDS within the given namespace.
 	// Shares are returned in a row-by-row order if the namespace spans multiple rows.
 	// Inclusion of returned data could be verified using Verify method on NamespacedShares.

share/shwap/getters/cascade.go

@@ -71,6 +71,18 @@ func (cg *CascadeGetter) GetEDS(
 	return cascadeGetters(ctx, cg.getters, get)
 }
 
+// GetRow gets row shares from any of registered shwap.Getters in cascading
+// order.
+func (cg *CascadeGetter) GetRow(ctx context.Context, header *header.ExtendedHeader, rowIdx int) (shwap.Row, error) {
+	ctx, span := tracer.Start(ctx, "cascade/get-row")
+	defer span.End()
+
+	get := func(ctx context.Context, get shwap.Getter) (shwap.Row, error) {
+		return get.GetRow(ctx, header, rowIdx)
+	}
+	return cascadeGetters(ctx, cg.getters, get)
+}
+
 // GetNamespaceData gets NamespacedShares from any of registered shwap.Getters in cascading
 // order.
 func (cg *CascadeGetter) GetNamespaceData(

share/shwap/getters/testing.go

@@ -58,6 +58,23 @@ func (seg *SingleEDSGetter) GetSamples(ctx context.Context, hdr *header.Extended
 	return smpls, nil
 }
 
+func (seg *SingleEDSGetter) GetRow(
+	ctx context.Context,
+	header *header.ExtendedHeader,
+	rowIdx int,
+) (shwap.Row, error) {
+	err := seg.checkRoots(header.DAH)
+	if err != nil {
+		return shwap.Row{}, err
+	}
+
+	axisHalf, err := seg.EDS.AxisHalf(ctx, rsmt2d.Row, rowIdx)
+	if err != nil {
+		return shwap.Row{}, err
+	}
+	return axisHalf.ToRow(), nil
+}
+
 // GetEDS returns a kept EDS if the correct root is given.
 func (seg *SingleEDSGetter) GetEDS(
 	_ context.Context,

share/shwap/p2p/bitswap/getter.go

@@ -134,6 +134,32 @@ func (g *Getter) GetSamples(
 	return smpls, nil
 }
 
+func (g *Getter) GetRow(ctx context.Context, hdr *header.ExtendedHeader, rowIdx int) (shwap.Row, error) {
+	ctx, span := tracer.Start(ctx, "get-eds")
+	defer span.End()
+
+	blk, err := NewEmptyRowBlock(hdr.Height(), rowIdx, len(hdr.DAH.RowRoots))
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, "NewEmptyRowBlock")
+		return shwap.Row{}, err
+	}
+
+	isArchival := g.isArchival(hdr)
+	span.SetAttributes(attribute.Bool("is_archival", isArchival))
+
+	ses, release := g.getSession(isArchival)
+	defer release()
+
+	err = Fetch(ctx, g.exchange, hdr.DAH, []Block{blk}, WithFetcher(ses))
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, "Fetch")
+		return shwap.Row{}, err
+	}
+	return blk.Container, nil
+}
+
 // GetEDS uses [RowBlock] and [Fetch] to get half of the first EDS quadrant(ODS) and
 // recomputes the whole EDS from it.
 // We fetch the ODS or Q1 to ensure better compatibility with archival nodes that only

share/shwap/p2p/shrex/shrex_getter/shrex.go

@@ -150,6 +150,10 @@ func (sg *Getter) GetSamples(context.Context, *header.ExtendedHeader, []shwap.Sa
 	return nil, fmt.Errorf("getter/shrex: GetShare %w", shwap.ErrOperationNotSupported)
 }
 
+func (sg *Getter) GetRow(_ context.Context, _ *header.ExtendedHeader, _ int) (shwap.Row, error) {
+	return shwap.Row{}, fmt.Errorf("getter/shrex: GetRow %w", shwap.ErrOperationNotSupported)
+}
+
 func (sg *Getter) GetEDS(ctx context.Context, header *header.ExtendedHeader) (*rsmt2d.ExtendedDataSquare, error) {
 	var err error
 	ctx, span := tracer.Start(ctx, "shrex/get-eds")

share/shwap/row.go

@@ -2,6 +2,7 @@ package shwap
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 
 	"github.com/celestiaorg/celestia-app/v3/pkg/wrapper"
@@ -171,6 +172,33 @@ func (r *Row) verifyInclusion(roots *share.AxisRoots, idx int) error {
 	return nil
 }
 
+// MarshalJSON encodes row to the json encoded bytes.
+func (r Row) MarshalJSON() ([]byte, error) {
+	jsonRow := struct {
+		Shares []libshare.Share `json:"shares"`
+		Side   string           `json:"side"`
+	}{
+		Shares: r.shares,
+		Side:   r.side.String(),
+	}
+	return json.Marshal(&jsonRow)
+}
+
+// UnmarshalJSON decodes json bytes to the row.
+func (r *Row) UnmarshalJSON(data []byte) error {
+	jsonRow := struct {
+		Shares []libshare.Share `json:"shares"`
+		Side   string           `json:"side"`
+	}{}
+	err := json.Unmarshal(data, &jsonRow)
+	if err != nil {
+		return err
+	}
+	r.shares = jsonRow.Shares
+	r.side = toRowSide(jsonRow.Side)
+	return nil
+}
+
 // ToProto converts a RowSide to its protobuf representation.
 func (s RowSide) ToProto() pb.Row_HalfSide {
 	if s == Left {
@@ -186,3 +214,29 @@ func sideFromProto(side pb.Row_HalfSide) RowSide {
 	}
 	return Right
 }
+
+func (s RowSide) String() string {
+	switch s {
+	case Left:
+		return "LEFT"
+	case Right:
+		return "RIGHT"
+	case Both:
+		return "BOTH"
+	default:
+		panic("invalid row side")
+	}
+}
+
+func toRowSide(s string) RowSide {
+	switch s {
+	case "LEFT":
+		return Left
+	case "RIGHT":
+		return Right
+	case "BOTH":
+		return Both
+	default:
+		panic("invalid row side")
+	}
+}

share/shwap/row_test.go

@@ -1,6 +1,7 @@
 package shwap
 
 import (
+	"encoding/json"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -29,6 +30,33 @@ func TestRowShares(t *testing.T) {
 	}
 }
 
+func TestRowMarshal(t *testing.T) {
+	const odsSize = 8
+	eds := edstest.RandEDS(t, odsSize)
+	for rowIdx := 0; rowIdx < odsSize*2; rowIdx++ {
+		for _, side := range []RowSide{Left, Right, Both} {
+			row, err := RowFromEDS(eds, rowIdx, side)
+			require.NoError(t, err)
+			rowData, err := json.Marshal(row)
+			require.NoError(t, err)
+
+			decodedRow := &Row{}
+			err = json.Unmarshal(rowData, decodedRow)
+			require.NoError(t, err)
+
+			require.Equal(t, side, decodedRow.side)
+			extended, err := decodedRow.Shares()
+			require.NoError(t, err)
+
+			shares, err := row.Shares()
+			require.NoError(t, err)
+
+			require.Equal(t, shares, extended)
+			require.Equal(t, row.side, decodedRow.side)
+		}
+	}
+}
+
 func TestRowValidate(t *testing.T) {
 	const odsSize = 8
 	eds := edstest.RandEDS(t, odsSize)

store/getter.go

@@ -71,6 +71,21 @@ func (g *Getter) GetEDS(ctx context.Context, h *header.ExtendedHeader) (*rsmt2d.
 	return rsmt2d.ExtendedDataSquare, nil
 }
 
+func (g *Getter) GetRow(ctx context.Context, h *header.ExtendedHeader, rowIdx int) (shwap.Row, error) {
+	acc, err := g.store.GetByHeight(ctx, h.Height())
+	if err != nil {
+		if errors.Is(err, ErrNotFound) {
+			return shwap.Row{}, shwap.ErrNotFound
+		}
+		return shwap.Row{}, fmt.Errorf("getting accessor from store: %w", err)
+	}
+	axisHalf, err := acc.AxisHalf(ctx, rsmt2d.Row, rowIdx)
+	if err != nil {
+		return shwap.Row{}, fmt.Errorf("getting axis half from accessor: %w", err)
+	}
+	return axisHalf.ToRow(), nil
+}
+
 func (g *Getter) GetNamespaceData(
 	ctx context.Context,
 	h *header.ExtendedHeader,

store/getter_test.go

@@ -69,6 +69,26 @@ func TestStoreGetter(t *testing.T) {
 		require.ErrorIs(t, err, shwap.ErrNotFound)
 	})
 
+	t.Run("GetRow", func(t *testing.T) {
+		eds, roots := randomEDS(t)
+		eh := headertest.RandExtendedHeaderWithRoot(t, roots)
+		height := height.Add(1)
+		eh.RawHeader.Height = int64(height)
+
+		err := edsStore.PutODSQ4(ctx, eh.DAH, height, eds)
+		require.NoError(t, err)
+
+		for i := 0; i < len(eh.DAH.RowRoots); i++ {
+			row, err := sg.GetRow(ctx, eh, i)
+			require.NoError(t, err)
+			retreivedShrs, err := row.Shares()
+			require.NoError(t, err)
+			edsShrs, err := libshare.FromBytes(eds.Row(uint(i)))
+			require.NoError(t, err)
+			require.Equal(t, edsShrs, retreivedShrs)
+		}
+	})
+
 	t.Run("GetNamespaceData", func(t *testing.T) {
 		ns := libshare.RandomNamespace()
 		eds, roots := edstest.RandEDSWithNamespace(t, ns, 8, 16)