review

libs/utils/close.go

@@ -6,6 +6,8 @@ import (
 	logging "github.com/ipfs/go-log/v2"
 )
 
+// CloseAndLog closes the closer and logs any error that occurs. The function is handy wrapping
+// to group closing and logging in one call for defer statements.
 func CloseAndLog(log logging.StandardLogger, name string, closer io.Closer) {
 	if err := closer.Close(); err != nil {
 		log.Warnf("closing %s: %s", name, err)

share/shwap/namespace_data.go

@@ -40,20 +40,35 @@ func (nd NamespacedData) Validate(root *share.AxisRoots, namespace share.Namespa
 }
 
 // ReadFrom reads the binary form of NamespacedData from the provided reader.
-func (nd *NamespacedData) ReadFrom(reader io.Reader) error {
+func (nd *NamespacedData) ReadFrom(reader io.Reader) (int64, error) {
 	var data []RowNamespaceData
+	var n int64
 	for {
 		var pbRow pb.RowNamespaceData
-		_, err := serde.Read(reader, &pbRow)
+		nn, err := serde.Read(reader, &pbRow)
+		n += int64(nn)
 		if err != nil {
 			if errors.Is(err, io.EOF) {
 				// all rows have been read
 				*nd = data
-				return nil
+				return n, nil
 			}
-			return err
+			return n, err
 		}
 		row := RowNamespaceDataFromProto(&pbRow)
 		data = append(data, row)
 	}
 }
+
+func (nd NamespacedData) WriteTo(writer io.Writer) (int64, error) {
+	var n int64
+	for _, row := range nd {
+		pbRow := row.ToProto()
+		nn, err := serde.Write(writer, pbRow)
+		n += int64(nn)
+		if err != nil {
+			return n, err
+		}
+	}
+	return n, nil
+}

share/shwap/p2p/shrex/shrexeds/pb/extended_data_square.proto → share/shwap/p2p/shrex/pb/shrex.proto

@@ -1,6 +1,6 @@
 syntax = "proto3";
 
-option go_package = "github.com/celestiaorg/celestia-node/share/shwap/p2p/shrex/shrexeds/pb";
+option go_package = "github.com/celestiaorg/celestia-node/share/shwap/p2p/shrex/pb";
 
 enum Status {
   INVALID = 0;
@@ -9,6 +9,6 @@ enum Status {
   INTERNAL = 3; // internal server error
 }
 
-message EDSResponse {
+message Response {
   Status status = 1;
 }

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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -14,7 +15,6 @@ import (
 	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/atomic"
 
 	libhead "github.com/celestiaorg/go-header"
 	"github.com/celestiaorg/nmt"
@@ -62,7 +62,9 @@ func TestShrexGetter(t *testing.T) {
 	getter := NewGetter(edsClient, ndClient, fullPeerManager, archivalPeerManager, light.Window)
 	require.NoError(t, getter.Start(ctx))
 
-	height := atomic.NewUint64(1)
+	height := atomic.Uint64{}
+	height.Add(1)
+
 	t.Run("ND_Available, total data size > 1mb", func(t *testing.T) {
 		ctx, cancel := context.WithTimeout(ctx, time.Second*10)
 		t.Cleanup(cancel)