Merge branch 'main' into main

.github/workflows/integration-tests.yml

@@ -130,7 +130,7 @@ jobs:
         run: make test-integration SHORT=true TAGS=sync
 
   pruning_tests:
-    name: Integration Tests Sync
+    name: Integration Tests Pruning
     runs-on: ubuntu-latest
 
     steps:
@@ -143,3 +143,18 @@ jobs:
 
       - name: run sync tests
         run: make test-integration SHORT=true TAGS=pruning
+
+  share_tests:
+    name: Integration Tests Share
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: set up go
+        uses: actions/setup-go@v5
+        with:
+          go-version: ${{ inputs.go-version }}
+
+      - name: run share tests
+        run: make test-integration SHORT=true TAGS=share

blob/helper.go

@@ -3,6 +3,8 @@ package blob
 import (
 	"sort"
 
+	"github.com/celestiaorg/go-square/merkle"
+	"github.com/celestiaorg/go-square/v2/inclusion"
 	libshare "github.com/celestiaorg/go-square/v2/share"
 )
 
@@ -32,6 +34,20 @@ func ToLibBlobs(blobs ...*Blob) []*libshare.Blob {
 	return libBlobs
 }
 
+// ToNodeBlobs converts libshare blob type to the node's specific blob type.
+func ToNodeBlobs(blobs ...*libshare.Blob) ([]*Blob, error) {
+	nodeBlobs := make([]*Blob, len(blobs))
+	hashFromByteSlices := merkle.HashFromByteSlices
+	for i, blob := range blobs {
+		com, err := inclusion.CreateCommitment(blob, hashFromByteSlices, subtreeRootThreshold)
+		if err != nil {
+			return nil, err
+		}
+		nodeBlobs[i] = &Blob{Blob: blob, Commitment: com, index: -1}
+	}
+	return nodeBlobs, nil
+}
+
 func calculateIndex(rowLength, blobIndex int) (row, col int) {
 	row = blobIndex / rowLength
 	col = blobIndex - (row * rowLength)

core/eds.go

@@ -24,8 +24,8 @@ import (
 // extendBlock extends the given block data, returning the resulting
 // ExtendedDataSquare (EDS). If there are no transactions in the block,
 // nil is returned in place of the eds.
-func extendBlock(data types.Data, appVersion uint64, options ...nmt.Option) (*rsmt2d.ExtendedDataSquare, error) {
-	if app.IsEmptyBlockRef(&data, appVersion) {
+func extendBlock(data *types.Data, appVersion uint64, options ...nmt.Option) (*rsmt2d.ExtendedDataSquare, error) {
+	if app.IsEmptyBlockRef(data, appVersion) {
 		return share.EmptyEDS(), nil
 	}
 

core/eds_test.go

@@ -22,7 +22,7 @@ func TestTrulyEmptySquare(t *testing.T) {
 		SquareSize: 1,
 	}
 
-	eds, err := extendBlock(data, appconsts.LatestVersion)
+	eds, err := extendBlock(&data, appconsts.LatestVersion)
 	require.NoError(t, err)
 	require.True(t, eds.Equals(share.EmptyEDS()))
 }
@@ -38,7 +38,7 @@ func TestEmptySquareWithZeroTxs(t *testing.T) {
 		Txs: []types.Tx{},
 	}
 
-	eds, err := extendBlock(data, appconsts.LatestVersion)
+	eds, err := extendBlock(&data, appconsts.LatestVersion)
 	require.NoError(t, err)
 	require.True(t, eds.Equals(share.EmptyEDS()))
 

core/exchange.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/tendermint/tendermint/types"
 	"golang.org/x/sync/errgroup"
 
 	libhead "github.com/celestiaorg/go-header"
@@ -62,8 +61,7 @@ func NewExchange(
 
 func (ce *Exchange) GetByHeight(ctx context.Context, height uint64) (*header.ExtendedHeader, error) {
 	log.Debugw("requesting header", "height", height)
-	intHeight := int64(height)
-	return ce.getExtendedHeaderByHeight(ctx, &intHeight)
+	return ce.getExtendedHeaderByHeight(ctx, int64(height))
 }
 
 func (ce *Exchange) GetRangeByHeight(
@@ -129,12 +127,12 @@ func (ce *Exchange) Get(ctx context.Context, hash libhead.Hash) (*header.Extende
 		return nil, fmt.Errorf("fetching block by hash %s: %w", hash.String(), err)
 	}
 
-	comm, vals, err := ce.fetcher.GetBlockInfo(ctx, &block.Height)
+	comm, vals, err := ce.fetcher.GetBlockInfo(ctx, block.Height)
 	if err != nil {
 		return nil, fmt.Errorf("fetching block info for height %d: %w", &block.Height, err)
 	}
 
-	eds, err := extendBlock(block.Data, block.Header.Version.App)
+	eds, err := extendBlock(&block.Data, block.Header.Version.App)
 	if err != nil {
 		return nil, fmt.Errorf("extending block data for height %d: %w", &block.Height, err)
 	}
@@ -162,29 +160,22 @@ func (ce *Exchange) Head(
 	_ ...libhead.HeadOption[*header.ExtendedHeader],
 ) (*header.ExtendedHeader, error) {
 	log.Debug("requesting head")
-	return ce.getExtendedHeaderByHeight(ctx, nil)
+	return ce.getExtendedHeaderByHeight(ctx, 0)
 }
 
-func (ce *Exchange) getExtendedHeaderByHeight(ctx context.Context, height *int64) (*header.ExtendedHeader, error) {
+func (ce *Exchange) getExtendedHeaderByHeight(ctx context.Context, height int64) (*header.ExtendedHeader, error) {
 	b, err := ce.fetcher.GetSignedBlock(ctx, height)
 	if err != nil {
-		if height == nil {
-			return nil, fmt.Errorf("fetching signed block for head from core: %w", err)
-		}
-		return nil, fmt.Errorf("fetching signed block at height %d from core: %w", *height, err)
+		return nil, fmt.Errorf("fetching signed block at height %d from core: %w", height, err)
 	}
 	log.Debugw("fetched signed block from core", "height", b.Header.Height)
 
 	eds, err := extendBlock(b.Data, b.Header.Version.App)
 	if err != nil {
 		return nil, fmt.Errorf("extending block data for height %d: %w", b.Header.Height, err)
 	}
-
-	// TODO(@Wondertan): This is a hack to deref Data, allowing GC to pick it up.
-	//  The better footgun-less solution is to change core.ResultSignedBlock fields to be pointers instead of values.
-	b.Data = types.Data{}
-
-	eh, err := ce.construct(&b.Header, &b.Commit, &b.ValidatorSet, eds)
+	// create extended header
+	eh, err := ce.construct(b.Header, b.Commit, b.ValidatorSet, eds)
 	if err != nil {
 		panic(fmt.Errorf("constructing extended header for height %d: %w", b.Header.Height, err))
 	}

core/exchange_test.go

@@ -3,6 +3,7 @@ package core
 import (
 	"bytes"
 	"context"
+	"net"
 	"testing"
 	"time"
 
@@ -34,7 +35,7 @@ func TestCoreExchange_RequestHeaders(t *testing.T) {
 
 	// initialize store with genesis block
 	genHeight := int64(1)
-	genBlock, err := fetcher.GetBlock(ctx, &genHeight)
+	genBlock, err := fetcher.GetBlock(ctx, genHeight)
 	require.NoError(t, err)
 	genHeader, err := ce.Get(ctx, genBlock.Header.Hash().Bytes())
 	require.NoError(t, err)
@@ -61,6 +62,7 @@ func TestCoreExchange_RequestHeaders(t *testing.T) {
 		require.NoError(t, err)
 		assert.True(t, has)
 	}
+	require.NoError(t, fetcher.Stop(ctx))
 }
 
 // TestExchange_DoNotStoreHistoric tests that the CoreExchange will not
@@ -87,7 +89,7 @@ func TestExchange_DoNotStoreHistoric(t *testing.T) {
 
 	// initialize store with genesis block
 	genHeight := int64(1)
-	genBlock, err := fetcher.GetBlock(ctx, &genHeight)
+	genBlock, err := fetcher.GetBlock(ctx, genHeight)
 	require.NoError(t, err)
 	genHeader, err := ce.Get(ctx, genBlock.Header.Hash().Bytes())
 	require.NoError(t, err)
@@ -136,15 +138,15 @@ func TestExchange_StoreHistoricIfArchival(t *testing.T) {
 
 	// initialize store with genesis block
 	genHeight := int64(1)
-	genBlock, err := fetcher.GetBlock(ctx, &genHeight)
+	genBlock, err := fetcher.GetBlock(ctx, genHeight)
 	require.NoError(t, err)
 	genHeader, err := ce.Get(ctx, genBlock.Header.Hash().Bytes())
 	require.NoError(t, err)
 
 	headers, err := ce.GetRangeByHeight(ctx, genHeader, 30)
 	require.NoError(t, err)
 
-	// ensure all "historic" EDSs were stored
+	// ensure all "historic" EDSs were stored but not the .q4 files
 	for _, h := range headers {
 		has, err := store.HasByHeight(ctx, h.Height())
 		require.NoError(t, err)
@@ -157,6 +159,11 @@ func TestExchange_StoreHistoricIfArchival(t *testing.T) {
 		has, err = store.HasByHash(ctx, h.DAH.Hash())
 		require.NoError(t, err)
 		assert.True(t, has)
+
+		// ensure .q4 file was not stored
+		has, err = store.HasQ4ByHash(ctx, h.DAH.Hash())
+		require.NoError(t, err)
+		assert.False(t, has)
 	}
 }
 
@@ -166,7 +173,12 @@ func createCoreFetcher(t *testing.T, cfg *testnode.Config) (*BlockFetcher, testn
 	// flakiness with accessing account state)
 	_, err := cctx.WaitForHeightWithTimeout(2, time.Second*2) // TODO @renaynay: configure?
 	require.NoError(t, err)
-	return NewBlockFetcher(cctx.Client), cctx
+	host, port, err := net.SplitHostPort(cctx.GRPCClient.Target())
+	require.NoError(t, err)
+	client := newTestClient(t, host, port)
+	fetcher, err := NewBlockFetcher(client)
+	require.NoError(t, err)
+	return fetcher, cctx
 }
 
 // fillBlocks fills blocks until the context is canceled.
@@ -202,7 +214,7 @@ func generateNonEmptyBlocks(
 	sub, err := fetcher.SubscribeNewBlockEvent(ctx)
 	require.NoError(t, err)
 	defer func() {
-		err = fetcher.UnsubscribeNewBlockEvent(ctx)
+		err = fetcher.Stop(ctx)
 		require.NoError(t, err)
 	}()