refactoring

Author: vgonkivs

api/client/client.go

@@ -9,7 +9,10 @@ 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"
@@ -139,8 +142,24 @@ 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, nil, nil, nil, nil)
+	blobSvc := blob.NewService(core, fibreSvc, nil, nil, nil)
 	err = blobSvc.Start(ctx)
 	if err != nil {
 		_ = core.Stop(ctx)
@@ -161,6 +180,11 @@ 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)

api/client/client_test.go

@@ -12,12 +12,14 @@ import (
 	"time"
 
 	coregrpc "github.com/cometbft/cometbft/rpc/grpc"
+	"github.com/cosmos/cosmos-sdk/crypto/hd"
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
 	"github.com/cristalhq/jwt/v5"
 	gomock "github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/fx"
 
+	appfibre "github.com/celestiaorg/celestia-app/v8/fibre"
 	"github.com/celestiaorg/celestia-app/v8/test/util/testnode"
 	libshare "github.com/celestiaorg/go-square/v4/share"
 
@@ -396,6 +398,9 @@ func bridgeNode(t *testing.T, ctx context.Context, cctx testnode.Context) (*node
 	}
 	kr, err := KeyringWithNewKey(keysCfg, tempDir)
 	require.NoError(t, err)
+	// add the fibre key required by the fibre module
+	_, _, err = kr.NewMnemonic(appfibre.DefaultKeyName, keyring.English, "", "", hd.Secp256k1)
+	require.NoError(t, err)
 
 	bn, err := nodebuilder.New(node.Bridge, p2p.Private, store,
 		addAuth,

api/docgen/examples.go

@@ -31,6 +31,7 @@ import (
 	"github.com/celestiaorg/celestia-node/das"
 	"github.com/celestiaorg/celestia-node/fibre"
 	"github.com/celestiaorg/celestia-node/header"
+	fibre2 "github.com/celestiaorg/celestia-node/nodebuilder/fibre"
 	"github.com/celestiaorg/celestia-node/nodebuilder/node"
 	"github.com/celestiaorg/celestia-node/share"
 	"github.com/celestiaorg/celestia-node/share/eds/byzantine"
@@ -196,10 +197,10 @@ func init() {
 	copy(fibreCommitment[:], commitment)
 	add(fibreCommitment)
 
-	exampleUploadResult := fibre.UploadResult{
+	exampleUploadResult := fibre2.UploadResult{
 		Commitment:          fibreCommitment,
-		ValidatorSignatures: []fibre.ValidatorSignature{[]byte("validator_signature_bytes")},
-		PaymentPromise: &fibre.PaymentPromise{
+		ValidatorSignatures: []fibre2.ValidatorSignature{[]byte("validator_signature_bytes")},
+		PaymentPromise: &fibre2.PaymentPromise{
 			ChainID:           "celestia",
 			Namespace:         namespace,
 			BlobSize:          1024,
@@ -212,16 +213,16 @@ func init() {
 	}
 	add(&exampleUploadResult)
 
-	exampleSubmitResult := fibre.SubmitResult{
+	exampleSubmitResult := fibre2.SubmitResult{
 		Commitment:          fibreCommitment,
-		ValidatorSignatures: []fibre.ValidatorSignature{[]byte("validator_signature_bytes")},
+		ValidatorSignatures: []fibre2.ValidatorSignature{[]byte("validator_signature_bytes")},
 		Height:              42,
 		TxHash:              "A5CF62609391B17E0340A6E07BD15860AFA4BE7F5DAF28F2E22A1C3B0CE85E64",
 		PaymentPromise:      exampleUploadResult.PaymentPromise,
 	}
 	add(&exampleSubmitResult)
 
-	add(&fibre.GetBlobResponse{
+	add(&fibre2.GetBlobResult{
 		Data: []byte("fibre blob data"),
 	})
 
@@ -238,7 +239,7 @@ func init() {
 		AvailableTimestamp: time.Date(2025, 1, 8, 0, 0, 0, 0, time.UTC),
 	})
 
-	add(fibre.PaymentPromise{
+	add(fibre2.PaymentPromise{
 		ChainID:           "celestia",
 		Namespace:         namespace,
 		BlobSize:          1024,

blob/service.go

@@ -17,14 +17,14 @@ 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"
 	libshare "github.com/celestiaorg/go-square/v4/share"
 	"github.com/celestiaorg/nmt"
 	"github.com/celestiaorg/rsmt2d"
 
-	"github.com/celestiaorg/celestia-node/fibre"
 	"github.com/celestiaorg/celestia-node/header"
 	"github.com/celestiaorg/celestia-node/libs/utils"
 	"github.com/celestiaorg/celestia-node/share"
@@ -34,8 +34,9 @@ import (
 )
 
 var (
-	ErrBlobNotFound = errors.New("blob: not found")
-	ErrInvalidProof = errors.New("blob: invalid proof")
+	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")
 
 	log    = logging.Logger("blob")
 	tracer = otel.Tracer("blob/service")
@@ -54,7 +55,16 @@ type Submitter interface {
 
 // 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) (*fibre.SubmitResult, error)
+	Submit(
+		ctx context.Context,
+		ns libshare.Namespace,
+		data []byte,
+		config *txclient.TxConfig,
+	) (
+		_ *appfibre.PutResult,
+		_ *appfibre.PaymentPromise,
+		err error,
+	)
 }
 
 type Service struct {
@@ -187,10 +197,14 @@ func (s *Service) Subscribe(ctx context.Context, ns libshare.Namespace) (<-chan
 	return blobCh, nil
 }
 
-// Submit sends PFB transaction and reports the height at which it was included.
+// Submit sends PFB/PFF 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() {
@@ -202,6 +216,17 @@ func (s *Service) Submit(ctx context.Context, blobs []*Blob, txConfig *SubmitOpt
 		}
 	}()
 
+	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 {
 		libBlobs[i] = blobs[i].Blob
@@ -213,39 +238,20 @@ func (s *Service) Submit(ctx context.Context, blobs []*Blob, txConfig *SubmitOpt
 	return uint64(resp.Height), nil
 }
 
-// SubmitFibreBlob submits a blob via the Fibre network.
-// It performs the full Fibre flow: upload to FSPs, aggregate validator signatures,
-// and submit MsgPayForFibre on-chain.
-func (s *Service) SubmitFibreBlob(
-	ctx context.Context,
-	ns libshare.Namespace,
-	data []byte,
-	txConfig *SubmitOptions,
-) (_ *fibre.SubmitResult, err error) {
-	ctx, span := tracer.Start(ctx, "blob/submit-fibre")
-	defer func() {
-		utils.SetStatusAndEnd(span, err)
-		if err != nil {
-			log.Errorw("submitting fibre blob failed", "err", err,
-				"namespace", ns.String(),
-			)
-		}
-	}()
-
+func (s *Service) submitFibreBlobs(ctx context.Context, blobs []*Blob, txConfig *SubmitOptions) (uint64, error) {
 	if s.fibreSubmitter == nil {
-		err = fmt.Errorf("fibre submitter is not configured: node is not connected to a core endpoint")
-		return nil, err
+		return 0, fmt.Errorf("fibre submitter is not available")
 	}
 
-	log.Infow("submitting fibre blob", "namespace", ns.String(), "data-size", len(data))
-	result, err := s.fibreSubmitter.Submit(ctx, ns, data, txConfig)
-	if err != nil {
-		err = fmt.Errorf("fibre submit: %w", err)
-		return nil, err
+	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
 	}
-
-	log.Debugw("fibre blob submitted", "namespace", ns.String(), "height", result.Height, "tx-hash", result.TxHash)
-	return result, nil
+	return height, nil
 }
 
 // Get retrieves a blob in a given namespace at the given height by commitment.