x

Author: alpe

alex_questions.md

@@ -0,0 +1,9 @@
+# direct TX
+
+- What format can we expect for the direct TX on the DA? some wrapper type would be useful to unpack and distinct from
+  random bytes
+- Should we always include direct TX although they may be duplicates to mempool TX? spike: yes
+- Should we fill the block space with directTX if possible or reserve space for mempool TX
+
+## Smarter sequencer
+build blocks by max bytes from the request rather than returning what was added as a batch before from the syncer.

apps/evm/single/cmd/run.go

@@ -72,6 +72,7 @@ var RunCmd = &cobra.Command{
 			logger,
 			datastore,
 			100, // todo (Alex): what is a good value?
+			nodeConfig.ForcedInclusion,
 		)
 
 		nodeKey, err := key.LoadNodeKey(filepath.Dir(nodeConfig.ConfigPath()))

apps/evm/single/go.mod

@@ -54,7 +54,7 @@ require (
 	github.com/buger/goterm v1.0.4 // indirect
 	github.com/celestiaorg/go-header v0.6.6 // indirect
 	github.com/celestiaorg/go-libp2p-messenger v0.2.2 // indirect
-	github.com/celestiaorg/go-square/v2 v2.2.0 // indirect
+	github.com/celestiaorg/go-square/v2 v2.3.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/compose-spec/compose-go/v2 v2.6.0 // indirect

apps/evm/single/go.sum

@@ -107,8 +107,8 @@ github.com/celestiaorg/go-header v0.6.6 h1:17GvSXU/w8L1YWHZP4pYm9/4YHA8iy5Ku2wTE
 github.com/celestiaorg/go-header v0.6.6/go.mod h1:RdnlTmsyuNerztNiJiQE5G/EGEH+cErhQ83xNjuGcaQ=
 github.com/celestiaorg/go-libp2p-messenger v0.2.2 h1:osoUfqjss7vWTIZrrDSy953RjQz+ps/vBFE7bychLEc=
 github.com/celestiaorg/go-libp2p-messenger v0.2.2/go.mod h1:oTCRV5TfdO7V/k6nkx7QjQzGrWuJbupv+0o1cgnY2i4=
-github.com/celestiaorg/go-square/v2 v2.2.0 h1:zJnUxCYc65S8FgUfVpyG/osDcsnjzo/JSXw/Uwn8zp4=
-github.com/celestiaorg/go-square/v2 v2.2.0/go.mod h1:j8kQUqJLYtcvCQMQV6QjEhUdaF7rBTXF74g8LbkR0Co=
+github.com/celestiaorg/go-square/v2 v2.3.1 h1:CDdiQ+QkKPOQEcyDPODgP/PbAEzqUcftsohCPcbvsnw=
+github.com/celestiaorg/go-square/v2 v2.3.1/go.mod h1:6M2txj0j6dkoE+cgwyG0EqrEPhbZpM2R1lsWEopMIBc=
 github.com/celestiaorg/utils v0.1.0 h1:WsP3O8jF7jKRgLNFmlDCwdThwOFMFxg0MnqhkLFVxPo=
 github.com/celestiaorg/utils v0.1.0/go.mod h1:vQTh7MHnvpIeCQZ2/Ph+w7K1R2UerDheZbgJEJD2hSU=
 github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=

block/direct_tx.go

@@ -5,14 +5,8 @@ import (
 	"crypto/sha256"
 	"errors"
 	"github.com/evstack/ev-node/types"
-	"sync"
 )
 
-type ForcedInclusionConfig struct {
-	MaxInclusionDelay uint64 // Max inclusion time in DA block time units
-	MinDADelay        uint64 // Minimum number of DA blocks before including a direct tx
-}
-
 type DirectTransaction struct {
 	TxHash            types.Hash
 	FirstSeenDAHeight uint64 // DA block time when the tx was seen
@@ -21,15 +15,6 @@ type DirectTransaction struct {
 	TX                []byte
 }
 
-type DirectTxTracker struct {
-	config ForcedInclusionConfig
-	mu     sync.RWMutex
-	//txs                   map[string]DirectTransaction // hash -> tx
-	txs                   []DirectTransaction // ordered by da height and position in blob
-	latestSeenDABlockTime uint64
-	latestDAHeight        uint64
-}
-
 func (m *Manager) handlePotentialDirectTXs(ctx context.Context, bz []byte, daHeight uint64) bool {
 	var unsignedData types.Data // todo (Alex): we need some type to separate from noise
 	err := unsignedData.UnmarshalBinary(bz)
@@ -65,40 +50,9 @@ func (m *Manager) handlePotentialDirectTXs(ctx context.Context, bz []byte, daHei
 			Included:          false,
 			IncludedAt:        0,
 		}
-		m.directTXTracker.mu.Lock()
-		m.directTXTracker.txs = append(m.directTXTracker.txs, d)
-		m.directTXTracker.mu.Unlock()
+		_ = d
 	}
 	return true
 }
 
 var ErrMissingDirectTx = errors.New("missing direct tx")
-
-func (m *Manager) getPendingDirectTXs(_ context.Context, maxBytes int) ([][]byte, error) {
-	remaining := maxBytes
-	currBlockTime := m.directTXTracker.latestSeenDABlockTime
-	var res [][]byte
-
-	m.directTXTracker.mu.Lock()
-	defer m.directTXTracker.mu.Unlock()
-	for _, tx := range m.directTXTracker.txs {
-		if tx.Included {
-			continue
-		}
-
-		if currBlockTime-tx.FirstSeenDAHeight > m.directTXTracker.config.MaxInclusionDelay {
-			// should have been forced included already.
-			// what should we do now? stop the world
-			return nil, ErrMissingDirectTx
-		}
-		if m.directTXTracker.latestDAHeight-tx.FirstSeenDAHeight < m.directTXTracker.config.MinDADelay {
-			// we can stop here as following tx are newer
-			break
-		}
-		if len(tx.TX) > remaining {
-			break
-		}
-		res = append(res, tx.TX)
-	}
-	return res, nil
-}

block/direct_tx_reaper.go

@@ -44,12 +44,16 @@ func NewDirectTxReaper(
 	interval time.Duration,
 	logger log.EventLogger,
 	store ds.Batching,
+	daStartHeight uint64,
 ) *DirectTxReaper {
+	if daStartHeight == 0 {
+		daStartHeight = 1
+	}
 	if interval <= 0 {
 		interval = 100 * time.Millisecond
 	}
 	daHeight := new(atomic.Uint64)
-	daHeight.Store(1)
+	daHeight.Store(daStartHeight)
 	return &DirectTxReaper{
 		da:        da,
 		sequencer: sequencer,
@@ -80,7 +84,7 @@ func (r *DirectTxReaper) Start(ctx context.Context) {
 			return
 		case <-ticker.C:
 			daHeight := r.daHeight.Load()
-			if err := r.SubmitTxs(daHeight); err != nil {
+			if err := r.retrieveDirectTXs(daHeight); err != nil {
 				if strings.Contains(err.Error(), coreda.ErrHeightFromFuture.Error()) {
 					r.logger.Debug("IDs not found at height", "height", daHeight)
 				} else {
@@ -94,8 +98,8 @@ func (r *DirectTxReaper) Start(ctx context.Context) {
 	}
 }
 
-// SubmitTxs retrieves direct transactions from the DA layer and submits them to the sequencer.
-func (r *DirectTxReaper) SubmitTxs(daHeight uint64) error {
+// retrieveDirectTXs retrieves direct transactions from the DA layer and submits them to the sequencer.
+func (r *DirectTxReaper) retrieveDirectTXs(daHeight uint64) error {
 	// Get the latest DA height
 	// Get all blob IDs at the current DA height
 	result, err := r.da.GetIDs(r.ctx, daHeight, nil)
@@ -115,7 +119,7 @@ func (r *DirectTxReaper) SubmitTxs(daHeight uint64) error {
 	}
 	r.logger.Debug("Blobs found at height", "height", daHeight, "count", len(blobs))
 
-	var newTxs [][]byte
+	var newTxs []sequencer.DirectTX
 	for _, blob := range blobs {
 		r.logger.Debug("Processing blob data")
 
@@ -134,18 +138,21 @@ func (r *DirectTxReaper) SubmitTxs(daHeight uint64) error {
 		}
 
 		// Process each transaction in the blob
-		for _, tx := range data.Txs {
+		for i, tx := range data.Txs {
 			txHash := hashTx(tx)
 			has, err := r.seenStore.Has(r.ctx, ds.NewKey(txHash))
 			if err != nil {
 				return fmt.Errorf("check seenStore: %w", err)
 			}
 			if !has {
-				newTxs = append(newTxs, tx)
+				newTxs = append(newTxs, sequencer.DirectTX{
+					TX:              tx,
+					ID:              result.IDs[i],
+					FirstSeenHeight: daHeight,
+					FirstSeenTime:   result.Timestamp.Unix(),
+				})
 			}
 		}
-		// todo: apply checks"
-		// DA header time: result.Timestamp
 	}
 
 	if len(newTxs) == 0 {
@@ -154,13 +161,13 @@ func (r *DirectTxReaper) SubmitTxs(daHeight uint64) error {
 	}
 
 	r.logger.Debug("Submitting direct txs to sequencer", "txCount", len(newTxs))
-	err = r.sequencer.SubmitDirectTxs(r.ctx, newTxs)
+	err = r.sequencer.SubmitDirectTxs(r.ctx, newTxs...)
 	if err != nil {
 		return fmt.Errorf("submit direct txs to sequencer: %w", err)
 	}
 	// Mark the transactions as seen
-	for _, tx := range newTxs {
-		txHash := hashTx(tx)
+	for _, v := range newTxs {
+		txHash := hashTx(v.TX)
 		if err := r.seenStore.Put(r.ctx, ds.NewKey(txHash), []byte{1}); err != nil {
 			return fmt.Errorf("persist seen tx: %w", err)
 		}

block/direct_tx_reaper_test.go

@@ -153,7 +153,7 @@ func TestSubmitTxs(t *testing.T) {
 			reaper := createTestDirectTxReaper(t, mockDA, mockSequencer, chainID)
 			tt.setupMocks(mockDA, mockSequencer, reaper)
 
-			err := reaper.SubmitTxs(1)
+			err := reaper.retrieveDirectTXs(1)
 
 			if tt.expectedError {
 				assert.Error(t, err)
@@ -208,7 +208,7 @@ func createTestDirectTxReaper(
 		ctx,
 		mockDA,
 		mockSequencer,
-		nil, // Manager is not used in SubmitTxs
+		nil, // Manager is not used in retrieveDirectTXs
 		chainID,
 		time.Second,
 		logger,

block/manager.go

@@ -8,6 +8,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"github.com/celestiaorg/go-square/v2/share"
 	"path/filepath"
 	"sync"
 	"sync/atomic"
@@ -168,8 +169,6 @@ type Manager struct {
 	// validatorHasherProvider is used to provide the validator hash for the header.
 	// It is used to set the validator hash in the header.
 	validatorHasherProvider types.ValidatorHasherProvider
-
-	directTXTracker DirectTxTracker
 }
 
 // getInitialState tries to load lastState from Store, and if it's not available it reads genesis.
@@ -402,7 +401,6 @@ func NewManager(
 		txNotifyCh:               make(chan struct{}, 1), // Non-blocking channel
 		signaturePayloadProvider: managerOpts.SignaturePayloadProvider,
 		validatorHasherProvider:  managerOpts.ValidatorHasherProvider,
-		directTXTracker:          DirectTxTracker{txs: make([]DirectTransaction, 0)},
 	}
 
 	// initialize da included height
@@ -546,14 +544,21 @@ func (m *Manager) GetExecutor() coreexecutor.Executor {
 	return m.exec
 }
 
+const ( // copied from da/jsonclient/internal
+	defaultGovMaxSquareSize = 64
+	defaultMaxBytes         = defaultGovMaxSquareSize * defaultGovMaxSquareSize * share.ContinuationSparseShareContentSize
+)
+
 func (m *Manager) retrieveBatch(ctx context.Context) (*BatchData, error) {
 	m.logger.Debug("Attempting to retrieve next batch",
 		"chainID", m.genesis.ChainID,
 		"lastBatchData", m.lastBatchData)
 
 	req := coresequencer.GetNextBatchRequest{
-		Id:            []byte(m.genesis.ChainID),
-		LastBatchData: m.lastBatchData,
+		DAIncludedHeight: m.daIncludedHeight.Load(),
+		Id:               []byte(m.genesis.ChainID),
+		LastBatchData:    m.lastBatchData,
+		MaxBytes:         defaultMaxBytes, // todo (Alex): do we need to reserve some space for headers and other data?
 	}
 
 	res, err := m.sequencer.GetNextBatch(ctx, req)

block/reaper_test.go

@@ -33,7 +33,7 @@ func TestReaper_SubmitTxs_Success(t *testing.T) {
 	// Prepare transaction and its hash
 	tx := []byte("tx1")
 
-	// Mock interactions for the first SubmitTxs call
+	// Mock interactions for the first retrieveDirectTXs call
 	mockExec.On("GetTxs", mock.Anything).Return([][]byte{tx}, nil).Once()
 	submitReqMatcher := mock.MatchedBy(func(req coresequencer.SubmitBatchTxsRequest) bool {
 		return string(req.Id) == chainID && len(req.Batch.Transactions) == 1 && string(req.Batch.Transactions[0]) == string(tx)

core/sequencer/direct_tx.go

@@ -2,6 +2,10 @@ package sequencer
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/binary"
+	"fmt"
+	"hash"
 )
 
 // DirectTxSequencer is an interface for sequencers that can handle direct transactions.
@@ -11,5 +15,46 @@ type DirectTxSequencer interface {
 
 	// SubmitDirectTxs adds direct transactions to the sequencer.
 	// This method is called by the DirectTxReaper.
-	SubmitDirectTxs(ctx context.Context, txs [][]byte) error
+	SubmitDirectTxs(ctx context.Context, txs ...DirectTX) error
+}
+
+type DirectTX struct {
+	TX              []byte
+	ID              []byte
+	FirstSeenHeight uint64
+	// unix time
+	FirstSeenTime int64
+}
+
+// ValidateBasic performs basic validation of DirectTX fields
+func (d *DirectTX) ValidateBasic() error {
+	if len(d.TX) == 0 {
+		return fmt.Errorf("tx cannot be empty")
+	}
+	if len(d.ID) == 0 {
+		return fmt.Errorf("id cannot be empty")
+	}
+	if d.FirstSeenHeight == 0 {
+		return fmt.Errorf("first seen height cannot be zero")
+	}
+	if d.FirstSeenTime == 0 {
+		return fmt.Errorf("first seen time cannot be zero")
+	}
+	return nil
+}
+
+// Hash Hash on the data
+func (d *DirectTX) Hash() ([]byte, error) {
+	hasher := sha256.New()
+	hashWriteInt(hasher, len(d.ID))
+	hasher.Write(d.ID)
+	hashWriteInt(hasher, len(d.TX))
+	hasher.Write(d.TX)
+	return hasher.Sum(nil), nil
+}
+
+func hashWriteInt(hasher hash.Hash, data int) {
+	txLen := make([]byte, 8) // 8 bytes for uint64
+	binary.BigEndian.PutUint64(txLen, uint64(data))
+	hasher.Write(txLen)
 }