feat: add key rotation

block/internal/executing/executor.go

@@ -126,7 +126,7 @@ func NewExecutor(
 			return nil, fmt.Errorf("failed to get address: %w", err)
 		}
 
-		if !bytes.Equal(addr, genesis.ProposerAddress) {
+		if !genesis.HasScheduledProposer(addr) {
 			return nil, common.ErrNotProposer
 		}
 	}
@@ -696,6 +696,10 @@ func (e *Executor) RetrieveBatch(ctx context.Context) (*BatchData, error) {
 func (e *Executor) CreateBlock(ctx context.Context, height uint64, batchData *BatchData) (*types.SignedHeader, *types.Data, error) {
 	currentState := e.getLastState()
 	headerTime := uint64(e.genesis.StartTime.UnixNano())
+	proposer, err := e.genesis.ProposerAtHeight(height)
+	if err != nil {
+		return nil, nil, fmt.Errorf("resolve proposer for height %d: %w", height, err)
+	}
 
 	var lastHeaderHash types.Hash
 	var lastDataHash types.Hash
@@ -728,22 +732,30 @@ func (e *Executor) CreateBlock(ctx context.Context, height uint64, batchData *Ba
 
 	// Get signer info and validator hash
 	var pubKey crypto.PubKey
+	var signerAddress []byte
 	var validatorHash types.Hash
 
 	if e.signer != nil {
-		var err error
 		pubKey, err = e.signer.GetPublic()
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to get public key: %w", err)
 		}
 
-		validatorHash, err = e.options.ValidatorHasherProvider(e.genesis.ProposerAddress, pubKey)
+		signerAddress, err = e.signer.GetAddress()
+		if err != nil {
+			return nil, nil, fmt.Errorf("failed to get signer address: %w", err)
+		}
+
+		if err := e.genesis.ValidateProposer(height, signerAddress, pubKey); err != nil {
+			return nil, nil, fmt.Errorf("signer does not match proposer schedule: %w", err)
+		}
+
+		validatorHash, err = e.options.ValidatorHasherProvider(proposer.Address, pubKey)
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to get validator hash: %w", err)
 		}
 	} else {
-		var err error
-		validatorHash, err = e.options.ValidatorHasherProvider(e.genesis.ProposerAddress, nil)
+		validatorHash, err = e.options.ValidatorHasherProvider(proposer.Address, nil)
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to get validator hash: %w", err)
 		}
@@ -763,13 +775,13 @@ func (e *Executor) CreateBlock(ctx context.Context, height uint64, batchData *Ba
 			},
 			LastHeaderHash:  lastHeaderHash,
 			AppHash:         currentState.AppHash,
-			ProposerAddress: e.genesis.ProposerAddress,
+			ProposerAddress: proposer.Address,
 			ValidatorHash:   validatorHash,
 		},
 		Signature: lastSignature,
 		Signer: types.Signer{
 			PubKey:  pubKey,
-			Address: e.genesis.ProposerAddress,
+			Address: proposer.Address,
 		},
 	}
 

block/internal/executing/executor_test.go

@@ -1,6 +1,7 @@
 package executing
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -12,6 +13,7 @@ import (
 
 	"github.com/evstack/ev-node/block/internal/cache"
 	"github.com/evstack/ev-node/block/internal/common"
+	coreseq "github.com/evstack/ev-node/core/sequencer"
 	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/genesis"
 	"github.com/evstack/ev-node/pkg/store"
@@ -121,3 +123,96 @@ func TestExecutor_NilBroadcasters(t *testing.T) {
 	assert.Equal(t, cacheManager, executor.cache)
 	assert.Equal(t, gen, executor.genesis)
 }
+
+func TestExecutor_CreateBlock_UsesScheduledProposerForHeight(t *testing.T) {
+	ds := sync.MutexWrap(datastore.NewMapDatastore())
+	memStore := store.New(ds)
+
+	cacheManager, err := cache.NewManager(config.DefaultConfig(), memStore, zerolog.Nop())
+	require.NoError(t, err)
+
+	metrics := common.NopMetrics()
+	oldAddr, oldSignerInfo, _ := buildTestSigner(t)
+	newAddr, newSignerInfo, newSigner := buildTestSigner(t)
+
+	entry1, err := genesis.NewProposerScheduleEntry(1, oldSignerInfo.PubKey)
+	require.NoError(t, err)
+	entry2, err := genesis.NewProposerScheduleEntry(2, newSignerInfo.PubKey)
+	require.NoError(t, err)
+
+	gen := genesis.Genesis{
+		ChainID:                "test-chain",
+		InitialHeight:          1,
+		StartTime:              time.Now().Add(-time.Second),
+		ProposerAddress:        entry1.Address,
+		ProposerSchedule:       []genesis.ProposerScheduleEntry{entry1, entry2},
+		DAEpochForcedInclusion: 1,
+	}
+
+	executor, err := NewExecutor(
+		memStore,
+		nil,
+		nil,
+		newSigner,
+		cacheManager,
+		metrics,
+		config.DefaultConfig(),
+		gen,
+		nil,
+		nil,
+		zerolog.Nop(),
+		common.DefaultBlockOptions(),
+		make(chan error, 1),
+		nil,
+	)
+	require.NoError(t, err)
+
+	prevHeader := &types.SignedHeader{
+		Header: types.Header{
+			Version: types.InitStateVersion,
+			BaseHeader: types.BaseHeader{
+				ChainID: gen.ChainID,
+				Height:  1,
+				Time:    uint64(gen.StartTime.UnixNano()),
+			},
+			AppHash:         []byte("state-root-0"),
+			ProposerAddress: oldAddr,
+			DataHash:        common.DataHashForEmptyTxs,
+		},
+		Signature: types.Signature([]byte("sig-1")),
+		Signer:    oldSignerInfo,
+	}
+	prevData := &types.Data{
+		Metadata: &types.Metadata{
+			ChainID: gen.ChainID,
+			Height:  1,
+			Time:    prevHeader.BaseHeader.Time,
+		},
+		Txs: nil,
+	}
+
+	batch, err := memStore.NewBatch(context.Background())
+	require.NoError(t, err)
+	require.NoError(t, batch.SaveBlockData(prevHeader, prevData, &prevHeader.Signature))
+	require.NoError(t, batch.SetHeight(1))
+	require.NoError(t, batch.Commit())
+
+	executor.setLastState(types.State{
+		Version:         types.InitStateVersion,
+		ChainID:         gen.ChainID,
+		InitialHeight:   gen.InitialHeight,
+		LastBlockHeight: 1,
+		LastBlockTime:   prevHeader.Time(),
+		LastHeaderHash:  prevHeader.Hash(),
+		AppHash:         []byte("state-root-1"),
+	})
+
+	header, data, err := executor.CreateBlock(context.Background(), 2, &BatchData{
+		Batch: &coreseq.Batch{},
+		Time:  time.Now(),
+	})
+	require.NoError(t, err)
+	require.Equal(t, newAddr, header.ProposerAddress)
+	require.Equal(t, newAddr, header.Signer.Address)
+	require.Equal(t, uint64(2), data.Height())
+}

block/internal/submitting/da_submitter.go

@@ -1,7 +1,6 @@
 package submitting
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -476,10 +475,6 @@ func (s *DASubmitter) signData(ctx context.Context, unsignedDataList []*types.Si
 		return nil, nil, fmt.Errorf("failed to get address: %w", err)
 	}
 
-	if len(genesis.ProposerAddress) > 0 && !bytes.Equal(addr, genesis.ProposerAddress) {
-		return nil, nil, fmt.Errorf("signer address mismatch with genesis proposer")
-	}
-
 	signerInfo := types.Signer{
 		PubKey:  pubKey,
 		Address: addr,
@@ -494,6 +489,10 @@ func (s *DASubmitter) signData(ctx context.Context, unsignedDataList []*types.Si
 			continue
 		}
 
+		if err := genesis.ValidateProposer(unsignedData.Height(), addr, pubKey); err != nil {
+			return nil, nil, fmt.Errorf("signer does not match proposer schedule for data at height %d: %w", unsignedData.Height(), err)
+		}
+
 		signature, err := signer.Sign(ctx, unsignedDataListBz[i])
 		if err != nil {
 			return nil, nil, fmt.Errorf("failed to sign data: %w", err)

block/internal/submitting/da_submitter_test.go

@@ -343,6 +343,97 @@ func TestDASubmitter_SubmitData_Success(t *testing.T) {
 	assert.True(t, ok)
 }
 
+func TestDASubmitter_SubmitData_UsesScheduledProposerForHeight(t *testing.T) {
+	submitter, st, cm, mockDA, gen := setupDASubmitterTest(t)
+	ctx := context.Background()
+	dataNamespace := datypes.NamespaceFromString(testDataNamespace).Bytes()
+
+	mockDA.On(
+		"Submit",
+		mock.Anything,
+		mock.AnythingOfType("[][]uint8"),
+		mock.AnythingOfType("float64"),
+		dataNamespace,
+		mock.Anything,
+	).Return(func(_ context.Context, blobs [][]byte, _ float64, _ []byte, _ []byte) datypes.ResultSubmit {
+		return datypes.ResultSubmit{BaseResult: datypes.BaseResult{Code: datypes.StatusSuccess, SubmittedCount: uint64(len(blobs)), Height: 2}}
+	}).Once()
+
+	oldAddr, oldPub, _ := createTestSigner(t)
+	nextAddr, nextPub, nextSigner := createTestSigner(t)
+
+	entry1, err := genesis.NewProposerScheduleEntry(gen.InitialHeight, oldPub)
+	require.NoError(t, err)
+	entry2, err := genesis.NewProposerScheduleEntry(2, nextPub)
+	require.NoError(t, err)
+
+	gen.ProposerAddress = entry1.Address
+	gen.ProposerSchedule = []genesis.ProposerScheduleEntry{entry1, entry2}
+	submitter.genesis = gen
+
+	data1 := &types.Data{
+		Metadata: &types.Metadata{
+			ChainID: gen.ChainID,
+			Height:  1,
+			Time:    uint64(time.Now().UnixNano()),
+		},
+		Txs: types.Txs{},
+	}
+
+	header1 := &types.SignedHeader{
+		Header: types.Header{
+			BaseHeader: types.BaseHeader{
+				ChainID: gen.ChainID,
+				Height:  1,
+				Time:    uint64(time.Now().UnixNano()),
+			},
+			ProposerAddress: oldAddr,
+			DataHash:        common.DataHashForEmptyTxs,
+		},
+		Signer: types.Signer{PubKey: oldPub, Address: oldAddr},
+	}
+
+	data := &types.Data{
+		Metadata: &types.Metadata{
+			ChainID: gen.ChainID,
+			Height:  2,
+			Time:    uint64(time.Now().UnixNano()),
+		},
+		Txs: types.Txs{types.Tx("rotated-key-tx")},
+	}
+
+	header := &types.SignedHeader{
+		Header: types.Header{
+			BaseHeader: types.BaseHeader{
+				ChainID: gen.ChainID,
+				Height:  2,
+				Time:    uint64(time.Now().UnixNano()),
+			},
+			ProposerAddress: nextAddr,
+			DataHash:        data.DACommitment(),
+		},
+		Signer: types.Signer{PubKey: nextPub, Address: nextAddr},
+	}
+
+	sig1 := types.Signature([]byte("sig-1"))
+	sig2 := types.Signature([]byte("sig-2"))
+	batch, err := st.NewBatch(ctx)
+	require.NoError(t, err)
+	require.NoError(t, batch.SaveBlockData(header1, data1, &sig1))
+	require.NoError(t, batch.SaveBlockData(header, data, &sig2))
+	require.NoError(t, batch.SetHeight(2))
+	require.NoError(t, batch.Commit())
+
+	signedDataList, marshalledData, err := cm.GetPendingData(ctx)
+	require.NoError(t, err)
+	err = submitter.SubmitData(ctx, signedDataList, marshalledData, cm, nextSigner, gen)
+	require.NoError(t, err)
+
+	_, ok := cm.GetDataDAIncludedByHeight(2)
+	assert.True(t, ok)
+	assert.NotEqual(t, oldAddr, nextAddr)
+}
+
 func TestDASubmitter_SubmitData_SkipsEmptyData(t *testing.T) {
 	submitter, st, cm, mockDA, gen := setupDASubmitterTest(t)
 	ctx := context.Background()

block/internal/syncing/assert.go

@@ -1,19 +1,20 @@
 package syncing
 
 import (
-	"bytes"
 	"errors"
 	"fmt"
 
+	"github.com/libp2p/go-libp2p/core/crypto"
+
 	"github.com/evstack/ev-node/pkg/genesis"
 	"github.com/evstack/ev-node/types"
 )
 
-func assertExpectedProposer(genesis genesis.Genesis, proposerAddr []byte) error {
-	if !bytes.Equal(proposerAddr, genesis.ProposerAddress) {
-		return fmt.Errorf("unexpected proposer: got %x, expected %x",
-			proposerAddr, genesis.ProposerAddress)
+func assertExpectedProposer(genesis genesis.Genesis, height uint64, proposerAddr []byte, pubKey crypto.PubKey) error {
+	if err := genesis.ValidateProposer(height, proposerAddr, pubKey); err != nil {
+		return fmt.Errorf("unexpected proposer at height %d: %w", height, err)
 	}
+
 	return nil
 }
 
@@ -22,7 +23,7 @@ func assertValidSignedData(signedData *types.SignedData, genesis genesis.Genesis
 		return errors.New("empty signed data")
 	}
 
-	if err := assertExpectedProposer(genesis, signedData.Signer.Address); err != nil {
+	if err := assertExpectedProposer(genesis, signedData.Height(), signedData.Signer.Address, signedData.Signer.PubKey); err != nil {
 		return err
 	}
 

block/internal/syncing/da_retriever.go

@@ -299,7 +299,7 @@ func (r *daRetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedH
 		return nil
 	}
 
-	if err := r.assertExpectedProposer(header.ProposerAddress); err != nil {
+	if err := r.assertExpectedProposer(header); err != nil {
 		r.logger.Debug().Err(err).Msg("unexpected proposer")
 		return nil
 	}
@@ -355,9 +355,9 @@ func (r *daRetriever) tryDecodeData(bz []byte, daHeight uint64) *types.Data {
 	return &signedData.Data
 }
 
-// assertExpectedProposer validates the proposer address
-func (r *daRetriever) assertExpectedProposer(proposerAddr []byte) error {
-	return assertExpectedProposer(r.genesis, proposerAddr)
+// assertExpectedProposer validates the proposer schedule entry for the header height.
+func (r *daRetriever) assertExpectedProposer(header *types.SignedHeader) error {
+	return assertExpectedProposer(r.genesis, header.Height(), header.ProposerAddress, header.Signer.PubKey)
 }
 
 // assertValidSignedData validates signed data using the configured signature provider

block/internal/syncing/p2p_handler.go

@@ -81,7 +81,7 @@ func (h *P2PHandler) ProcessHeight(ctx context.Context, height uint64, heightInC
 		}
 		return err
 	}
-	if err := h.assertExpectedProposer(p2pHeader.ProposerAddress); err != nil {
+	if err := h.assertExpectedProposer(p2pHeader.SignedHeader); err != nil {
 		h.logger.Debug().Uint64("height", height).Err(err).Msg("invalid header from P2P")
 		return err
 	}
@@ -125,11 +125,7 @@ func (h *P2PHandler) ProcessHeight(ctx context.Context, height uint64, heightInC
 	return nil
 }
 
-// assertExpectedProposer validates the proposer address.
-func (h *P2PHandler) assertExpectedProposer(proposerAddr []byte) error {
-	if !bytes.Equal(h.genesis.ProposerAddress, proposerAddr) {
-		return fmt.Errorf("proposer address mismatch: got %x, expected %x",
-			proposerAddr, h.genesis.ProposerAddress)
-	}
-	return nil
+// assertExpectedProposer validates the proposer schedule entry for the header height.
+func (h *P2PHandler) assertExpectedProposer(header *types.SignedHeader) error {
+	return assertExpectedProposer(h.genesis, header.Height(), header.ProposerAddress, header.Signer.PubKey)
 }