Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(modshare): implement GetRow #4002

Merged
merged 8 commits into from
Jan 9, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions nodebuilder/share/mocks/api.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

19 changes: 19 additions & 0 deletions nodebuilder/share/share.go
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
}
6 changes: 5 additions & 1 deletion share/availability/light/availability_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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")
}
Expand Down
2 changes: 2 additions & 0 deletions share/shwap/getter.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
12 changes: 12 additions & 0 deletions share/shwap/getters/cascade.go
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
15 changes: 15 additions & 0 deletions share/shwap/getters/mock/getter.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

17 changes: 17 additions & 0 deletions share/shwap/getters/testing.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
26 changes: 26 additions & 0 deletions share/shwap/p2p/bitswap/getter.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
vgonkivs marked this conversation as resolved.
Show resolved Hide resolved

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
Expand Down
4 changes: 4 additions & 0 deletions share/shwap/p2p/shrex/shrex_getter/shrex.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
54 changes: 54 additions & 0 deletions share/shwap/row.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package shwap

import (
"bytes"
"encoding/json"
"fmt"

"github.com/celestiaorg/celestia-app/v3/pkg/wrapper"
Expand Down Expand Up @@ -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 {
Expand All @@ -186,3 +214,29 @@ func sideFromProto(side pb.Row_HalfSide) RowSide {
}
return Right
}

func (s RowSide) String() string {
switch s {
case Left:
return "LEFT"
Wondertan marked this conversation as resolved.
Show resolved Hide resolved
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")
}
}
28 changes: 28 additions & 0 deletions share/shwap/row_test.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package shwap

import (
"encoding/json"
"testing"

"github.com/stretchr/testify/require"
Expand Down Expand Up @@ -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)
Expand Down
15 changes: 15 additions & 0 deletions store/getter.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
20 changes: 20 additions & 0 deletions store/getter_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading