fork-choice interface

block/manager.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/rollkit/rollkit/config"
 	"github.com/rollkit/rollkit/da"
+	"github.com/rollkit/rollkit/fork_choice"
 	"github.com/rollkit/rollkit/log"
 	"github.com/rollkit/rollkit/mempool"
 	"github.com/rollkit/rollkit/state"
@@ -63,7 +64,7 @@ type Manager struct {
 	FraudProofInCh chan *abci.FraudProof
 
 	blockInCh chan newBlockEvent
-	syncCache map[uint64]*types.Block
+	syncCache map[uint64][]*types.Block
 
 	// retrieveMtx is used by retrieveCond
 	retrieveMtx *sync.Mutex
@@ -74,6 +75,7 @@ type Manager struct {
 
 	logger log.Logger
 
+	fcr fork_choice.ForkChoiceRule
 	// For usage by Lazy Aggregator mode
 	buildingBlock bool
 	txsAvailable <-chan struct{}
@@ -133,6 +135,13 @@ func NewManager(
  }
 	}
 
+	fcr, ok := fork_choice.GetRule(conf.ForkChoiceRule)
+	if !ok {
+ fcr, ok = fork_choice.GetRule(config.DefaultNodeConfig.BlockManagerConfig.ForkChoiceRule)
+ if !ok {
+ panic("Invalid default fork-choice rule")
+ }
+	}
 	var txsAvailableCh <-chan struct{}
 	if mempool != nil {
  txsAvailableCh = mempool.TxsAvailable()
@@ -155,9 +164,10 @@ func NewManager(
  blockInCh: make(chan newBlockEvent, 100),
  FraudProofInCh: make(chan *abci.FraudProof, 100),
  retrieveMtx: new(sync.Mutex),
+ syncCache: make(map[uint64][]*types.Block),
  lastStateMtx: new(sync.Mutex),
- syncCache: make(map[uint64]*types.Block),
  logger: logger,
+ fcr: fcr,
  txsAvailable: txsAvailableCh,
  doneBuildingBlock: doneBuildingCh,
  buildingBlock: false,
@@ -267,7 +277,10 @@ func (m *Manager) SyncLoop(ctx context.Context, cancel context.CancelFunc) {
  "daHeight", daHeight,
  "hash", block.Hash(),
  )
- m.syncCache[block.SignedHeader.Header.BaseHeader.Height] = block
+ if m.syncCache[block.SignedHeader.Header.BaseHeader.Height] == nil {
+ m.syncCache[block.SignedHeader.Header.BaseHeader.Height] = make([]*types.Block, 1)
+ }
+ m.syncCache[block.SignedHeader.Header.BaseHeader.Height] = append(m.syncCache[block.SignedHeader.Header.BaseHeader.Height], block)
  m.retrieveCond.Signal()
 
  err := m.trySyncNextBlock(ctx, daHeight)
@@ -313,8 +326,7 @@ func (m *Manager) SyncLoop(ctx context.Context, cancel context.CancelFunc) {
 func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
 	var commit *types.Commit
 	currentHeight := m.store.Height() // TODO(tzdybal): maybe store a copy in memory
-
-	b, ok := m.syncCache[currentHeight+1]
+	b, ok := m.fcr.Apply(m.syncCache[currentHeight+1])
 	if !ok {
  return nil
 	}
@@ -621,7 +633,7 @@ func (m *Manager) submitBlockToDA(ctx context.Context, block *types.Block) error
 	}
 
 	if !submitted {
- return fmt.Errorf("Failed to submit block to DA layer after %d attempts", maxSubmitAttempts)
+ return fmt.Errorf("failed to submit block to DA layer after %d attempts", maxSubmitAttempts)
 	}
 
 	return nil

config/config.go

@@ -21,6 +21,7 @@ const (
 	flagFraudProofs = "rollkit.experimental_insecure_fraud_proofs"
 	flagLight = "rollkit.light"
 	flagTrustedHash = "rollkit.trusted_hash"
+	flagForkChoiceRule = "rollkit.fork_choice_rule"
 	flagLazyAggregator = "rollkit.lazy_aggregator"
 )
 
@@ -53,16 +54,18 @@ type BlockManagerConfig struct {
 	// DABlockTime informs about block time of underlying data availability layer
 	DABlockTime time.Duration `mapstructure:"da_block_time"`
 	// DAStartHeight allows skipping first DAStartHeight-1 blocks when querying for blocks.
-	DAStartHeight uint64 `mapstructure:"da_start_height"`
-	NamespaceID types.NamespaceID `mapstructure:"namespace_id"`
-	FraudProofs bool `mapstructure:"fraud_proofs"`
+	DAStartHeight uint64 `mapstructure:"da_start_height"`
+	NamespaceID types.NamespaceID `mapstructure:"namespace_id"`
+	FraudProofs bool `mapstructure:"fraud_proofs"`
+	ForkChoiceRule string `mapstructure:"fork_choice_rule"`
 }
 
 // GetViperConfig reads configuration parameters from Viper instance.
 //
 // This method is called in cosmos-sdk.
 func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error {
 	nc.Aggregator = v.GetBool(flagAggregator)
+	nc.ForkChoiceRule = v.GetString(flagForkChoiceRule)
 	nc.DALayer = v.GetString(flagDALayer)
 	nc.DAConfig = v.GetString(flagDAConfig)
 	nc.DAStartHeight = v.GetUint64(flagDAStartHeight)
@@ -87,6 +90,7 @@ func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error {
 func AddFlags(cmd *cobra.Command) {
 	def := DefaultNodeConfig
 	cmd.Flags().Bool(flagAggregator, def.Aggregator, "run node in aggregator mode")
+	cmd.Flags().String(flagForkChoiceRule, def.ForkChoiceRule, "Fork-choice rule / leader selection scheme")
 	cmd.Flags().Bool(flagLazyAggregator, def.LazyAggregator, "wait for transactions, don't build empty blocks")
 	cmd.Flags().String(flagDALayer, def.DALayer, "Data Availability Layer Client name (mock or grpc")
 	cmd.Flags().String(flagDAConfig, def.DAConfig, "Data Availability Layer Client config")

config/defaults.go

@@ -3,6 +3,7 @@ package config
 import (
 	"time"
 
+	"github.com/rollkit/rollkit/fork_choice"
 	"github.com/rollkit/rollkit/types"
 )
 
@@ -21,9 +22,10 @@ var DefaultNodeConfig = NodeConfig{
 	Aggregator: false,
 	LazyAggregator: false,
 	BlockManagerConfig: BlockManagerConfig{
- BlockTime: 30 * time.Second,
- NamespaceID: types.NamespaceID{},
- FraudProofs: false,
+ ForkChoiceRule: fork_choice.FirstOrderedPure,
+ BlockTime: 30 * time.Second,
+ NamespaceID: types.NamespaceID{},
+ FraudProofs: false,
 	},
 	DALayer: "mock",
 	DAConfig: "",

da/celestia/mock/server.go

@@ -70,7 +70,7 @@ func (s *Server) Stop() {
 
 func (s *Server) getHandler() http.Handler {
 	mux := mux2.NewRouter()
-	mux.HandleFunc("/submit_pfb", s.submit).Methods(http.MethodPost)
+	mux.HandleFunc("/submit_pfd", s.submit).Methods(http.MethodPost)
 	mux.HandleFunc("/namespaced_shares/{namespace}/height/{height}", s.shares).Methods(http.MethodGet)
 	mux.HandleFunc("/namespaced_data/{namespace}/height/{height}", s.data).Methods(http.MethodGet)
 

fork_choice/first_ordered_pure.go

@@ -0,0 +1,17 @@
+package fork_choice
+
+import "github.com/rollkit/rollkit/types"
+
+var _ ForkChoiceRule = firstOrderedPure{}
+
+type firstOrderedPure struct{}
+
+func (f firstOrderedPure) Apply(candidates []*types.Block) (*types.Block, bool) {
+	// Return first non-nil block from the candidates
+	for _, b := range candidates {
+ if b != nil {
+ return b, true
+ }
+	}
+	return nil, false
+}

fork_choice/rule.go

@@ -0,0 +1,25 @@
+package fork_choice
+
+import (
+	"github.com/rollkit/rollkit/types"
+)
+
+const (
+	//FirstOrderedCentralized = "FOC"
+	FirstOrderedPure = "FOP"
+	//HighestGasPure = "HGP"
+	//Astria = "ASTRIA"
+)
+
+func GetRule(ruleName string) (ForkChoiceRule, bool) {
+	switch ruleName {
+	case FirstOrderedPure:
+ return firstOrderedPure{}, true
+	default:
+ return nil, false
+	}
+}
+
+type ForkChoiceRule interface {
+	Apply(candidates []*types.Block) (*types.Block, bool)
+}

node/full_client_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/rollkit/rollkit/conv"
 	abciconv "github.com/rollkit/rollkit/conv/abci"
 	mockda "github.com/rollkit/rollkit/da/mock"
+	"github.com/rollkit/rollkit/fork_choice"
 	"github.com/rollkit/rollkit/mocks"
 	"github.com/rollkit/rollkit/store"
 	"github.com/rollkit/rollkit/types"
@@ -433,7 +434,8 @@ func TestTx(t *testing.T) {
  DALayer: "mock",
  Aggregator: true,
  BlockManagerConfig: config.BlockManagerConfig{
- BlockTime: 1 * time.Second, // blocks must be at least 1 sec apart for adjacent headers to get verified correctly
+ ForkChoiceRule: fork_choice.FirstOrderedPure,
+ BlockTime: 1 * time.Second, // blocks must be at least 1 sec apart for adjacent headers to get verified correctly
  }},
  key, signingKey, abcicli.NewLocalClient(nil, mockApp),
  &tmtypes.GenesisDoc{ChainID: "test", Validators: genesisValidators},

node/full_node_integration_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/rollkit/rollkit/config"
 	"github.com/rollkit/rollkit/da"
 	mockda "github.com/rollkit/rollkit/da/mock"
+	"github.com/rollkit/rollkit/fork_choice"
 	"github.com/rollkit/rollkit/mocks"
 	"github.com/rollkit/rollkit/p2p"
 	"github.com/rollkit/rollkit/store"
@@ -48,8 +49,9 @@ func TestAggregatorMode(t *testing.T) {
 	anotherKey, _, _ := crypto.GenerateEd25519Key(rand.Reader)
 	genesisValidators, signingKey := getGenesisValidatorSetWithSigner(1)
 	blockManagerConfig := config.BlockManagerConfig{
- BlockTime: 1 * time.Second,
- NamespaceID: types.NamespaceID{1, 2, 3, 4, 5, 6, 7, 8},
+ ForkChoiceRule: fork_choice.FirstOrderedPure,
+ BlockTime: 1 * time.Second,
+ NamespaceID: types.NamespaceID{1, 2, 3, 4, 5, 6, 7, 8},
 	}
 	node, err := newFullNode(context.Background(), config.NodeConfig{DALayer: "mock", Aggregator: true, BlockManagerConfig: blockManagerConfig}, key, signingKey, abcicli.NewLocalClient(nil, app), &tmtypes.GenesisDoc{ChainID: "test", Validators: genesisValidators}, log.TestingLogger())
 	require.NoError(err)
@@ -494,10 +496,11 @@ func createNode(ctx context.Context, n int, isMalicious bool, aggregator bool, d
  ListenAddress: "/ip4/127.0.0.1/tcp/" + strconv.Itoa(startPort+n),
 	}
 	bmConfig := config.BlockManagerConfig{
- DABlockTime: 100 * time.Millisecond,
- BlockTime: 1 * time.Second, // blocks must be at least 1 sec apart for adjacent headers to get verified correctly
- NamespaceID: types.NamespaceID{8, 7, 6, 5, 4, 3, 2, 1},
- FraudProofs: true,
+ ForkChoiceRule: fork_choice.FirstOrderedPure,
+ BlockTime: 1 * time.Second, // blocks must be at least 1 sec apart for adjacent headers to get verified correctly
+ NamespaceID: types.NamespaceID{8, 7, 6, 5, 4, 3, 2, 1},
+ FraudProofs: true,
+ DABlockTime: 100 * time.Millisecond,
 	}
 	for i := 0; i < len(keys); i++ {
  if i == n {

node/header_exchange.go

@@ -10,7 +10,6 @@ import (
 	goheaderp2p "github.com/celestiaorg/go-header/p2p"
 	goheaderstore "github.com/celestiaorg/go-header/store"
 	"github.com/celestiaorg/go-header/sync"
-	goheadersync "github.com/celestiaorg/go-header/sync"
 	ds "github.com/ipfs/go-datastore"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/host"
@@ -145,7 +144,7 @@ func (hExService *HeaderExchangeService) Start() error {
  return fmt.Errorf("error while starting exchange: %w", err)
 	}
 
-	if hExService.syncer, err = newSyncer(hExService.ex, hExService.headerStore, hExService.sub, goheadersync.WithBlockTime(hExService.conf.BlockTime)); err != nil {
+	if hExService.syncer, err = newSyncer(hExService.ex, hExService.headerStore, hExService.sub, sync.WithBlockTime(hExService.conf.BlockTime)); err != nil {
  return err
 	}
 
@@ -229,7 +228,7 @@ func newSyncer(
 	ex header.Exchange[*types.SignedHeader],
 	store header.Store[*types.SignedHeader],
 	sub header.Subscriber[*types.SignedHeader],
-	opt goheadersync.Options,
+	opt sync.Options,
 ) (*sync.Syncer[*types.SignedHeader], error) {
 	return sync.NewSyncer[*types.SignedHeader](ex, store, sub, opt)
 }