Direct Sequencer & Progressive Block Building

block/manager.go

@@ -72,6 +72,12 @@ type Manager struct {
 	// retrieveCond is used to notify sync goroutine (SyncLoop) that it needs to retrieve data
 	retrieveCond *sync.Cond
 
+	// whether or not the centralized sequencer is currently building a block
+	buildingBlock     bool
+	txsAvailable      <-chan struct{}
+	moreTxsAvailable  chan struct{}
+	doneBuildingBlock chan struct{}
+
 	logger log.Logger
 }
 
@@ -95,6 +101,7 @@ func NewManager(
 	dalc da.DataAvailabilityLayerClient,
 	eventBus *tmtypes.EventBus,
 	logger log.Logger,
+	doneBuildingCh chan struct{},
 ) (*Manager, error) {
 	s, err := getInitialState(store, genesis)
 	if err != nil {
@@ -127,6 +134,13 @@ func NewManager(
 		}
 	}
 
+	var txsAvailableChan <-chan struct{}
+	if mempool != nil {
+		txsAvailableChan = mempool.TxsAvailable()
+	} else {
+		txsAvailableChan = nil
+	}
+
 	agg := &Manager{
 		proposerKey: proposerKey,
 		conf:        conf,
@@ -138,13 +152,17 @@ func NewManager(
 		retriever:   dalc.(da.BlockRetriever), // TODO(tzdybal): do it in more gentle way (after MVP)
 		daHeight:    s.DAHeight,
 		// channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
-		HeaderOutCh:    make(chan *types.SignedHeader, 100),
-		HeaderInCh:     make(chan *types.SignedHeader, 100),
-		blockInCh:      make(chan newBlockEvent, 100),
-		FraudProofInCh: make(chan *abci.FraudProof, 100),
-		retrieveMtx:    new(sync.Mutex),
-		syncCache:      make(map[uint64]*types.Block),
-		logger:         logger,
+		HeaderOutCh:       make(chan *types.SignedHeader, 100),
+		HeaderInCh:        make(chan *types.SignedHeader, 100),
+		blockInCh:         make(chan newBlockEvent, 100),
+		FraudProofInCh:    make(chan *abci.FraudProof, 100),
+		retrieveMtx:       new(sync.Mutex),
+		syncCache:         make(map[uint64]*types.Block),
+		buildingBlock:     false,
+		txsAvailable:      txsAvailableChan,
+		moreTxsAvailable:  make(chan struct{}),
+		doneBuildingBlock: doneBuildingCh,
+		logger:            logger,
 	}
 	agg.retrieveCond = sync.NewCond(agg.retrieveMtx)
 
@@ -170,36 +188,67 @@ func (m *Manager) GetFraudProofOutChan() chan *abci.FraudProof {
 }
 
 // AggregationLoop is responsible for aggregating transactions into rollup-blocks.
-func (m *Manager) AggregationLoop(ctx context.Context) {
-	initialHeight := uint64(m.genesis.InitialHeight)
-	height := m.store.Height()
-	var delay time.Duration
-
-	// TODO(tzdybal): double-check when https://github.com/celestiaorg/rollmint/issues/699 is resolved
-	if height < initialHeight {
-		delay = time.Until(m.genesis.GenesisTime)
-	} else {
-		delay = time.Until(m.lastState.LastBlockTime.Add(m.conf.BlockTime))
-	}
+func (m *Manager) AggregationLoop(ctx context.Context, ingress chan []byte, lazy bool) {
+	var timer time.Timer
+	if !lazy {
+		initialHeight := uint64(m.genesis.InitialHeight)
+		height := m.store.Height()
+		var delay time.Duration
+
+		// TODO(tzdybal): double-check when https://github.com/celestiaorg/rollmint/issues/699 is resolved
+		if height < initialHeight {
+			delay = time.Until(m.genesis.GenesisTime)
+		} else {
+			delay = time.Until(m.lastState.LastBlockTime.Add(m.conf.BlockTime))
+		}
 
-	if delay > 0 {
-		m.logger.Info("Waiting to produce block", "delay", delay)
-		time.Sleep(delay)
-	}
+		if delay > 0 {
+			m.logger.Info("Waiting to produce block", "delay", delay)
+			time.Sleep(delay)
+		}
 
-	timer := time.NewTimer(0)
+		timer = *time.NewTimer(0)
 
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-timer.C:
-			start := time.Now()
-			err := m.publishBlock(ctx)
-			if err != nil {
-				m.logger.Error("error while publishing block", "error", err)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-timer.C:
+				start := time.Now()
+				err := m.publishBlock(ctx, false, m.moreTxsAvailable)
+				if err != nil {
+					m.logger.Error("error while publishing block", "error", err)
+				}
+				timer.Reset(m.getRemainingSleep(start))
+			}
+		}
+	} else {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-m.txsAvailable:
+				m.logger.Debug("Txs available! Starting block building...")
+				if !m.buildingBlock {
+					m.buildingBlock = true
+					timer = *time.NewTimer(1 * time.Second)
+				}
+			case <-m.moreTxsAvailable:
+				m.logger.Debug("Txs remained in mempool after building last block. Building another block.")
+				if !m.buildingBlock {
+					m.buildingBlock = true
+					timer = *time.NewTimer(1 * time.Second)
+				}
+			case <-timer.C:
+				m.logger.Debug("Block building time elapsed.")
+				m.buildingBlock = false
+				err := m.publishBlock(ctx, true, m.moreTxsAvailable)
+				if err != nil {
+					m.logger.Error("error while publishing block", "error", err)
+				}
+				close(m.doneBuildingBlock)
+				m.doneBuildingBlock = make(chan struct{})
 			}
-			timer.Reset(m.getRemainingSleep(start))
 		}
 	}
 }
@@ -316,7 +365,7 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
 		if err != nil {
 			return fmt.Errorf("failed to save block: %w", err)
 		}
-		_, _, err = m.executor.Commit(ctx, newState, b1, responses)
+		_, _, err = m.executor.Commit(ctx, newState, b1, responses, nil)
 		if err != nil {
 			return fmt.Errorf("failed to Commit: %w", err)
 		}
@@ -446,7 +495,7 @@ func (m *Manager) getCommit(header types.Header) (*types.Commit, error) {
 	}, nil
 }
 
-func (m *Manager) publishBlock(ctx context.Context) error {
+func (m *Manager) publishBlock(ctx context.Context, maxTx bool, moreTxsAvailable chan struct{}) error {
 	var lastCommit *types.Commit
 	var lastHeaderHash types.Hash
 	var err error
@@ -521,7 +570,7 @@ func (m *Manager) publishBlock(ctx context.Context) error {
 	}
 
 	// Commit the new state and block which writes to disk on the proxy app
-	_, _, err = m.executor.Commit(ctx, newState, block, responses)
+	_, _, err = m.executor.Commit(ctx, newState, block, responses, moreTxsAvailable)
 	if err != nil {
 		return err
 	}

block/manager_test.go

@@ -80,7 +80,8 @@ func TestInitialState(t *testing.T) {
 			assert := assert.New(t)
 			logger := log.TestingLogger()
 			dalc := getMockDALC(logger)
-			agg, err := NewManager(key, conf, c.genesis, c.store, nil, nil, dalc, nil, logger)
+			dumbChan := make(chan struct{})
+			agg, err := NewManager(key, conf, c.genesis, c.store, nil, nil, dalc, nil, logger, dumbChan)
 			assert.NoError(err)
 			assert.NotNil(agg)
 			assert.Equal(c.expectedChainID, agg.lastState.ChainID)

config/config.go

@@ -20,6 +20,8 @@ const (
 	flagNamespaceID   = "rollkit.namespace_id"
 	flagFraudProofs   = "rollkit.experimental_insecure_fraud_proofs"
 	flagLight         = "rollkit.light"
+
+	flagLazyAggregator = "rollkit.lazy_aggregator"
 )
 
 // NodeConfig stores Rollkit node configuration.
@@ -30,11 +32,14 @@ type NodeConfig struct {
 	P2P     P2PConfig
 	RPC     RPCConfig
 	// parameters below are Rollkit specific and read from config
-	Aggregator         bool `mapstructure:"aggregator"`
 	BlockManagerConfig `mapstructure:",squash"`
 	DALayer            string `mapstructure:"da_layer"`
 	DAConfig           string `mapstructure:"da_config"`
 	Light              bool   `mapstructure:"light"`
+
+	// Sequencer-specific settings
+	Aggregator     bool `mapstructure:"aggregator"`
+	LazyAggregator bool `mapstructure:"lazy_aggregator"`
 }
 
 // BlockManagerConfig consists of all parameters required by BlockManagerConfig
@@ -62,6 +67,7 @@ func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error {
 	nsID := v.GetString(flagNamespaceID)
 	nc.FraudProofs = v.GetBool(flagFraudProofs)
 	nc.Light = v.GetBool(flagLight)
+	nc.LazyAggregator = v.GetBool(flagLazyAggregator)
 	bytes, err := hex.DecodeString(nsID)
 	if err != nil {
 		return err
@@ -84,4 +90,5 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().BytesHex(flagNamespaceID, def.NamespaceID[:], "namespace identifies (8 bytes in hex)")
 	cmd.Flags().Bool(flagFraudProofs, def.FraudProofs, "enable fraud proofs (experimental & insecure)")
 	cmd.Flags().Bool(flagLight, def.Light, "run light client")
+	cmd.Flags().Bool(flagLazyAggregator, def.LazyAggregator, "progressive sequencer")
 }

config/defaults.go

@@ -24,7 +24,8 @@ var DefaultNodeConfig = NodeConfig{
 		NamespaceID: types.NamespaceID{},
 		FraudProofs: false,
 	},
-	DALayer:  "mock",
-	DAConfig: "",
-	Light:    false,
+	DALayer:        "mock",
+	DAConfig:       "",
+	Light:          false,
+	LazyAggregator: false,
 }

node/full.go

@@ -65,9 +65,10 @@ type FullNode struct {
 	P2P  *p2p.Client
 
 	// TODO(tzdybal): consider extracting "mempool reactor"
-	Mempool      mempool.Mempool
-	mempoolIDs   *mempoolIDs
-	incomingTxCh chan *p2p.GossipMessage
+	Mempool           mempool.Mempool
+	mempoolIDs        *mempoolIDs
+	incomingTxCh      chan []byte
+	DoneBuildingBlock chan struct{}
 
 	Store        store.Store
 	blockManager *block.Manager
@@ -138,31 +139,35 @@ func newFullNode(
 
 	mp := mempoolv1.NewTxMempool(logger, llcfg.DefaultMempoolConfig(), appClient, 0)
 	mpIDs := newMempoolIDs()
+	mp.EnableTxsAvailable()
 
-	blockManager, err := block.NewManager(signingKey, conf.BlockManagerConfig, genesis, s, mp, appClient, dalc, eventBus, logger.With("module", "BlockManager"))
+	doneBuildingChannel := make(chan struct{})
+
+	blockManager, err := block.NewManager(signingKey, conf.BlockManagerConfig, genesis, s, mp, appClient, dalc, eventBus, logger.With("module", "BlockManager"), doneBuildingChannel)
 	if err != nil {
 		return nil, fmt.Errorf("BlockManager initialization error: %w", err)
 	}
 
 	ctx, cancel := context.WithCancel(ctx)
 
 	node := &FullNode{
-		appClient:      appClient,
-		eventBus:       eventBus,
-		genesis:        genesis,
-		conf:           conf,
-		P2P:            client,
-		blockManager:   blockManager,
-		dalc:           dalc,
-		Mempool:        mp,
-		mempoolIDs:     mpIDs,
-		incomingTxCh:   make(chan *p2p.GossipMessage),
-		Store:          s,
-		TxIndexer:      txIndexer,
-		IndexerService: indexerService,
-		BlockIndexer:   blockIndexer,
-		ctx:            ctx,
-		cancel:         cancel,
+		appClient:         appClient,
+		eventBus:          eventBus,
+		genesis:           genesis,
+		conf:              conf,
+		P2P:               client,
+		blockManager:      blockManager,
+		dalc:              dalc,
+		Mempool:           mp,
+		mempoolIDs:        mpIDs,
+		incomingTxCh:      make(chan []byte),
+		Store:             s,
+		TxIndexer:         txIndexer,
+		IndexerService:    indexerService,
+		BlockIndexer:      blockIndexer,
+		DoneBuildingBlock: doneBuildingChannel,
+		ctx:               ctx,
+		cancel:            cancel,
 	}
 
 	node.BaseService = *service.NewBaseService(logger, "Node", node)
@@ -255,7 +260,7 @@ func (n *FullNode) OnStart() error {
 	}
 	if n.conf.Aggregator {
 		n.Logger.Info("working in aggregator mode", "block time", n.conf.BlockTime)
-		go n.blockManager.AggregationLoop(n.ctx)
+		go n.blockManager.AggregationLoop(n.ctx, n.incomingTxCh, n.conf.LazyAggregator)
 		go n.headerPublishLoop(n.ctx)
 	}
 	go n.blockManager.RetrieveLoop(n.ctx)
@@ -313,6 +318,11 @@ func (n *FullNode) AppClient() abciclient.Client {
 	return n.appClient
 }
 
+type ResultDirectTx struct {
+	Included bool   `json:"included"`
+	Height   uint64 `json:"height"`
+}
+
 // newTxValidator creates a pubsub validator that uses the node's mempool to check the
 // transaction. If the transaction is valid, then it is added to the mempool
 func (n *FullNode) newTxValidator() p2p.GossipValidator {