rpc(autobahn): return the committee that certified /validators height (CON-358)

integration_test/autobahn/autobahn_test.go

@@ -847,7 +847,7 @@ func testBlockProduction(t *testing.T) {
 // assertTmRPCEndpoints exercises the tmRPC surface that PR #3310 wires up
 // under Autobahn (env.Block, env.BlockResults, env.BlockByHash, env.Validators).
 // One call per endpoint is enough — these handlers are pure RPC translation
-// over the same data.State / GenDoc plumbing, so a single positive case at
+// over data.State / the epoch registry, so a single positive case at
 // a real height catches both wrong-routing (e.g. CometBFT path returning
 // nulls because BlockStore is empty) and shape-drift regressions.
 func assertTmRPCEndpoints(t *testing.T, h int64) {
@@ -891,19 +891,28 @@ func assertTmRPCEndpoints(t *testing.T, h int64) {
 		t.Fatalf("/block_results?height=%d: got height=%d", h, rbr.Height)
 	}
 
-	// /validators at h: committee is fixed at genesis under Autobahn, so
-	// any retained height returns it. block_height in the response must
-	// match the requested height (catches the old "stuck at 1" StateStore
-	// behavior).
+	// /validators at h: committee covering that global block. Omitted
+	// height is Comet's "latest" and must resolve to the app tip
+	// (autobahnCheckAndGetHeight → LastBlockHeight).
 	var rv coretypes.ResultValidators
 	fetchTmRPC(t, fmt.Sprintf("%s/validators?height=%d", tmRPCBase, h), &rv)
 	if rv.BlockHeight != h {
-		t.Fatalf("/validators?height=%d: got block_height=%d (StateStore-stuck-at-1 regression?)",
-			h, rv.BlockHeight)
-	}
-	if rv.Total < 1 || len(rv.Validators) < 1 {
-		t.Fatalf("/validators?height=%d: empty committee (total=%d, count=%d)",
-			h, rv.Total, len(rv.Validators))
+		t.Fatalf("/validators?height=%d: got block_height=%d", h, rv.BlockHeight)
+	}
+	if rv.Total != clusterSize || len(rv.Validators) != clusterSize {
+		t.Fatalf("/validators?height=%d: committee size total=%d count=%d, want %d",
+			h, rv.Total, len(rv.Validators), clusterSize)
+	}
+	var latest coretypes.ResultValidators
+	fetchTmRPC(t, tmRPCBase+"/validators", &latest)
+	tip := currentHeight(t)
+	if latest.BlockHeight < h || latest.BlockHeight > tip {
+		t.Fatalf("/validators: got block_height=%d, want in [%d, %d] (requested height, /abci_info last_block_height)",
+			latest.BlockHeight, h, tip)
+	}
+	if latest.Total != clusterSize || len(latest.Validators) != clusterSize {
+		t.Fatalf("/validators: committee size total=%d count=%d, want %d",
+			latest.Total, len(latest.Validators), clusterSize)
 	}
 }
 

sei-tendermint/autobahn/types/msg.go

@@ -96,6 +96,9 @@ func (k PublicKey) Compare(other PublicKey) int { return k.key.Compare(other.key
 // Bytes converts the public key to bytes.
 func (k PublicKey) Bytes() []byte { return k.key.Bytes() }
 
+// ED25519 returns the underlying Ed25519 public key.
+func (k PublicKey) ED25519() ed25519.PublicKey { return k.key }
+
 // PublicKeyFromBytes constructs a public key from bytes.
 func PublicKeyFromBytes(b []byte) (PublicKey, error) {
 	k, err := ed25519.PublicKeyFromBytes(b)

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

@@ -560,6 +560,22 @@ func (s *State) TryBlock(n types.GlobalBlockNumber) (*types.Block, error) {
 	return s.blockFromDB(n)
 }
 
+// TryQC returns the FullCommitQC covering global height n without waiting.
+// Returns ErrNotFound if n is not yet covered (n >= nextQC).
+// Returns ErrPruned if BlockStore no longer has an evicted height.
+func (s *State) TryQC(n types.GlobalBlockNumber) (*types.FullCommitQC, error) {
+	for inner := range s.inner.Lock() {
+		if n >= inner.nextQC {
+			return nil, types.ErrNotFound
+		}
+		if n < inner.first {
+			break
+		}
+		return inner.qcs[n].qc, nil
+	}
+	return s.qcFromDB(n)
+}
+
 // NeedBlock reports whether catch-up still needs to fetch height n.
 // False when n is already past nextBlock (including heights pruned or
 // evicted from RAM) or an in-memory gap-fill is present. Unlike TryBlock,

sei-tendermint/internal/autobahn/epoch/registry.go

@@ -304,6 +304,7 @@ func (r *Registry) Pending() utils.Option[types.EpochIndex] {
 // PruneBefore drops epochs in [live.First, keep). Epochs 0 and 1 and the
 // latest live epoch are kept. keep is exclusive and only moves live.First
 // forward. Staged committees are not dropped.
+// TODO: archive nodes need special handling (never prune epochs).
 func (r *Registry) PruneBefore(keep types.EpochIndex) error {
 	for s, ctrl := range r.state.Lock() {
 		if s.live.First == s.live.Next {

sei-tendermint/internal/p2p/giga_router.go

@@ -75,4 +75,5 @@ type GigaRouter interface {
 	BlockByHash(ctx context.Context, hash atypes.BlockHeaderHash) (*coretypes.ResultBlock, error)
 	EvmProxy(sender common.Address) utils.Option[*rpc.Client]
 	Mempool() utils.Option[*producer.State]
+	Validators(n atypes.GlobalBlockNumber) ([]*types.Validator, atypes.GlobalBlockNumber, error)
 }

sei-tendermint/internal/p2p/giga_router_common.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"slices"
+	"sort"
 	"sync/atomic"
 
 	ethrpc "github.com/ethereum/go-ethereum/rpc"
@@ -565,6 +566,52 @@ func (r *gigaRouterCommon) RunInboundConn(ctx context.Context, hConn *handshaked
 	})
 }
 
+// Validators returns the Autobahn validator set that certified global height n.
+// Before the first CommitQC, FirstBlock resolves to the genesis committee.
+func (r *gigaRouterCommon) Validators(n atypes.GlobalBlockNumber) ([]*types.Validator, atypes.GlobalBlockNumber, error) {
+	first := r.data.Registry().FirstBlock()
+	qc, err := r.data.TryQC(n)
+	var epochIndex atypes.EpochIndex
+	if errors.Is(err, atypes.ErrNotFound) && n == first {
+		epochIndex = 0
+	} else if err != nil {
+		return nil, 0, heightLookupError(n, err)
+	} else {
+		epochIndex = qc.QC().Proposal().EpochIndex()
+	}
+	ep, err := r.data.Registry().EpochByIndex(epochIndex)
+	if err != nil {
+		return nil, 0, heightLookupError(n, err)
+	}
+	vs, err := committeeValidators(ep.Committee())
+	return vs, n, err
+}
+
+// heightLookupError returns the RPC error for a data lookup at global height n.
+func heightLookupError(n atypes.GlobalBlockNumber, err error) error {
+	switch {
+	case errors.Is(err, atypes.ErrPruned):
+		return coretypes.WrapErrHeightNotAvailable(utils.Clamp[int64](n), utils.None[int64]())
+	case errors.Is(err, atypes.ErrNotFound):
+		return fmt.Errorf("%w (requested height: %d)", coretypes.ErrHeightExceedsChainHead, utils.Clamp[int64](n))
+	default:
+		return err
+	}
+}
+
+func committeeValidators(committee *atypes.Committee) ([]*types.Validator, error) {
+	vs := make([]*types.Validator, 0, committee.Lanes().Len())
+	for lane := range committee.Lanes().All() {
+		power, ok := utils.SafeCast[int64](committee.Weight(lane.Validator))
+		if !ok {
+			return nil, fmt.Errorf("committee member %v: weight %d does not fit int64", lane.Validator, committee.Weight(lane.Validator))
+		}
+		vs = append(vs, types.NewValidator(lane.Validator.ED25519(), power))
+	}
+	sort.Sort(types.ValidatorsByVotingPower(vs))
+	return vs, nil
+}
+
 // EvmProxy returns the shard owner's EVMRPC client for an EVM tx sender, or
 // None if the caller should handle it locally. Overridden on
 // *gigaValidatorRouter to short-circuit self-shard sends.

sei-tendermint/internal/p2p/giga_router_common_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/rpc/coretypes"
 	tmtypes "github.com/sei-protocol/sei-chain/sei-tendermint/types"
 )
 
@@ -148,6 +149,102 @@ func TestBuildDataStateStartsRecoveryAtAppTip(t *testing.T) {
 	require.Equal(t, blocks[gr.Len()/2].Header().Hash(), got.Header().Hash())
 }
 
+func TestGigaRouterCommon_ValidatorsAtGlobalHeight(t *testing.T) {
+	rng := utils.TestRng()
+	low := atypes.GenSecretKey(rng)
+	mid := atypes.GenSecretKey(rng)
+	high := atypes.GenSecretKey(rng)
+	keys := []atypes.SecretKey{low, mid, high}
+	router := testGigaRouterWithData(t, map[atypes.PublicKey]GigaNodeAddr{
+		low.Public():  {},
+		mid.Public():  {},
+		high.Public(): {},
+	})
+	first := router.data.Registry().FirstBlock()
+
+	got, h, err := router.Validators(first)
+	require.NoError(t, err)
+	require.Equal(t, first, h)
+	require.Len(t, got, 3)
+	require.Equal(t, []int64{1, 1, 1}, []int64{got[0].VotingPower, got[1].VotingPower, got[2].VotingPower})
+
+	_, _, err = router.Validators(0)
+	require.ErrorIs(t, err, coretypes.ErrHeightNotAvailable)
+
+	_, _, err = router.Validators(first + 100)
+	require.ErrorIs(t, err, coretypes.ErrHeightExceedsChainHead)
+
+	weights := map[atypes.PublicKey]uint64{
+		low.Public():  1,
+		mid.Public():  5,
+		high.Public(): 10,
+	}
+	require.NoError(t, router.data.Registry().StageAndActivate(0, weights))
+	fakeNext := utils.NewAtomicSend(router.data.Registry().MustEpoch(2))
+	router.nextCommitEpoch = fakeNext.Subscribe()
+	got, h, err = router.Validators(first)
+	require.NoError(t, err)
+	require.Equal(t, first, h)
+	require.Equal(t, []int64{1, 1, 1}, []int64{got[0].VotingPower, got[1].VotingPower, got[2].VotingPower})
+
+	n := pushQCAtRoad(t, router, keys, router.data.Registry().MustEpoch(2), epoch.FirstRoad(2))
+	got, h, err = router.Validators(n)
+	require.NoError(t, err)
+	require.Equal(t, n, h)
+	require.Equal(t, []int64{10, 5, 1}, []int64{got[0].VotingPower, got[1].VotingPower, got[2].VotingPower})
+	require.Equal(t, high.Public().Bytes(), got[0].PubKey.Bytes())
+	require.Equal(t, mid.Public().Bytes(), got[1].PubKey.Bytes())
+	require.Equal(t, low.Public().Bytes(), got[2].PubKey.Bytes())
+
+	require.NoError(t, router.data.Registry().StageAndActivate(1, weights))
+	require.NoError(t, router.data.Registry().StageAndActivate(2, weights))
+	require.NoError(t, router.data.Registry().PruneBefore(4))
+	_, _, err = router.Validators(n)
+	require.ErrorIs(t, err, coretypes.ErrHeightNotAvailable)
+}
+
+func pushQCAtRoad(t *testing.T, router *gigaRouterCommon, keys []atypes.SecretKey, ep *atypes.Epoch, road atypes.RoadIndex) atypes.GlobalBlockNumber {
+	t.Helper()
+	first := router.data.Registry().FirstBlock()
+	proposal, blocks := atypes.ProposalAtBlocks(ep, atypes.View{Index: road, Number: 0}, first, 1)
+	votes := make([]*atypes.Signed[*atypes.CommitVote], 0, len(keys))
+	for _, k := range keys {
+		votes = append(votes, atypes.Sign(k, atypes.NewCommitVote(proposal)))
+	}
+	headers := make([]*atypes.BlockHeader, len(blocks))
+	for i, b := range blocks {
+		headers[i] = b.Header()
+	}
+	qc := atypes.NewFullCommitQC(atypes.NewCommitQC(votes), headers)
+	require.NoError(t, router.data.PushQC(t.Context(), qc, blocks))
+	return first
+}
+
+func testGigaRouterWithData(t *testing.T, addrs map[atypes.PublicKey]GigaNodeAddr) *gigaRouterCommon {
+	t.Helper()
+	genDoc := &tmtypes.GenesisDoc{
+		ChainID:         "validators-road-test",
+		InitialHeight:   1,
+		GenesisTime:     time.Now(),
+		ConsensusParams: tmtypes.DefaultConsensusParams(),
+	}
+	require.NoError(t, genDoc.ValidateAndComplete())
+	db, err := blockstore.New(memblock.NewBlockDB())
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, db.Close()) })
+	state, err := BuildDataState(&GigaRouterCommonConfig{
+		DialInterval:   time.Second,
+		ValidatorAddrs: addrs,
+		GenDoc:         genDoc,
+		App:            proxy.New(&fixedHeightApp{height: 1}),
+	}, db)
+	require.NoError(t, err)
+	return &gigaRouterCommon{
+		data:            state,
+		nextCommitEpoch: state.NextCommitEpoch(),
+	}
+}
+
 func TestCommitteeWeights(t *testing.T) {
 	rng := utils.TestRng()
 	sk := ed25519.TestSecretKey(utils.GenBytes(rng, 32))

sei-tendermint/internal/rpc/core/autobahn_env_test.go

@@ -0,0 +1,101 @@
+package core
+
+import (
+	"testing"
+	"time"
+
+	dbm "github.com/tendermint/tm-db"
+	"golang.org/x/time/rate"
+
+	"github.com/sei-protocol/sei-chain/sei-db/ledger_db/block/memblock"
+	abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/blockstore"
+	atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/config"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/producer"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/conn"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer"
+	kvsink "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer/sink/kv"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/tcp"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/types"
+)
+
+func newAutobahnBroadcastEnv(t *testing.T) *Environment {
+	t.Helper()
+	rng := utils.TestRng()
+	_, keys := atypes.GenCommittee(rng, 1)
+	valKey := keys[0]
+	nodeKey := p2p.NodeSecretKey(ed25519.TestSecretKey(utils.GenBytes(rng, 32)))
+	genDoc := &types.GenesisDoc{
+		ChainID:         "rpc-autobahn",
+		InitialHeight:   1,
+		GenesisTime:     time.Now(),
+		ConsensusParams: types.DefaultConsensusParams(),
+	}
+	require.NoError(t, genDoc.ValidateAndComplete())
+	addrs := map[atypes.PublicKey]p2p.GigaNodeAddr{
+		valKey.Public(): {
+			Key:      nodeKey.Public(),
+			HostPort: tcp.HostPort{Hostname: "127.0.0.1", Port: 26657},
+		},
+	}
+	blockStore, err := blockstore.New(memblock.NewBlockDB())
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, blockStore.Close()) })
+	app := proxy.New(&abci.BaseApplication{})
+	commonCfg := p2p.GigaRouterCommonConfig{
+		DialInterval:   time.Second,
+		ValidatorAddrs: addrs,
+		App:            app,
+		GenDoc:         genDoc,
+	}
+	dataState, err := p2p.BuildDataState(&commonCfg, blockStore)
+	require.NoError(t, err)
+	giga, err := p2p.NewGigaValidatorRouter(&p2p.GigaValidatorConfig{
+		GigaRouterCommonConfig: commonCfg,
+		ValidatorKey:           valKey,
+		ViewTimeout:            func(atypes.View) time.Duration { return time.Hour },
+		Producer: &producer.Config{
+			MaxGasWantedPerBlock:    1,
+			MaxGasEstimatedPerBlock: 1,
+			MaxTxsPerBlock:          1,
+			MaxTxsPerSecond:         utils.None[uint64](),
+			BlockInterval:           time.Second,
+		},
+	}, nodeKey, dataState)
+	require.NoError(t, err)
+	require.True(t, giga.Mempool().IsPresent(), "validator GigaRouter must expose a mempool so a regression would wait, not take the fullnode shortcut")
+	endpoint := p2p.Endpoint{AddrPort: tcp.TestReserveAddr()}
+	nodeInfo := types.NodeInfo{
+		NodeID:     nodeKey.Public().NodeID(),
+		ListenAddr: endpoint.String(),
+		Moniker:    string(nodeKey.Public().NodeID()),
+		Network:    genDoc.ChainID,
+	}
+	router, err := p2p.NewRouter(
+		nodeKey,
+		func() *types.NodeInfo { return &nodeInfo },
+		dbm.NewMemDB(),
+		&p2p.RouterOptions{
+			Endpoint:                 endpoint,
+			Connection:               conn.DefaultMConnConfig(),
+			IncomingConnectionWindow: utils.Some(time.Duration(0)),
+			MaxAcceptRate:            rate.Inf,
+			MaxDialRate:              rate.Inf,
+			Giga:                     utils.Some[p2p.GigaRouter](giga),
+		},
+	)
+	require.NoError(t, err)
+	return &Environment{
+		App:        app,
+		GenDoc:     genDoc,
+		Router:     router,
+		EventSinks: []indexer.EventSink{kvsink.NewEventSink(dbm.NewMemDB(), nil)},
+		Config:     *config.DefaultRPCConfig(),
+	}
+}