Merge branch 'main' into alex/deterministic_queue

Author: julienrbrt

.mockery.yaml

@@ -63,17 +63,20 @@ packages:
           dir: ./test/mocks
           pkgname: mocks
           filename: da.go
-  github.com/evstack/ev-node/pkg/da/types:
-    interfaces:
       Verifier:
         config:
           dir: ./test/mocks
           pkgname: mocks
-          filename: da_verifier.go
+          filename: da.go
   github.com/evstack/ev-node/pkg/da/jsonrpc:
     interfaces:
       BlobModule:
         config:
           dir: ./pkg/da/jsonrpc/mocks
           pkgname: mocks
           filename: blob_module_mock.go
+      HeaderModule:
+        config:
+          dir: ./pkg/da/jsonrpc/mocks
+          pkgname: mocks
+          filename: header_module_mock.go

apps/evm/cmd/rollback.go

@@ -5,12 +5,9 @@ import (
 	"errors"
 	"fmt"
 
-	ds "github.com/ipfs/go-datastore"
-	kt "github.com/ipfs/go-datastore/keytransform"
 	"github.com/spf13/cobra"
 
 	goheaderstore "github.com/celestiaorg/go-header/store"
-	"github.com/evstack/ev-node/node"
 	rollcmd "github.com/evstack/ev-node/pkg/cmd"
 	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/types"
@@ -50,10 +47,7 @@ func NewRollbackCmd() *cobra.Command {
 			}()
 
 			// prefixed evolve db
-			evolveDB := kt.Wrap(rawEvolveDB, &kt.PrefixTransform{
-				Prefix: ds.NewKey(node.EvPrefix),
-			})
-
+			evolveDB := store.NewEvNodeKVStore(rawEvolveDB)
 			evolveStore := store.New(evolveDB)
 			if height == 0 {
 				currentHeight, err := evolveStore.Height(goCtx)

apps/evm/cmd/run.go

@@ -88,7 +88,7 @@ var RunCmd = &cobra.Command{
 		}
 
 		// Create sequencer based on configuration
-		sequencer, err := createSequencer(context.Background(), logger, datastore, nodeConfig, genesis, daClient)
+		sequencer, err := createSequencer(logger, datastore, nodeConfig, genesis, daClient)
 		if err != nil {
 			return err
 		}
@@ -154,22 +154,20 @@ func init() {
 // If BasedSequencer is enabled, it creates a based sequencer that fetches transactions from DA.
 // Otherwise, it creates a single (traditional) sequencer.
 func createSequencer(
-	ctx context.Context,
 	logger zerolog.Logger,
 	datastore datastore.Batching,
 	nodeConfig config.Config,
 	genesis genesis.Genesis,
 	daClient block.FullDAClient,
 ) (coresequencer.Sequencer, error) {
-	fiRetriever := block.NewForcedInclusionRetriever(daClient, genesis, logger)
-
 	if nodeConfig.Node.BasedSequencer {
 		// Based sequencer mode - fetch transactions only from DA
 		if !nodeConfig.Node.Aggregator {
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
+		fiRetriever := block.NewForcedInclusionRetriever(daClient, logger, genesis.DAStartHeight, genesis.DAEpochForcedInclusion)
+		basedSeq, err := based.NewBasedSequencer(fiRetriever, datastore, genesis, logger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}
@@ -183,15 +181,12 @@ func createSequencer(
 	}
 
 	sequencer, err := single.NewSequencer(
-		ctx,
 		logger,
 		datastore,
 		daClient,
 		[]byte(genesis.ChainID),
 		nodeConfig.Node.BlockTime.Duration,
-		nodeConfig.Node.Aggregator,
 		1000,
-		fiRetriever,
 		genesis,
 	)
 	if err != nil {

apps/grpc/cmd/run.go

@@ -120,15 +120,15 @@ func createSequencer(
 	}
 
 	daClient := block.NewDAClient(blobClient, nodeConfig, logger)
-	fiRetriever := block.NewForcedInclusionRetriever(daClient, genesis, logger)
 
 	if nodeConfig.Node.BasedSequencer {
 		// Based sequencer mode - fetch transactions only from DA
 		if !nodeConfig.Node.Aggregator {
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
+		fiRetriever := block.NewForcedInclusionRetriever(daClient, logger, genesis.DAStartHeight, genesis.DAEpochForcedInclusion)
+		basedSeq, err := based.NewBasedSequencer(fiRetriever, datastore, genesis, logger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}
@@ -142,15 +142,12 @@ func createSequencer(
 	}
 
 	sequencer, err := single.NewSequencer(
-		ctx,
 		logger,
 		datastore,
 		daClient,
 		[]byte(genesis.ChainID),
 		nodeConfig.Node.BlockTime.Duration,
-		nodeConfig.Node.Aggregator,
 		1000,
-		fiRetriever,
 		genesis,
 	)
 	if err != nil {

apps/testapp/cmd/rollback.go

@@ -6,14 +6,11 @@ import (
 	"fmt"
 
 	kvexecutor "github.com/evstack/ev-node/apps/testapp/kv"
-	"github.com/evstack/ev-node/node"
 	rollcmd "github.com/evstack/ev-node/pkg/cmd"
 	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/types"
 
 	goheaderstore "github.com/celestiaorg/go-header/store"
-	ds "github.com/ipfs/go-datastore"
-	kt "github.com/ipfs/go-datastore/keytransform"
 	"github.com/spf13/cobra"
 )
 
@@ -51,10 +48,7 @@ func NewRollbackCmd() *cobra.Command {
 			}()
 
 			// prefixed evolve db
-			evolveDB := kt.Wrap(rawEvolveDB, &kt.PrefixTransform{
-				Prefix: ds.NewKey(node.EvPrefix),
-			})
-
+			evolveDB := store.NewEvNodeKVStore(rawEvolveDB)
 			evolveStore := store.New(evolveDB)
 			if height == 0 {
 				currentHeight, err := evolveStore.Height(goCtx)

apps/testapp/cmd/run.go

@@ -121,15 +121,15 @@ func createSequencer(
 	}
 
 	daClient := block.NewDAClient(blobClient, nodeConfig, logger)
-	fiRetriever := block.NewForcedInclusionRetriever(daClient, genesis, logger)
 
 	if nodeConfig.Node.BasedSequencer {
 		// Based sequencer mode - fetch transactions only from DA
 		if !nodeConfig.Node.Aggregator {
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
+		fiRetriever := block.NewForcedInclusionRetriever(daClient, logger, genesis.DAStartHeight, genesis.DAEpochForcedInclusion)
+		basedSeq, err := based.NewBasedSequencer(fiRetriever, datastore, genesis, logger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}
@@ -143,15 +143,12 @@ func createSequencer(
 	}
 
 	sequencer, err := single.NewSequencer(
-		ctx,
 		logger,
 		datastore,
 		daClient,
 		[]byte(genesis.ChainID),
 		nodeConfig.Node.BlockTime.Duration,
-		nodeConfig.Node.Aggregator,
 		1000,
-		fiRetriever,
 		genesis,
 	)
 	if err != nil {

block/components.go

@@ -161,6 +161,7 @@ func NewSyncComponents(
 		config,
 		genesis,
 		daSubmitter,
+		nil, // No sequencer for sync nodes
 		nil, // No signer for sync nodes
 		logger,
 		errorCh,
@@ -250,6 +251,7 @@ func NewAggregatorComponents(
 		config,
 		genesis,
 		daSubmitter,
+		sequencer,
 		signer, // Signer for aggregator nodes to submit to DA
 		logger,
 		errorCh,

block/internal/cache/pending_headers.go

@@ -17,7 +17,7 @@ import (
 // - DA submission of multiple headers is atomic - it's impossible to submit only part of a batch
 //
 // lastSubmittedHeaderHeight is updated only after receiving confirmation from DA.
-// Worst case scenario is when headers was successfully submitted to DA, but confirmation was not received (e.g. node was
+// Worst case scenario is when headers were successfully submitted to DA, but confirmation was not received (e.g. node was
 // restarted, networking issue occurred). In this case headers are re-submitted to DA (it's extra cost).
 // evolve is able to skip duplicate headers so this shouldn't affect full nodes.
 // TODO(tzdybal): we shouldn't try to push all pending headers at once; this should depend on max blob size

block/internal/da/client.go

@@ -30,6 +30,7 @@ type Config struct {
 // It is unexported; callers should use the exported Client interface.
 type client struct {
 	blobAPI            *blobrpc.BlobAPI
+	headerAPI          *blobrpc.HeaderAPI
 	logger             zerolog.Logger
 	defaultTimeout     time.Duration
 	namespaceBz        []byte
@@ -58,6 +59,7 @@ func NewClient(cfg Config) FullClient {
 
 	return &client{
 		blobAPI:            &cfg.DA.Blob,
+		headerAPI:          &cfg.DA.Header,
 		logger:             cfg.Logger.With().Str("component", "da_client").Logger(),
 		defaultTimeout:     cfg.DefaultTimeout,
 		namespaceBz:        datypes.NamespaceFromString(cfg.Namespace).Bytes(),
@@ -180,8 +182,22 @@ func (c *client) Submit(ctx context.Context, data [][]byte, _ float64, namespace
 	}
 }
 
+// getBlockTimestamp fetches the block timestamp from the DA layer header.
+func (c *client) getBlockTimestamp(ctx context.Context, height uint64) (time.Time, error) {
+	headerCtx, cancel := context.WithTimeout(ctx, c.defaultTimeout)
+	defer cancel()
+
+	header, err := c.headerAPI.GetByHeight(headerCtx, height)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("failed to get header timestamp for block %d: %w", height, err)
+	}
+
+	return header.Time(), nil
+}
+
 // Retrieve retrieves blobs from the DA layer at the specified height and namespace.
 // It uses GetAll to fetch all blobs at once.
+// The timestamp is derived from the DA block header to ensure determinism.
 func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte) datypes.ResultRetrieve {
 	ns, err := share.NewNamespaceFromBytes(namespace)
 	if err != nil {
@@ -195,21 +211,29 @@ func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte)
 		}
 	}
 
-	getCtx, cancel := context.WithTimeout(ctx, c.defaultTimeout)
-	defer cancel()
+	blobCtx, blobCancel := context.WithTimeout(ctx, c.defaultTimeout)
+	defer blobCancel()
 
-	blobs, err := c.blobAPI.GetAll(getCtx, height, []share.Namespace{ns})
+	blobs, err := c.blobAPI.GetAll(blobCtx, height, []share.Namespace{ns})
 	if err != nil {
 		// Handle known errors by substring because RPC may wrap them.
 		switch {
 		case strings.Contains(err.Error(), datypes.ErrBlobNotFound.Error()):
 			c.logger.Debug().Uint64("height", height).Msg("No blobs found at height")
+			// Fetch block timestamp for deterministic responses using parent context
+			blockTime, err := c.getBlockTimestamp(ctx, height)
+			if err != nil {
+				c.logger.Error().Uint64("height", height).Err(err).Msg("failed to get block timestamp")
+				blockTime = time.Now()
+				// TODO: we should retry fetching the timestamp. Current time may mess block time consistency for based sequencers.
+			}
+
 			return datypes.ResultRetrieve{
 				BaseResult: datypes.BaseResult{
 					Code:      datypes.StatusNotFound,
 					Message:   datypes.ErrBlobNotFound.Error(),
 					Height:    height,
-					Timestamp: time.Now(),
+					Timestamp: blockTime,
 				},
 			}
 		case strings.Contains(err.Error(), datypes.ErrHeightFromFuture.Error()):
@@ -234,14 +258,22 @@ func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte)
 		}
 	}
 
+	// Fetch block timestamp for deterministic responses using parent context
+	blockTime, err := c.getBlockTimestamp(ctx, height)
+	if err != nil {
+		c.logger.Error().Uint64("height", height).Err(err).Msg("failed to get block timestamp")
+		blockTime = time.Now()
+		// TODO: we should retry fetching the timestamp. Current time may mess block time consistency for based sequencers.
+	}
+
 	if len(blobs) == 0 {
 		c.logger.Debug().Uint64("height", height).Msg("No blobs found at height")
 		return datypes.ResultRetrieve{
 			BaseResult: datypes.BaseResult{
 				Code:      datypes.StatusNotFound,
 				Message:   datypes.ErrBlobNotFound.Error(),
 				Height:    height,
-				Timestamp: time.Now(),
+				Timestamp: blockTime,
 			},
 		}
 	}
@@ -261,7 +293,7 @@ func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte)
 			Code:      datypes.StatusSuccess,
 			Height:    height,
 			IDs:       ids,
-			Timestamp: time.Now(),
+			Timestamp: blockTime,
 		},
 		Data: data,
 	}