remove submission path from blob service

Author: vgonkivs

api/client/client.go

@@ -9,10 +9,7 @@ import (
 	logging "github.com/ipfs/go-log/v2"
 	"google.golang.org/grpc"
 
-	appfibre "github.com/celestiaorg/celestia-app/v8/fibre"
-
 	"github.com/celestiaorg/celestia-node/blob"
-	"github.com/celestiaorg/celestia-node/fibre"
 	"github.com/celestiaorg/celestia-node/nodebuilder/p2p"
 	stateapi "github.com/celestiaorg/celestia-node/nodebuilder/state"
 	"github.com/celestiaorg/celestia-node/state"
@@ -142,24 +139,8 @@ func (c *Client) initTxClient(
 	}
 	c.State = core
 
-	// setup fibre client if the fibre key is available in the keyring
-	var fibreSvc *fibre.Service
-	fibreCfg := appfibre.DefaultClientConfig()
-	fibreCfg.StateAddress = conn.Target()
-	fibreClient, err := appfibre.NewClient(kr, fibreCfg)
-	if err != nil {
-		log.Warnw("fibre client not available, fibre blob submission disabled", "err", err)
-	} else {
-		if err := fibreClient.Start(ctx); err != nil {
-			log.Warnw("failed to start fibre client, fibre blob submission disabled", "err", err)
-		} else {
-			acc := fibre.NewAccountClient(tc, conn)
-			fibreSvc = fibre.NewService(fibreClient, tc, acc)
-		}
-	}
-
 	// setup blob submission service using core
-	blobSvc := blob.NewService(core, fibreSvc, nil, nil, nil)
+	blobSvc := blob.NewService(core, nil, nil, nil)
 	err = blobSvc.Start(ctx)
 	if err != nil {
 		_ = core.Stop(ctx)
@@ -180,11 +161,6 @@ func (c *Client) initTxClient(
 		if err != nil {
 			return fmt.Errorf("failed to stop blob service: %w", err)
 		}
-		if fibreClient != nil {
-			if stopErr := fibreClient.Stop(ctx); stopErr != nil {
-				log.Warnw("failed to stop fibre client", "err", stopErr)
-			}
-		}
 		err = core.Stop(ctx)
 		if err != nil {
 			return fmt.Errorf("failed to stop core accessor: %w", err)

blob/service.go

@@ -17,7 +17,6 @@ import (
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/trace"
 
-	appfibre "github.com/celestiaorg/celestia-app/v8/fibre"
 	"github.com/celestiaorg/celestia-app/v8/pkg/appconsts"
 	pkgproof "github.com/celestiaorg/celestia-app/v8/pkg/proof"
 	"github.com/celestiaorg/go-square/v4/inclusion"
@@ -30,13 +29,11 @@ import (
 	"github.com/celestiaorg/celestia-node/share"
 	"github.com/celestiaorg/celestia-node/share/shwap"
 	"github.com/celestiaorg/celestia-node/state"
-	"github.com/celestiaorg/celestia-node/state/txclient"
 )
 
 var (
-	ErrBlobNotFound   = errors.New("blob: not found")
-	ErrInvalidProof   = errors.New("blob: invalid proof")
-	ErrMixedBlobTypes = errors.New("blob: cannot mix fibre (v2) and regular blobs in a single submit")
+	ErrBlobNotFound = errors.New("blob: not found")
+	ErrInvalidProof = errors.New("blob: invalid proof")
 
 	log    = logging.Logger("blob")
 	tracer = otel.Tracer("blob/service")
@@ -53,28 +50,12 @@ type Submitter interface {
 	SubmitPayForBlob(context.Context, []*libshare.Blob, *state.TxConfig) (*types.TxResponse, error)
 }
 
-// FibreSubmitter is an interface for submitting blobs to the Fibre network.
-type FibreSubmitter interface {
-	Submit(
-		ctx context.Context,
-		ns libshare.Namespace,
-		data []byte,
-		config *txclient.TxConfig,
-	) (
-		_ *appfibre.PutResult,
-		_ *appfibre.PaymentPromise,
-		err error,
-	)
-}
-
 type Service struct {
 	// ctx represents the Service's lifecycle context.
 	ctx    context.Context
 	cancel context.CancelFunc
 	// accessor dials the given celestia-core endpoint to submit blobs.
 	blobSubmitter Submitter
-	// fibreSubmitter submits blobs to the Fibre network.
-	fibreSubmitter FibreSubmitter
 	// shareGetter retrieves the EDS to fetch all shares from the requested header.
 	shareGetter shwap.Getter
 	// headerGetter fetches header by the provided height
@@ -87,18 +68,16 @@ type Service struct {
 
 func NewService(
 	submitter Submitter,
-	fibreSubmitter FibreSubmitter,
 	getter shwap.Getter,
 	headerGetter func(context.Context, uint64) (*header.ExtendedHeader, error),
 	headerSub func(ctx context.Context) (<-chan *header.ExtendedHeader, error),
 ) *Service {
 	return &Service{
-		blobSubmitter:  submitter,
-		fibreSubmitter: fibreSubmitter,
-		shareGetter:    getter,
-		headerGetter:   headerGetter,
-		headerSub:      headerSub,
-		metrics:        nil, // Will be initialized via WithMetrics() if needed
+		blobSubmitter: submitter,
+		shareGetter:   getter,
+		headerGetter:  headerGetter,
+		headerSub:     headerSub,
+		metrics:       nil, // Will be initialized via WithMetrics() if needed
 	}
 }
 
@@ -197,14 +176,10 @@ func (s *Service) Subscribe(ctx context.Context, ns libshare.Namespace) (<-chan
 	return blobCh, nil
 }
 
-// Submit sends PFB/PFF transaction and reports the height at which it was included.
+// Submit sends PFB transaction and reports the height at which it was included.
 // Allows sending multiple Blobs atomically synchronously.
 // Uses default wallet registered on the Node.
 // Handles gas estimation and fee calculation.
-//
-// If all blobs are fibre blobs (share version 2), they are submitted via the fibre network.
-// If all blobs are regular blobs, they are submitted via the standard PFB path.
-// Mixing fibre and regular blobs in a single submit is not allowed.
 func (s *Service) Submit(ctx context.Context, blobs []*Blob, txConfig *SubmitOptions) (_ uint64, err error) {
 	ctx, span := tracer.Start(ctx, "blob/submit")
 	defer func() {
@@ -216,46 +191,21 @@ func (s *Service) Submit(ctx context.Context, blobs []*Blob, txConfig *SubmitOpt
 		}
 	}()
 
-	if len(blobs) == 0 {
-		return 0, errors.New("no blobs provided")
-	}
-
-	allFibre := slices.IndexFunc(blobs, func(b *Blob) bool { return !b.IsFibreBlob() }) == -1
-	anyFibre := slices.IndexFunc(blobs, func(b *Blob) bool { return b.IsFibreBlob() }) != -1
-
-	if anyFibre && !allFibre {
-		return 0, ErrMixedBlobTypes
-	}
-
-	if allFibre {
-		return s.submitFibreBlobs(ctx, blobs, txConfig)
-	}
-
 	libBlobs := make([]*libshare.Blob, len(blobs))
 	for i := range blobs {
+		if blobs[i].IsFibreBlob() {
+			return 0, errors.New("cannot submit fibre blob. please use Fibre Submit instead")
+		}
 		libBlobs[i] = blobs[i].Blob
 	}
-	resp, err := s.blobSubmitter.SubmitPayForBlob(ctx, libBlobs, txConfig)
+
+	spanCtx := trace.ContextWithSpan(ctx, span)
+	resp, err := s.blobSubmitter.SubmitPayForBlob(spanCtx, libBlobs, txConfig)
 	if err != nil {
 		return 0, err
 	}
-	return uint64(resp.Height), nil
-}
-
-func (s *Service) submitFibreBlobs(ctx context.Context, blobs []*Blob, txConfig *SubmitOptions) (uint64, error) {
-	if s.fibreSubmitter == nil {
-		return 0, fmt.Errorf("fibre submitter is not available")
-	}
 
-	var height uint64
-	for _, blob := range blobs {
-		res, _, err := s.fibreSubmitter.Submit(ctx, blob.Namespace(), blob.Data(), txConfig)
-		if err != nil {
-			return 0, err
-		}
-		height = res.Height
-	}
-	return height, nil
+	return uint64(resp.Height), nil
 }
 
 // Get retrieves a blob in a given namespace at the given height by commitment.

blob/service_test.go

@@ -14,14 +14,12 @@ import (
 	"time"
 
 	tmrand "github.com/cometbft/cometbft/libs/rand"
-	"github.com/cosmos/cosmos-sdk/types"
 	"github.com/golang/mock/gomock"
 	ds "github.com/ipfs/go-datastore"
 	ds_sync "github.com/ipfs/go-datastore/sync"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	appfibre "github.com/celestiaorg/celestia-app/v8/fibre"
 	"github.com/celestiaorg/celestia-app/v8/pkg/wrapper"
 	"github.com/celestiaorg/go-header/store"
 	libshare "github.com/celestiaorg/go-square/v4/share"
@@ -36,8 +34,6 @@ import (
 	"github.com/celestiaorg/celestia-node/share/ipld"
 	"github.com/celestiaorg/celestia-node/share/shwap"
 	"github.com/celestiaorg/celestia-node/share/shwap/getters/mock"
-	"github.com/celestiaorg/celestia-node/state"
-	"github.com/celestiaorg/celestia-node/state/txclient"
 )
 
 func TestBlobService_Get(t *testing.T) {
@@ -904,7 +900,7 @@ func createServiceWithSub(ctx context.Context, t testing.TB, blobs []*Blob) (*Se
 			nd, err := eds.NamespaceData(ctx, accessor, ns)
 			return nd, err
 		})
-	return NewService(nil, nil, shareGetter, fn, fn2), headers
+	return NewService(nil, shareGetter, fn, fn2), headers
 }
 
 func createService(ctx context.Context, t testing.TB, shares []libshare.Share) *Service {
@@ -948,7 +944,7 @@ func createService(ctx context.Context, t testing.TB, shares []libshare.Share) *
 	fn2 := func(ctx context.Context) (<-chan *header.ExtendedHeader, error) {
 		return nil, fmt.Errorf("not implemented")
 	}
-	return NewService(nil, nil, shareGetter, fn, fn2)
+	return NewService(nil, shareGetter, fn, fn2)
 }
 
 // TestProveCommitmentAllCombinations tests proving all the commitments in a block.
@@ -1157,127 +1153,6 @@ func buildEDS(t *testing.T, shares []libshare.Share) (*rsmt2d.ExtendedDataSquare
 	return eds, roots.Hash()
 }
 
-// mockSubmitter implements the Submitter interface for testing.
-type mockSubmitter struct {
-	height int64
-	err    error
-	called bool
-}
-
-func (m *mockSubmitter) SubmitPayForBlob(
-	context.Context,
-	[]*libshare.Blob,
-	*state.TxConfig,
-) (*types.TxResponse, error) {
-	m.called = true
-	if m.err != nil {
-		return nil, m.err
-	}
-	return &types.TxResponse{Height: m.height}, nil
-}
-
-// mockFibreSubmitter implements the FibreSubmitter interface for testing.
-type mockFibreSubmitter struct {
-	height uint64
-	err    error
-	calls  int
-}
-
-func (m *mockFibreSubmitter) Submit(context.Context, libshare.Namespace, []byte, *txclient.TxConfig,
-) (*appfibre.PutResult, *appfibre.PaymentPromise, error) {
-	m.calls++
-	if m.err != nil {
-		return nil, nil, m.err
-	}
-	return &appfibre.PutResult{Height: m.height}, nil, nil
-}
-
-func newTestFibreBlob(t *testing.T) *Blob {
-	t.Helper()
-	ns := libshare.RandomBlobNamespace()
-	commitment := bytes.Repeat([]byte{0xAA}, libshare.FibreCommitmentSize)
-	signer := bytes.Repeat([]byte{0xBB}, 20)
-	libBlob, err := libshare.NewV2Blob(ns, 0, commitment, signer)
-	require.NoError(t, err)
-	blob, err := convertBlobs(libBlob)
-	require.NoError(t, err)
-	return blob[0]
-}
-
-func TestService_Submit_AllFibreBlobs(t *testing.T) {
-	fibreSub := &mockFibreSubmitter{height: 42}
-	blobSub := &mockSubmitter{height: 100}
-
-	svc := &Service{
-		blobSubmitter:  blobSub,
-		fibreSubmitter: fibreSub,
-	}
-	svc.ctx, svc.cancel = context.WithCancel(context.Background())
-	defer svc.cancel()
-
-	blobs := []*Blob{newTestFibreBlob(t), newTestFibreBlob(t)}
-	height, err := svc.Submit(context.Background(), blobs, nil)
-	require.NoError(t, err)
-	assert.Equal(t, uint64(42), height)
-	assert.Equal(t, 2, fibreSub.calls)
-	assert.False(t, blobSub.called)
-}
-
-func TestService_Submit_AllRegularBlobs(t *testing.T) {
-	fibreSub := &mockFibreSubmitter{height: 42}
-	blobSub := &mockSubmitter{height: 100}
-
-	svc := &Service{
-		blobSubmitter:  blobSub,
-		fibreSubmitter: fibreSub,
-	}
-	svc.ctx, svc.cancel = context.WithCancel(context.Background())
-	defer svc.cancel()
-
-	libBlobs, err := libshare.GenerateV0Blobs([]int{10, 10}, false)
-	require.NoError(t, err)
-	blobs, err := convertBlobs(libBlobs...)
-	require.NoError(t, err)
-
-	height, err := svc.Submit(context.Background(), blobs, nil)
-	require.NoError(t, err)
-	assert.Equal(t, uint64(100), height)
-	assert.True(t, blobSub.called)
-	assert.Equal(t, 0, fibreSub.calls)
-}
-
-func TestService_Submit_MixedBlobsError(t *testing.T) {
-	svc := &Service{
-		blobSubmitter:  &mockSubmitter{height: 100},
-		fibreSubmitter: &mockFibreSubmitter{height: 42},
-	}
-	svc.ctx, svc.cancel = context.WithCancel(context.Background())
-	defer svc.cancel()
-
-	libBlobs, err := libshare.GenerateV0Blobs([]int{10}, false)
-	require.NoError(t, err)
-	regularBlobs, err := convertBlobs(libBlobs...)
-	require.NoError(t, err)
-
-	mixed := []*Blob{regularBlobs[0], newTestFibreBlob(t)}
-	_, err = svc.Submit(context.Background(), mixed, nil)
-	require.ErrorIs(t, err, ErrMixedBlobTypes)
-}
-
-func TestService_Submit_FibreWithoutSubmitter(t *testing.T) {
-	svc := &Service{
-		blobSubmitter:  &mockSubmitter{height: 100},
-		fibreSubmitter: nil,
-	}
-	svc.ctx, svc.cancel = context.WithCancel(context.Background())
-	defer svc.cancel()
-
-	blobs := []*Blob{newTestFibreBlob(t)}
-	_, err := svc.Submit(context.Background(), blobs, nil)
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "fibre submitter is not available")
-}
-
 func rawBlobSize(totalSize int) int {
 	return totalSize - delimLen(uint64(totalSize))
 }