autobahn mempool

sei-tendermint/abci/types/types.go

@@ -232,12 +232,12 @@ type ResponseCheckTxV2 struct {
 	*ResponseCheckTx
 
 	// helper properties for prioritization in mempool
+	IsEVM    bool
 	EVMNonce uint64
 	// EVM and sei addresses are both derived from the sender's public key.
 	// TODO(gprusak): include just the secp256k1 public key and let the CheckTx caller derive evm/sei address on their own.
 	EVMSenderAddress   common.Address
 	SeiSenderAddress   []byte
-	IsEVM              bool
 	EVMRequiredBalance *big.Int
 }
 

sei-tendermint/cmd/tendermint/commands/gen_autobahn_config.go

@@ -82,13 +82,11 @@ Output is written to the file specified by --output.`,
 			}
 
 			cfg := config.AutobahnFileConfig{
-				Validators:       validators,
-				MaxTxsPerBlock:   5_000,
-				MempoolSize:      5_000,
-				BlockInterval:    utils.Duration(400 * time.Millisecond),
-				AllowEmptyBlocks: true,
-				ViewTimeout:      utils.Duration(1500 * time.Millisecond),
-				DialInterval:     utils.Duration(10 * time.Second),
+				Validators:     validators,
+				MaxTxsPerBlock: 5_000,
+				BlockInterval:  utils.Duration(400 * time.Millisecond),
+				ViewTimeout:    utils.Duration(1500 * time.Millisecond),
+				DialInterval:   utils.Duration(10 * time.Second),
 			}
 			// The flag defaults to "data/autobahn" so persistence is on without
 			// operator action. node/setup.go rootifies the relative path against

sei-tendermint/config/autobahn.go

@@ -45,9 +45,7 @@ type AutobahnFileConfig struct {
 	Validators         []AutobahnValidator  `json:"validators"`
 	MaxTxsPerBlock     uint64               `json:"max_txs_per_block"`
 	MaxTxsPerSecond    utils.Option[uint64] `json:"max_txs_per_second"`
-	MempoolSize        uint64               `json:"mempool_size"`
 	BlockInterval      utils.Duration       `json:"block_interval"`
-	AllowEmptyBlocks   bool                 `json:"allow_empty_blocks"`
 	ViewTimeout        utils.Duration       `json:"view_timeout"`
 	PersistentStateDir utils.Option[string] `json:"persistent_state_dir"`
 	DialInterval       utils.Duration       `json:"dial_interval"`
@@ -61,9 +59,6 @@ func (fc *AutobahnFileConfig) Validate() error {
 	if fc.MaxTxsPerBlock == 0 {
 		return errors.New("max_txs_per_block must be > 0")
 	}
-	if fc.MempoolSize == 0 {
-		return errors.New("mempool_size must be > 0")
-	}
 	if fc.BlockInterval <= 0 {
 		return errors.New("block_interval must be > 0")
 	}

sei-tendermint/internal/autobahn/autobahn.proto

@@ -74,12 +74,11 @@ message BlockHeader {
 }
 
 message Payload {
+  reserved "edge_count", "coinbase", "basefee";
+  reserved 3, 4, 5;
   option (hashable.hashable) = true;
   optional Timestamp created_at = 1; // required
   optional uint64 total_gas = 2; // required
-  optional int64 edge_count = 3; // required
-  optional bytes coinbase = 4; // required
-  optional int64 basefee = 5; // required
   repeated bytes txs = 6;
 }
 

sei-tendermint/internal/autobahn/avail/state.go

@@ -40,6 +40,10 @@ type State struct {
 	persisters persisters
 }
 
+func (s *State) PublicKey() types.PublicKey {
+	return s.key.Public()
+}
+
 // persisters holds all disk persistence components. Either all are present
 // (real I/O) or all are no-op (testing). It is a pure I/O struct — all inner
 // state access goes through State methods.
@@ -521,11 +525,11 @@ func (s *State) fullCommitQC(ctx context.Context, n types.RoadIndex) (*types.Ful
 }
 
 // WaitForCapacity waits until the given lane has enough capacity for a new block.
-func (s *State) WaitForCapacity(ctx context.Context, lane types.LaneID) error {
+func (s *State) WaitForCapacity(ctx context.Context, toProduce types.BlockNumber) error {
+	lane := s.key.Public()
 	for inner, ctrl := range s.inner.Lock() {
-		q := inner.blocks[lane]
 		if err := ctrl.WaitUntil(ctx, func() bool {
-			return q.next < inner.persistedBlockStart[lane]+BlocksPerLane
+			return toProduce < inner.persistedBlockStart[lane]+BlocksPerLane
 		}); err != nil {
 			return err
 		}
@@ -564,23 +568,24 @@ func (s *State) WaitForLaneQCs(
 
 // ProduceBlock appends a new block to the producers lane.
 // Blocks until the lane has enough capacity for the new block.
-func (s *State) ProduceBlock(ctx context.Context, payload *types.Payload) (*types.Signed[*types.LaneProposal], error) {
-	return s.produceBlock(ctx, s.key, payload)
+func (s *State) ProduceBlock(n types.BlockNumber, payload *types.Payload) (*types.Signed[*types.LaneProposal], error) {
+	return s.produceBlock(n, s.key, payload)
 }
 
 // TODO: produceBlock is a separate function for testing - consider improving the tests to use ProduceBlock only.
-func (s *State) produceBlock(ctx context.Context, key types.SecretKey, payload *types.Payload) (*types.Signed[*types.LaneProposal], error) {
+func (s *State) produceBlock(n types.BlockNumber, key types.SecretKey, payload *types.Payload) (*types.Signed[*types.LaneProposal], error) {
 	lane := key.Public()
 	var result *types.Signed[*types.LaneProposal]
 	for inner, ctrl := range s.inner.Lock() {
 		q, ok := inner.blocks[lane]
 		if !ok {
 			return nil, ErrBadLane
 		}
-		if err := ctrl.WaitUntil(ctx, func() bool {
-			return q.next < inner.persistedBlockStart[lane]+BlocksPerLane
-		}); err != nil {
-			return nil, err
+		if n >= inner.persistedBlockStart[lane]+BlocksPerLane {
+			return nil, fmt.Errorf("lane full")
+		}
+		if q.next != n {
+			return nil, fmt.Errorf("unexpected block number: got %v, want %v", n, q.next)
 		}
 		var parent types.BlockHeaderHash
 		if q.first < q.next {

sei-tendermint/internal/autobahn/avail/state_test.go

@@ -128,7 +128,7 @@ func testState(t *testing.T, stateDir utils.Option[string]) {
 				lane := key.Public()
 				p := types.GenPayload(rng)
 				want[lane] = append(want[lane], p.Hash())
-				b, err := state.produceBlock(ctx, key, p)
+				b, err := state.produceBlock(state.NextBlock(lane), key, p)
 				if err != nil {
 					return fmt.Errorf("state.ProduceBlock(): %w", err)
 				}
@@ -254,7 +254,7 @@ func TestStateRestartFromPersisted(t *testing.T) {
 
 			for range 5 {
 				key := keys[rng.Intn(len(keys))]
-				if _, err := state.produceBlock(ctx, key, types.GenPayload(rng)); err != nil {
+				if _, err := state.produceBlock(state.NextBlock(key.Public()), key, types.GenPayload(rng)); err != nil {
 					return fmt.Errorf("produceBlock: %w", err)
 				}
 			}
@@ -340,7 +340,6 @@ func TestStateMismatchedQCs(t *testing.T) {
 	}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
 	state, err := NewState(keys[0], ds, utils.None[string]())
 	require.NoError(t, err)
-	ctx := t.Context()
 
 	// Helper to create a CommitQC for a specific index
 	makeQC := func(prev utils.Option[*types.CommitQC], laneQCs map[types.LaneID]*types.LaneQC) *types.CommitQC {
@@ -364,7 +363,7 @@ func TestStateMismatchedQCs(t *testing.T) {
 	// 1. Produce a block so we have a non-empty range
 	lane := keys[0].Public()
 	p := types.GenPayload(rng)
-	b, err := state.ProduceBlock(ctx, p)
+	b, err := state.ProduceBlock(state.NextBlock(lane), p)
 	require.NoError(t, err)
 
 	// 2. Form a LaneQC for it
@@ -398,7 +397,7 @@ func TestPushBlockRejectsBadParentHash(t *testing.T) {
 	state := utils.OrPanic1(NewState(keys[0], ds, utils.None[string]()))
 
 	// Produce a valid first block on our lane.
-	_, err := state.ProduceBlock(ctx, types.GenPayload(rng))
+	_, err := state.ProduceBlock(state.NextBlock(keys[0].Public()), types.GenPayload(rng))
 	require.NoError(t, err)
 
 	// Create a second block with a fake parentHash.

sei-tendermint/internal/autobahn/consensus/state.go

@@ -154,18 +154,6 @@ func (s *State) commitQC() utils.AtomicRecv[utils.Option[*types.CommitQC]] {
 	panic("unreachable")
 }
 
-// WaitForCapacity waits until a new block can be produced by this node.
-func (s *State) WaitForCapacity(ctx context.Context) error {
-	return s.avail.WaitForCapacity(ctx, s.cfg.Key.Public())
-}
-
-// ProduceBlock produces a new block with the given payload.
-// Returns ErrNoCapacity if there is currently no capacity for the next block.
-// Run WaitForCapacity before calling ProduceBlock.
-func (s *State) ProduceBlock(ctx context.Context, payload *types.Payload) (*types.Signed[*types.LaneProposal], error) {
-	return s.avail.ProduceBlock(ctx, payload)
-}
-
 // PushProposal processes an unverified FullProposal message.
 func (s *State) PushProposal(ctx context.Context, proposal *types.FullProposal) error {
 	return s.pushProposal(ctx, proposal)

sei-tendermint/internal/autobahn/data/state.go

@@ -145,8 +145,9 @@ type appProposalWithTimestamp struct {
 }
 
 type inner struct {
-	qcs          map[types.GlobalBlockNumber]*types.FullCommitQC      // [first,nextQC)
-	blocks       map[types.GlobalBlockNumber]*types.Block             // [first,nextBlock) + subset of [nextBlock,nextQC)
+	qcs    map[types.GlobalBlockNumber]*types.FullCommitQC // [first,nextQC)
+	blocks map[types.GlobalBlockNumber]*types.Block        // [first,nextBlock) + subset of [nextBlock,nextQC)
+	// appProposal[n] contains appProposal block >=n.
 	appProposals map[types.GlobalBlockNumber]appProposalWithTimestamp // [first,nextAppProposal)
 
 	// blockHashes is a hash → height index mirroring blocks. Maintained
@@ -565,6 +566,9 @@ func (s *State) PushAppHash(ctx context.Context, n types.GlobalBlockNumber, hash
 			proposal:  proposal,
 			timestamp: t,
 		}
+		// TODO(gprusak): this will be problematic on restart,
+		// nextAppProposal should be initiated wrt current application height,
+		// so that we don't iterate over all blocks in storage on startup.
 		for inner.nextAppProposal <= n {
 			b := inner.blocks[inner.nextAppProposal]
 			latency := t.Sub(b.Payload().CreatedAt()).Seconds()
@@ -594,6 +598,43 @@ func (s *State) AppProposal(ctx context.Context, n types.GlobalBlockNumber) (*ty
 	panic("unreachable")
 }
 
+func (i *inner) nextToExecute(lane types.LaneID) types.BlockNumber {
+	// TODO(gprusak): decide whether 0 is a good result in this case in general.
+	if i.first == i.nextAppProposal {
+		return 0
+	}
+	n := i.nextAppProposal - 1
+	r := i.qcs[n].QC().LaneRange(lane)
+	// TODO: this header can be actually extracted from FullCommitQC, so consider moving all this logic there.
+	h := i.blocks[n].Header()
+	x := lane.Compare(h.Lane())
+	// NOTE: here we assume the specific ordering of lane blocks in the CommitQC:
+	// TODO(gprusak): move this logic closer to CommitQC
+	switch {
+	case x < 0:
+		return r.Next()
+	case x > 0:
+		return r.First()
+	default:
+		return h.BlockNumber() + 1
+	}
+}
+
+// Waits until lane block n is executed, returns the next block of this lane to be executed (>n)
+func (s *State) WaitUntilExecuted(ctx context.Context, lane types.LaneID, n types.BlockNumber) (types.BlockNumber, error) {
+	for inner, ctrl := range s.inner.Lock() {
+		for {
+			if next := inner.nextToExecute(lane); n < next {
+				return next, nil
+			}
+			if err := ctrl.Wait(ctx); err != nil {
+				return 0, err
+			}
+		}
+	}
+	panic("unreachable")
+}
+
 // PruneBefore removes blocks, QCs, and AppProposals before retainFrom.
 // Blocks at retainFrom and above are kept. Per-block pruning may split
 // a QC range; this is handled on recovery (NewState skips partial QC prefixes).