integrate shwap into shrex

libs/utils/close.go

@@ -0,0 +1,13 @@
+package utils
+
+import (
+	"io"
+
+	logging "github.com/ipfs/go-log/v2"
+)
+
+func CloseAndLog(log logging.StandardLogger, name string, closer io.Closer) {
+	if err := closer.Close(); err != nil {
+		log.Warnf("closing %s: %s", name, err)
+	}
+}

share/getters/utils.go → libs/utils/ctx.go

@@ -1,24 +1,22 @@
-package getters
+package utils
 
 import (
 	"context"
-	"errors"
 	"time"
-
-	logging "github.com/ipfs/go-log/v2"
-	"go.opentelemetry.io/otel"
 )
 
-var (
-	tracer = otel.Tracer("share/getters")
-	log    = logging.Logger("share/getters")
+// ResetContextOnError returns a fresh context if the given context has an error.
+func ResetContextOnError(ctx context.Context) context.Context {
+	if ctx.Err() != nil {
+		ctx = context.Background()
+	}
 
-	errOperationNotSupported = errors.New("operation is not supported")
-)
+	return ctx
+}
 
-// ctxWithSplitTimeout will split timeout stored in context by splitFactor and return the result if
+// CtxWithSplitTimeout will split timeout stored in context by splitFactor and return the result if
 // it is greater than minTimeout. minTimeout == 0 will be ignored, splitFactor <= 0 will be ignored
-func ctxWithSplitTimeout(
+func CtxWithSplitTimeout(
 	ctx context.Context,
 	splitFactor int,
 	minTimeout time.Duration,
@@ -42,16 +40,3 @@ func ctxWithSplitTimeout(
 	}
 	return context.WithTimeout(ctx, splitTimeout)
 }
-
-// ErrorContains reports whether any error in err's tree matches any error in targets tree.
-func ErrorContains(err, target error) bool {
-	if errors.Is(err, target) || target == nil {
-		return true
-	}
-
-	target = errors.Unwrap(target)
-	if target == nil {
-		return false
-	}
-	return ErrorContains(err, target)
-}

share/getters/utils_test.go → libs/utils/ctx_test.go

@@ -1,119 +1,13 @@
-package getters
+package utils
 
 import (
 	"context"
-	"errors"
-	"fmt"
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-func Test_ErrorContains(t *testing.T) {
-	err1 := errors.New("1")
-	err2 := errors.New("2")
-
-	w1 := func(err error) error {
-		return fmt.Errorf("wrap1: %w", err)
-	}
-	w2 := func(err error) error {
-		return fmt.Errorf("wrap1: %w", err)
-	}
-
-	type args struct {
-		err    error
-		target error
-	}
-	tests := []struct {
-		name string
-		args args
-		want bool
-	}{
-		{
-			"nil err",
-			args{
-				err:    nil,
-				target: err1,
-			},
-			false,
-		},
-		{
-			"nil target",
-			args{
-				err:    err1,
-				target: nil,
-			},
-			true,
-		},
-		{
-			"errors.Is true",
-			args{
-				err:    w1(err1),
-				target: err1,
-			},
-			true,
-		},
-		{
-			"errors.Is false",
-			args{
-				err:    w1(err1),
-				target: err2,
-			},
-			false,
-		},
-		{
-			"same wrap but different base error",
-			args{
-				err:    w1(err1),
-				target: w1(err2),
-			},
-			false,
-		},
-		{
-			"both wrapped true",
-			args{
-				err:    w1(err1),
-				target: w2(err1),
-			},
-			true,
-		},
-		{
-			"both wrapped false",
-			args{
-				err:    w1(err1),
-				target: w2(err2),
-			},
-			false,
-		},
-		{
-			"multierr first in slice",
-			args{
-				err:    errors.Join(w1(err1), w2(err2)),
-				target: w2(err1),
-			},
-			true,
-		},
-		{
-			"multierr second in slice",
-			args{
-				err:    errors.Join(w1(err1), w2(err2)),
-				target: w1(err2),
-			},
-			true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			assert.Equalf(t,
-				tt.want,
-				ErrorContains(tt.args.err, tt.args.target),
-				"ErrorContains(%v, %v)", tt.args.err, tt.args.target)
-		})
-	}
-}
-
 func Test_ctxWithSplitTimeout(t *testing.T) {
 	type args struct {
 		ctxTimeout  time.Duration
@@ -216,7 +110,7 @@ func Test_ctxWithSplitTimeout(t *testing.T) {
 					ctx, cancel = context.WithTimeout(ctx, tt.args.ctxTimeout)
 				}
 				t.Cleanup(cancel)
-				got, _ := ctxWithSplitTimeout(ctx, sf, tt.args.minTimeout)
+				got, _ := CtxWithSplitTimeout(ctx, sf, tt.args.minTimeout)
 				dl, ok := got.Deadline()
 				// in case no deadline is found in ctx or not expected to be found, check both cases apply at the
 				// same time

libs/utils/error.go

@@ -0,0 +1,16 @@
+package utils
+
+import "errors"
+
+// ErrorContains reports whether any error in err's tree matches any error in targets tree.
+func ErrorContains(err, target error) bool {
+	if errors.Is(err, target) || target == nil {
+		return true
+	}
+
+	target = errors.Unwrap(target)
+	if target == nil {
+		return false
+	}
+	return ErrorContains(err, target)
+}

libs/utils/error_test.go

@@ -0,0 +1,112 @@
+package utils
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_ErrorContains(t *testing.T) {
+	err1 := errors.New("1")
+	err2 := errors.New("2")
+
+	w1 := func(err error) error {
+		return fmt.Errorf("wrap1: %w", err)
+	}
+	w2 := func(err error) error {
+		return fmt.Errorf("wrap1: %w", err)
+	}
+
+	type args struct {
+		err    error
+		target error
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			"nil err",
+			args{
+				err:    nil,
+				target: err1,
+			},
+			false,
+		},
+		{
+			"nil target",
+			args{
+				err:    err1,
+				target: nil,
+			},
+			true,
+		},
+		{
+			"errors.Is true",
+			args{
+				err:    w1(err1),
+				target: err1,
+			},
+			true,
+		},
+		{
+			"errors.Is false",
+			args{
+				err:    w1(err1),
+				target: err2,
+			},
+			false,
+		},
+		{
+			"same wrap but different base error",
+			args{
+				err:    w1(err1),
+				target: w1(err2),
+			},
+			false,
+		},
+		{
+			"both wrapped true",
+			args{
+				err:    w1(err1),
+				target: w2(err1),
+			},
+			true,
+		},
+		{
+			"both wrapped false",
+			args{
+				err:    w1(err1),
+				target: w2(err2),
+			},
+			false,
+		},
+		{
+			"multierr first in slice",
+			args{
+				err:    errors.Join(w1(err1), w2(err2)),
+				target: w2(err1),
+			},
+			true,
+		},
+		{
+			"multierr second in slice",
+			args{
+				err:    errors.Join(w1(err1), w2(err2)),
+				target: w1(err2),
+			},
+			true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equalf(t,
+				tt.want,
+				ErrorContains(tt.args.err, tt.args.target),
+				"ErrorContains(%v, %v)", tt.args.err, tt.args.target)
+		})
+	}
+}

nodebuilder/share/constructors.go

@@ -11,6 +11,7 @@ import (
 	"github.com/celestiaorg/celestia-node/share/eds"
 	"github.com/celestiaorg/celestia-node/share/getters"
 	"github.com/celestiaorg/celestia-node/share/ipld"
+	"github.com/celestiaorg/celestia-node/share/shwap/p2p/shrex/shrex_getter"
 )
 
 func newShareModule(getter share.Getter, avail share.Availability) Module {
@@ -42,7 +43,7 @@ func ensureEmptyEDSInBS(ctx context.Context, bServ blockservice.BlockService) er
 }
 
 func lightGetter(
-	shrexGetter *getters.ShrexGetter,
+	shrexGetter *shrex_getter.Getter,
 	ipldGetter *getters.IPLDGetter,
 	cfg Config,
 ) share.Getter {
@@ -54,13 +55,13 @@ func lightGetter(
 	return getters.NewCascadeGetter(cascade)
 }
 
-// ShrexGetter is added to bridge nodes for the case that a shard is removed
+// Getter is added to bridge nodes for the case that a shard is removed
 // after detected shard corruption. This ensures the block is fetched and stored
 // by shrex the next time the data is retrieved (meaning shard recovery is
 // manual after corruption is detected).
 func bridgeGetter(
 	storeGetter *getters.StoreGetter,
-	shrexGetter *getters.ShrexGetter,
+	shrexGetter *shrex_getter.Getter,
 	cfg Config,
 ) share.Getter {
 	var cascade []share.Getter
@@ -73,7 +74,7 @@ func bridgeGetter(
 
 func fullGetter(
 	storeGetter *getters.StoreGetter,
-	shrexGetter *getters.ShrexGetter,
+	shrexGetter *shrex_getter.Getter,
 	ipldGetter *getters.IPLDGetter,
 	cfg Config,
 ) share.Getter {