From 5cf9b6fe837513c548b795ea90fe5d21b187d461 Mon Sep 17 00:00:00 2001
From: Wen <crocoxu@gmail.com>
Date: Wed, 2 Sep 2026 14:11:21 -0700
Subject: [PATCH 1/6] Dial Autobahn peers from the live commit-epoch committee,
 not the static address book.

Inbound giga role is fixed at accept: a membership change closes the connection so the peer's dialer reconnects into the role it now has. Skip republishing an unchanged next-commit epoch so watchers do not thrash every QC.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 sei-tendermint/config/config.go               |   6 +-
 .../internal/autobahn/data/state.go           |   3 +
 .../internal/p2p/giga_router_common.go        | 195 +++++++++++++----
 .../internal/p2p/giga_router_common_test.go   | 203 ++++++++++++++++++
 .../internal/p2p/giga_router_fullnode.go      |  61 ++++--
 .../internal/p2p/giga_router_validator.go     |  38 ++--
 sei-tendermint/node/setup.go                  |  14 +-
 7 files changed, 434 insertions(+), 86 deletions(-)

diff --git a/sei-tendermint/config/config.go b/sei-tendermint/config/config.go
index 5dd177a0ed..6123018868 100644
--- a/sei-tendermint/config/config.go
+++ b/sei-tendermint/config/config.go
@@ -80,9 +80,9 @@ type Config struct {
 	// AutobahnConfigFile is the path to a JSON file containing the Autobahn (GigaRouter)
 	// configuration. Leave empty to disable Autobahn. The autobahn role
 	// follows the top-level `mode` field: "validator" runs the validator
-	// path; any other mode runs as a fullnode (loads the committee as a
-	// routing table and pulls blocks from committee members). A warning is
-	// logged at startup if mode disagrees with committee membership.
+	// path; any other mode runs as a fullnode (loads the address book and
+	// pulls blocks from committee members). A warning is logged at startup
+	// if mode disagrees with address-book membership.
 	AutobahnConfigFile string `mapstructure:"autobahn-config-file"`
 
 	// HashVaultDisabledUnsafe disables the app-hash equivocation guard (HashVault). The vault is
diff --git a/sei-tendermint/internal/autobahn/data/state.go b/sei-tendermint/internal/autobahn/data/state.go
index f61764fc72..b2744abd61 100644
--- a/sei-tendermint/internal/autobahn/data/state.go
+++ b/sei-tendermint/internal/autobahn/data/state.go
@@ -89,6 +89,9 @@ func (i *inner) publishNextCommitEpoch(registry *epoch.Registry) {
 	if err != nil {
 		return
 	}
+	if ep.EpochIndex() == i.nextCommitEpoch.Load().EpochIndex() {
+		return
+	}
 	i.nextCommitEpoch.Store(ep)
 }
 
diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go
index e1cbf00066..a597b6d4a6 100644
--- a/sei-tendermint/internal/p2p/giga_router_common.go
+++ b/sei-tendermint/internal/p2p/giga_router_common.go
@@ -46,9 +46,9 @@ type gigaRouterCommon struct {
 	// EvmProxy can Load() without taking the data lock on every call.
 	nextCommitEpoch utils.AtomicRecv[*atypes.Epoch]
 
-	// inboundFullnodeCount tracks live non-committee inbound block-sync
-	// connections. Optimistic Add(1) + compare against cap; over-rejects
-	// by one or two under contention but never over-accepts.
+	// inboundFullnodeCount tracks inbound connections currently served the
+	// block-sync subset. Optimistic Add(1) + compare against cap;
+	// over-rejects by one or two under contention but never over-accepts.
 	inboundFullnodeCount atomic.Int64
 	inboundFullnodeCap   int64
 }
@@ -257,37 +257,29 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo
 	return commitResp, nil
 }
 
-// manages lifecycle of evmrpc connections to validators.
-func (r *gigaRouterCommon) runEvmProxies(ctx context.Context) error {
-	return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
-		for validator, addr := range r.cfg.ValidatorAddrs {
-			s.SpawnNamed(addr.String(), func() error {
-				for {
-					client, err := ethrpc.DialContext(ctx, addr.EVMRPC.String())
-					if err != nil {
-						logger.Info("evm proxy dial failed", "url", addr.EVMRPC, "err", err)
-						if err := utils.Sleep(ctx, r.cfg.DialInterval); err != nil {
-							return err
-						}
-						continue
-					}
-
-					for proxies := range r.proxies.Lock() {
-						proxies[validator] = client
-					}
-					<-ctx.Done()
-					client.Close()
-					for proxies := range r.proxies.Lock() {
-						if proxies[validator] == client {
-							delete(proxies, validator)
-						}
-					}
-					return ctx.Err()
-				}
-			})
+// runEvmProxy maintains an EVM RPC client for one committee member.
+func (r *gigaRouterCommon) runEvmProxy(ctx context.Context, validator atypes.PublicKey, addr GigaNodeAddr) error {
+	for {
+		client, err := ethrpc.DialContext(ctx, addr.EVMRPC.String())
+		if err != nil {
+			logger.Info("evm proxy dial failed", "url", addr.EVMRPC, "err", err)
+			if err := utils.Sleep(ctx, r.cfg.DialInterval); err != nil {
+				return err
+			}
+			continue
 		}
-		return nil
-	})
+		for proxies := range r.proxies.Lock() {
+			proxies[validator] = client
+		}
+		<-ctx.Done()
+		client.Close()
+		for proxies := range r.proxies.Lock() {
+			if proxies[validator] == client {
+				delete(proxies, validator)
+			}
+		}
+		return ctx.Err()
+	}
 }
 
 // gas used, as reported by finalizeBlock() call.
@@ -525,26 +517,132 @@ func (r *gigaRouterCommon) dialAndRunConn(
 	})
 }
 
+// committeeMemberTask is work for one reachable committee member.
+type committeeMemberTask func(ctx context.Context, validator atypes.PublicKey, addr GigaNodeAddr) error
+
+// memberSession is a committee member's cancellable task session.
+type memberSession struct {
+	cancel context.CancelFunc
+	done   chan struct{}
+}
+
+// stopDepartingMembers stops sessions for validators outside committee.
+func stopDepartingMembers(ctx context.Context, live map[atypes.PublicKey]*memberSession, committee *atypes.Committee) error {
+	var departing []*memberSession
+	// Cancel every departing session before waiting for any of them.
+	for validator, session := range live {
+		if committee.HasReplica(validator) {
+			continue
+		}
+		session.cancel()
+		departing = append(departing, session)
+		delete(live, validator)
+	}
+	for _, session := range departing {
+		if _, _, err := utils.RecvOrClosed(ctx, session.done); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// runPerCommitteeMember runs tasks for each reachable member of the
+// committee covering the next CommitQC.
+func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...committeeMemberTask) error {
+	return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
+		live := map[atypes.PublicKey]*memberSession{}
+		// End all sessions before the scope waits for them.
+		defer func() {
+			for _, session := range live {
+				session.cancel()
+			}
+		}()
+		return r.nextCommitEpoch.Iter(ctx, func(_ context.Context, epoch *atypes.Epoch) error {
+			committee := epoch.Committee()
+			if err := stopDepartingMembers(ctx, live, committee); err != nil {
+				return err
+			}
+			for lane := range committee.Lanes().All() {
+				validator := lane.Validator
+				if _, ok := live[validator]; ok {
+					continue
+				}
+				addr, ok := r.cfg.ValidatorAddrs[validator]
+				if !ok {
+					logger.Error("committee member has no configured address; not dialing", "validator", validator)
+					continue
+				}
+				taskCtx, cancel := context.WithCancel(ctx)
+				done := make(chan struct{})
+				live[validator] = &memberSession{cancel: cancel, done: done}
+				s.SpawnNamed(addr.String(), func() error {
+					defer close(done)
+					return utils.IgnoreCancel(scope.Run(taskCtx, func(ctx context.Context, ms scope.Scope) error {
+						for _, task := range tasks {
+							ms.Spawn(func() error { return task(ctx, validator, addr) })
+						}
+						return nil
+					}))
+				})
+			}
+			return nil
+		})
+	})
+}
+
+// runUntilMembershipChange runs f while validator's membership matches
+// isCommittee. It reports whether a membership change ended f.
+func (r *gigaRouterCommon) runUntilMembershipChange(
+	ctx context.Context,
+	validator atypes.PublicKey,
+	isCommittee bool,
+	f func(ctx context.Context) error,
+) (changed bool, err error) {
+	err = scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
+		s.SpawnBg(func() error {
+			_, err := r.nextCommitEpoch.Wait(ctx, func(epoch *atypes.Epoch) bool {
+				return epoch.Committee().HasReplica(validator) != isCommittee
+			})
+			if err != nil {
+				return err
+			}
+			changed = true
+			s.Cancel(nil)
+			return nil
+		})
+		return f(ctx)
+	})
+	return changed, utils.IgnoreCancel(err)
+}
+
 // RunInboundConn serves an inbound giga connection. Non-committee peers
-// get the block-sync subset (StreamFullCommitQCs + GetBlock), capped at
-// inboundFullnodeCap. Committee peers get the full RunServer on
-// validators; on a fullnode the connection is refused (committee peers
-// shouldn't be dialing fullnodes — see Service.RunInbound).
+// get the block-sync subset (StreamFullCommitQCs + GetBlock). Committee peers
+// get the full RunServer on validators; on a fullnode the connection is refused.
+//
+// The role and the fullnode cap are fixed for the lifetime of the connection: a
+// membership change ends it, and the peer's dialer reconnects into the role it
+// now has.
 func (r *gigaRouterCommon) RunInboundConn(ctx context.Context, hConn *handshakedConn) error {
 	if !hConn.msg.SeiGigaConnection {
 		return fmt.Errorf("not a SeiGiga connection")
 	}
 	// Filter unwanded connections.
 	key := hConn.msg.NodeAuth.Key()
-	isCommittee := false
-	for _, addr := range r.cfg.ValidatorAddrs {
+	// TODO: support committee members absent from the address book.
+	validator := utils.None[atypes.PublicKey]()
+	for v, addr := range r.cfg.ValidatorAddrs {
 		if addr.Key == key {
-			isCommittee = true
+			validator = utils.Some(v)
 			break
 		}
 	}
+	isCommittee := false
+	if v, ok := validator.Get(); ok {
+		isCommittee = r.nextCommitEpoch.Load().Committee().HasReplica(v)
+	}
 	if !isCommittee {
-		// Optimistic acquire: Add(1), compare, Add(-1) on overflow.
+		// Optimistic acquire: Add(1), compare, Add(-1) on overflow. Acquired
+		// before InsertAndRun, which evicts any live connection for this key.
 		if r.inboundFullnodeCount.Add(1) > r.inboundFullnodeCap {
 			r.inboundFullnodeCount.Add(-1)
 			return fmt.Errorf("inbound fullnode peer limit (%d) reached", r.inboundFullnodeCap)
@@ -558,9 +656,22 @@ func (r *gigaRouterCommon) RunInboundConn(ctx context.Context, hConn *handshaked
 			Global.gigaNewConnsAt("in").Add(1)
 			Global.gigaConnsAt("in").Add(1)
 			defer Global.gigaConnsAt("in").Add(-1)
-			if err := r.service.RunServer(ctx, server, isCommittee); err != nil {
+			v, ok := validator.Get()
+			if !ok {
+				if err := r.service.RunServer(ctx, server, false); err != nil {
+					return fmt.Errorf("inbound from %v: %w", key, err)
+				}
+				return nil
+			}
+			changed, err := r.runUntilMembershipChange(ctx, v, isCommittee, func(ctx context.Context) error {
+				return r.service.RunServer(ctx, server, isCommittee)
+			})
+			if err != nil {
 				return fmt.Errorf("inbound from %v: %w", key, err)
 			}
+			if changed {
+				logger.Info("inbound giga peer changed committee membership; closing", "addr", key, "was_committee", isCommittee)
+			}
 			return nil
 		})
 	})
diff --git a/sei-tendermint/internal/p2p/giga_router_common_test.go b/sei-tendermint/internal/p2p/giga_router_common_test.go
index 191b3fbdd5..fb2a8f81d8 100644
--- a/sei-tendermint/internal/p2p/giga_router_common_test.go
+++ b/sei-tendermint/internal/p2p/giga_router_common_test.go
@@ -2,7 +2,9 @@ package p2p
 
 import (
 	"context"
+	"fmt"
 	"net/url"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -19,6 +21,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/libs/utils/scope"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/rpc/coretypes"
 	tmtypes "github.com/sei-protocol/sei-chain/sei-tendermint/types"
 )
@@ -245,6 +248,206 @@ func testGigaRouterWithData(t *testing.T, addrs map[atypes.PublicKey]GigaNodeAdd
 	}
 }
 
+func testEpoch(index atypes.EpochIndex, weights map[atypes.PublicKey]uint64) *atypes.Epoch {
+	committee := utils.OrPanic1(atypes.NewCommittee(weights))
+	return atypes.NewEpoch(index, atypes.RoadRange{}, time.Time{}, committee, 1)
+}
+
+func TestGigaRouterCommon_RunPerCommitteeMemberFollowsCommittee(t *testing.T) {
+	rng := utils.TestRng()
+	a := atypes.GenSecretKey(rng).Public()
+	b := atypes.GenSecretKey(rng).Public()
+	nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{b: 1}))
+	router := &gigaRouterCommon{
+		cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{
+			a: {Key: makeKey(rng).Public()},
+			b: {Key: makeKey(rng).Public()},
+		}},
+		nextCommitEpoch: nextEpoch.Subscribe(),
+	}
+
+	startedA := make(chan struct{}, 1)
+	startedB := make(chan struct{}, 1)
+	stoppedB := make(chan struct{}, 1)
+	err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error {
+		s.SpawnBg(func() error {
+			return utils.IgnoreCancel(router.runPerCommitteeMember(ctx, func(ctx context.Context, validator atypes.PublicKey, _ GigaNodeAddr) error {
+				switch {
+				case validator.Compare(a) == 0:
+					startedA <- struct{}{}
+				case validator.Compare(b) == 0:
+					startedB <- struct{}{}
+				}
+				<-ctx.Done()
+				if validator.Compare(b) == 0 {
+					stoppedB <- struct{}{}
+				}
+				return ctx.Err()
+			}))
+		})
+		<-startedB
+		select {
+		case <-startedA:
+			return fmt.Errorf("a started while outside the committee")
+		default:
+		}
+		nextEpoch.Store(testEpoch(3, map[atypes.PublicKey]uint64{a: 1, b: 1}))
+		<-startedA
+		nextEpoch.Store(testEpoch(4, map[atypes.PublicKey]uint64{a: 1}))
+		<-stoppedB
+		select {
+		case <-startedA:
+			return fmt.Errorf("a restarted while still a member")
+		default:
+		}
+		return nil
+	})
+	require.NoError(t, err)
+}
+
+func TestGigaRouterCommon_RunPerCommitteeMemberRunsOneSessionPerMember(t *testing.T) {
+	rng := utils.TestRng()
+	a := atypes.GenSecretKey(rng).Public()
+	// b has no address-book entry, so an epoch containing only b starts nothing.
+	b := atypes.GenSecretKey(rng).Public()
+	nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{a: 1}))
+	router := &gigaRouterCommon{
+		cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{
+			a: {Key: makeKey(rng).Public()},
+		}},
+		nextCommitEpoch: nextEpoch.Subscribe(),
+	}
+
+	started := make(chan struct{}, 2)
+	canceled := make(chan struct{}, 2)
+	release := make(chan struct{})
+	var live atomic.Int64
+	err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error {
+		s.SpawnBg(func() error {
+			return utils.IgnoreCancel(router.runPerCommitteeMember(ctx, func(ctx context.Context, _ atypes.PublicKey, _ GigaNodeAddr) error {
+				started <- struct{}{}
+				if n := live.Add(1); n > 1 {
+					return fmt.Errorf("%d concurrent sessions for the same member", n)
+				}
+				<-ctx.Done()
+				canceled <- struct{}{}
+				// Keep the first session live after cancellation.
+				<-release
+				live.Add(-1)
+				return ctx.Err()
+			}))
+		})
+		<-started
+		nextEpoch.Store(testEpoch(3, map[atypes.PublicKey]uint64{b: 1}))
+		<-canceled
+		nextEpoch.Store(testEpoch(4, map[atypes.PublicKey]uint64{a: 1}))
+		close(release)
+		<-started
+		return nil
+	})
+	require.NoError(t, err)
+}
+
+func TestGigaRouterCommon_RunPerCommitteeMemberCancelsAllDepartingBeforeAwait(t *testing.T) {
+	rng := utils.TestRng()
+	a := atypes.GenSecretKey(rng).Public()
+	b := atypes.GenSecretKey(rng).Public()
+	c := atypes.GenSecretKey(rng).Public()
+	nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{a: 1, b: 1}))
+	router := &gigaRouterCommon{
+		cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{
+			a: {Key: makeKey(rng).Public()},
+			b: {Key: makeKey(rng).Public()},
+		}},
+		nextCommitEpoch: nextEpoch.Subscribe(),
+	}
+
+	started := make(chan atypes.PublicKey, 2)
+	canceled := make(chan struct{}, 2)
+	release := make(chan struct{})
+	err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error {
+		s.SpawnBg(func() error {
+			return utils.IgnoreCancel(router.runPerCommitteeMember(ctx, func(ctx context.Context, validator atypes.PublicKey, _ GigaNodeAddr) error {
+				started <- validator
+				<-ctx.Done()
+				canceled <- struct{}{}
+				<-release
+				return ctx.Err()
+			}))
+		})
+		for range 2 {
+			<-started
+		}
+		nextEpoch.Store(testEpoch(3, map[atypes.PublicKey]uint64{c: 1}))
+		for range 2 {
+			<-canceled
+		}
+		close(release)
+		return nil
+	})
+	require.NoError(t, err)
+}
+
+func TestGigaRouterCommon_CommitteeTasksReturnWhenWorkReturns(t *testing.T) {
+	rng := utils.TestRng()
+	a := atypes.GenSecretKey(rng).Public()
+	nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{a: 1}))
+	router := &gigaRouterCommon{nextCommitEpoch: nextEpoch.Subscribe()}
+
+	changed, err := router.runUntilMembershipChange(t.Context(), a, true, func(context.Context) error {
+		return nil
+	})
+	require.NoError(t, err)
+	require.False(t, changed)
+}
+
+func TestGigaRouterCommon_RunUntilMembershipChangeCancelsFWhenMembershipChanges(t *testing.T) {
+	rng := utils.TestRng()
+	a := atypes.GenSecretKey(rng).Public()
+	b := atypes.GenSecretKey(rng).Public()
+	for _, tc := range []struct {
+		name        string
+		isCommittee bool
+	}{
+		{"leaves", true},
+		{"joins", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			members := map[atypes.PublicKey]uint64{b: 1}
+			if tc.isCommittee {
+				members[a] = 1
+			}
+			nextEpoch := utils.NewAtomicSend(testEpoch(2, members))
+			router := &gigaRouterCommon{nextCommitEpoch: nextEpoch.Subscribe()}
+			started := make(chan struct{})
+			err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error {
+				s.SpawnBg(func() error {
+					changed, err := router.runUntilMembershipChange(ctx, a, tc.isCommittee, func(ctx context.Context) error {
+						close(started)
+						<-ctx.Done()
+						return ctx.Err()
+					})
+					if err != nil {
+						return err
+					}
+					if !changed {
+						return fmt.Errorf("membership change must cancel f and report true")
+					}
+					return nil
+				})
+				<-started
+				flipped := map[atypes.PublicKey]uint64{b: 1}
+				if !tc.isCommittee {
+					flipped[a] = 1
+				}
+				nextEpoch.Store(testEpoch(3, flipped))
+				return nil
+			})
+			require.NoError(t, err)
+		})
+	}
+}
+
 func TestCommitteeWeights(t *testing.T) {
 	rng := utils.TestRng()
 	sk := ed25519.TestSecretKey(utils.GenBytes(rng, 32))
diff --git a/sei-tendermint/internal/p2p/giga_router_fullnode.go b/sei-tendermint/internal/p2p/giga_router_fullnode.go
index cdc7afd9e5..70b05d2a36 100644
--- a/sei-tendermint/internal/p2p/giga_router_fullnode.go
+++ b/sei-tendermint/internal/p2p/giga_router_fullnode.go
@@ -2,9 +2,7 @@ package p2p
 
 import (
 	"context"
-	"maps"
 	"math/rand/v2"
-	"slices"
 
 	"github.com/ethereum/go-ethereum/common"
 	ethrpc "github.com/ethereum/go-ethereum/rpc"
@@ -58,7 +56,9 @@ func (r *gigaFullnodeRouter) Run(ctx context.Context) error {
 		s.SpawnNamed("data", func() error { return r.data.Run(ctx) })
 		s.SpawnNamed("execute", func() error { return r.runExecute(ctx) })
 		s.SpawnNamed("service", func() error { return r.service.Run(ctx) })
-		s.SpawnNamed("evmProxies", func() error { return r.runEvmProxies(ctx) })
+		s.SpawnNamed("committeeMembers", func() error {
+			return r.runPerCommitteeMember(ctx, r.runEvmProxy)
+		})
 		return nil
 	})
 }
@@ -70,8 +70,10 @@ func (r *gigaFullnodeRouter) EvmProxy(sender common.Address) utils.Option[*ethrp
 }
 
 // runFullnodeSubscriber: pick a committee member, dial + block-sync,
-// advance on disconnect/reject. Committee list shuffled once at startup
-// so multiple fullnodes don't all converge on the same first choice.
+// advance on disconnect/reject. The inner ring is one shuffled pass of
+// the commit committee. After a disconnect, a committee change (not merely
+// an epoch tick) rebuilds the ring; an unchanged committee keeps walking
+// it. A live connection is not dropped just because the epoch advanced.
 //
 // TODO(autobahn-state-sync): block sync from a single peer is bounded by
 // GetBlock's per-stream rate limit (rpc.Limit{Rate:10, Concurrent:10}) —
@@ -80,18 +82,43 @@ func (r *gigaFullnodeRouter) EvmProxy(sender common.Address) utils.Option[*ethrp
 // sync). This loop is correct for "fresh cluster" and "restart of a
 // near-tip node."
 func (r *gigaFullnodeRouter) runFullnodeSubscriber(ctx context.Context) error {
-	validators := slices.Collect(maps.Keys(r.cfg.ValidatorAddrs))
-	rand.Shuffle(len(validators), func(i, j int) { validators[i], validators[j] = validators[j], validators[i] })
-	for i := 0; ; i = (i + 1) % len(validators) {
-		validator := validators[i]
-		addr := r.cfg.ValidatorAddrs[validator]
-		err := r.dialAndRunConn(ctx, utils.Some(addr.Key), addr.HostPort, func(ctx context.Context, client rpc.Client[giga.API]) error {
-			// Consensus PublicKey (map key), not GigaNodeAddr.Key (p2p NodePublicKey).
-			return r.service.RunClient(ctx, client, validator, true)
-		})
-		logger.Info("fullnode giga connection ended; failing over", "addr", addr, "err", err)
-		if err := utils.Sleep(ctx, r.cfg.DialInterval); err != nil {
-			return err
+	for {
+		ep := r.nextCommitEpoch.Load()
+		var validators []atypes.PublicKey
+		for lane := range ep.Committee().Lanes().All() {
+			if _, ok := r.cfg.ValidatorAddrs[lane.Validator]; ok {
+				validators = append(validators, lane.Validator)
+			}
+		}
+		if len(validators) == 0 {
+			if err := utils.Sleep(ctx, r.cfg.DialInterval); err != nil {
+				return err
+			}
+			continue
+		}
+		snap := ep.Committee()
+		// TODO: delay or stagger reshuffles so fullnodes rebalance slowly
+		// instead of all moving to a new ring at once.
+		rand.Shuffle(len(validators), func(i, j int) { validators[i], validators[j] = validators[j], validators[i] })
+		for _, validator := range validators {
+			if !r.nextCommitEpoch.Load().Committee().Equal(snap) {
+				break
+			}
+			addr := r.cfg.ValidatorAddrs[validator]
+			left, err := r.runUntilMembershipChange(ctx, validator, true, func(ctx context.Context) error {
+				return r.dialAndRunConn(ctx, utils.Some(addr.Key), addr.HostPort, func(ctx context.Context, client rpc.Client[giga.API]) error {
+					// Consensus PublicKey (committee member), not GigaNodeAddr.Key (p2p NodePublicKey).
+					return r.service.RunClient(ctx, client, validator, true)
+				})
+			})
+			if left {
+				logger.Info("fullnode giga peer left the committee; failing over", "addr", addr)
+			} else {
+				logger.Info("fullnode giga connection ended; failing over", "addr", addr, "err", err)
+			}
+			if err := utils.Sleep(ctx, r.cfg.DialInterval); err != nil {
+				return err
+			}
 		}
 	}
 }
diff --git a/sei-tendermint/internal/p2p/giga_router_validator.go b/sei-tendermint/internal/p2p/giga_router_validator.go
index 3476fbcf1b..9891360919 100644
--- a/sei-tendermint/internal/p2p/giga_router_validator.go
+++ b/sei-tendermint/internal/p2p/giga_router_validator.go
@@ -73,31 +73,32 @@ func (r *gigaValidatorRouter) Run(ctx context.Context) error {
 		// gap-fills keep retrying. Compare against the p2p node key
 		// (r.key.Public), not validatorKey (consensus signing key used by
 		// EvmProxy): GigaNodeAddr.Key is a NodePublicKey.
-		selfKey := r.key.Public()
-		for validatorKey, addr := range r.cfg.ValidatorAddrs {
-			getBlock := addr.Key != selfKey
-			s.Spawn(func() error {
-				for {
-					err := r.dialAndRunConn(ctx, utils.Some(addr.Key), addr.HostPort, func(ctx context.Context, client rpc.Client[giga.API]) error {
-						return r.service.RunClient(ctx, client, validatorKey, getBlock)
-					})
-					logger.Info("giga connection failed", "addr", addr, "err", err)
-					if err := utils.Sleep(ctx, r.cfg.DialInterval); err != nil {
-						return err
-					}
-				}
-			})
-		}
+		s.SpawnNamed("committeeMembers", func() error {
+			return r.runPerCommitteeMember(ctx, r.runCommitteePeer, r.runEvmProxy)
+		})
 		s.SpawnNamed("consensus", func() error { return r.consensus.Run(ctx) })
 		s.SpawnNamed("producer", func() error { return r.producer.Run(ctx) })
 		s.SpawnNamed("data", func() error { return r.data.Run(ctx) })
 		s.SpawnNamed("execute", func() error { return r.runExecute(ctx) })
 		s.SpawnNamed("service", func() error { return r.service.Run(ctx) })
-		s.SpawnNamed("evmProxies", func() error { return r.runEvmProxies(ctx) })
 		return nil
 	})
 }
 
+// runCommitteePeer maintains an outbound connection to a committee member.
+func (r *gigaValidatorRouter) runCommitteePeer(ctx context.Context, validatorKey atypes.PublicKey, addr GigaNodeAddr) error {
+	getBlock := addr.Key != r.key.Public()
+	for {
+		err := r.dialAndRunConn(ctx, utils.Some(addr.Key), addr.HostPort, func(ctx context.Context, client rpc.Client[giga.API]) error {
+			return r.service.RunClient(ctx, client, validatorKey, getBlock)
+		})
+		logger.Info("giga connection failed", "addr", addr, "err", err)
+		if err := utils.Sleep(ctx, r.cfg.DialInterval); err != nil {
+			return err
+		}
+	}
+}
+
 // EvmProxy on the validator returns None when the sender's shard owner is
 // us (handle locally via mempool). For remote
 // shards, we proxy only while the target validator is currently connected;
@@ -110,7 +111,10 @@ func (r *gigaValidatorRouter) EvmProxy(sender common.Address) utils.Option[*ethr
 	if r.validatorKey == validator {
 		return utils.None[*ethrpc.Client]()
 	}
-	target := r.cfg.ValidatorAddrs[validator]
+	target, ok := r.cfg.ValidatorAddrs[validator]
+	if !ok {
+		return utils.None[*ethrpc.Client]()
+	}
 	if _, ok := r.poolOut.Get(target.Key); !ok {
 		return utils.None[*ethrpc.Client]()
 	}
diff --git a/sei-tendermint/node/setup.go b/sei-tendermint/node/setup.go
index b21e42fe02..310ce54103 100644
--- a/sei-tendermint/node/setup.go
+++ b/sei-tendermint/node/setup.go
@@ -275,9 +275,9 @@ func buildValidatorGigaConfig(
 // buildGigaRouter picks validator-vs-fullnode by cfg.Mode:
 // "validator" runs the validator path, any other mode runs as a fullnode.
 // Mode is the operator's explicit role declaration, kept separate from
-// committee membership so a newly-joined committee member can finish
+// live committee membership so a newly-joined committee member can finish
 // catch-up as a fullnode before the operator flips to mode = "validator".
-// A warning is logged if mode and committee membership disagree so an
+// A warning is logged if mode and address-book membership disagree so an
 // operator misconfiguration is visible at startup.
 //
 // The returned BlockStore is owned by the caller (nodeImpl): open happens here
@@ -296,12 +296,12 @@ func buildGigaRouter(
 		return nil, nil, err
 	}
 	if valKey, ok := validatorKey.Get(); ok {
-		_, inCommittee := validatorAddrs[valKey.Public()]
+		_, inAddressBook := validatorAddrs[valKey.Public()]
 		switch {
-		case cfg.Mode == config.ModeValidator && !inCommittee:
-			logger.Warn("Autobahn: mode is \"validator\" but local validator key is not in the committee", "valKey", valKey.Public())
-		case cfg.Mode != config.ModeValidator && inCommittee:
-			logger.Warn("Autobahn: local validator key is in the committee but mode is not \"validator\"; starting as fullnode", "mode", cfg.Mode)
+		case cfg.Mode == config.ModeValidator && !inAddressBook:
+			logger.Warn("Autobahn: mode is \"validator\" but local validator key is not in the address book", "valKey", valKey.Public())
+		case cfg.Mode != config.ModeValidator && inAddressBook:
+			logger.Warn("Autobahn: local validator key is in the address book but mode is not \"validator\"; starting as fullnode", "mode", cfg.Mode)
 		}
 	}
 	if cfg.Mode == config.ModeValidator {

From 73f1afe4172dfd1cb64bbf7002a831f208a9b119 Mon Sep 17 00:00:00 2001
From: Wen <crocoxu@gmail.com>
Date: Wed, 2 Sep 2026 17:10:39 -0700
Subject: [PATCH 2/6] Close inbound giga on committee membership change instead
 of waiting for the peer.

Spawn the mux as a background task so the inbound scope can cancel it; document that committee-member tasks run until cancel.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 sei-tendermint/internal/p2p/giga_router_common.go |  8 ++++++--
 .../internal/p2p/giga_router_validator.go         | 15 ++++++---------
 2 files changed, 12 insertions(+), 11 deletions(-)

diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go
index a597b6d4a6..58a0dbe1c1 100644
--- a/sei-tendermint/internal/p2p/giga_router_common.go
+++ b/sei-tendermint/internal/p2p/giga_router_common.go
@@ -517,7 +517,9 @@ func (r *gigaRouterCommon) dialAndRunConn(
 	})
 }
 
-// committeeMemberTask is work for one reachable committee member.
+// committeeMemberTask is work for one reachable committee member. It must run
+// until ctx is cancelled; returning earlier leaves the member unmarked in live
+// and it is not restarted while it stays in the committee.
 type committeeMemberTask func(ctx context.Context, validator atypes.PublicKey, addr GigaNodeAddr) error
 
 // memberSession is a committee member's cancellable task session.
@@ -652,7 +654,9 @@ func (r *gigaRouterCommon) RunInboundConn(ctx context.Context, hConn *handshaked
 	server := rpc.NewServer[giga.API]()
 	return r.poolIn.InsertAndRun(ctx, key, server, func(ctx context.Context) error {
 		return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
-			s.Spawn(func() error { return server.Run(ctx, hConn.conn) })
+			// Background: a membership change must cancel the mux. Spawn would
+			// keep this scope alive until the peer closes the socket.
+			s.SpawnBg(func() error { return server.Run(ctx, hConn.conn) })
 			Global.gigaNewConnsAt("in").Add(1)
 			Global.gigaConnsAt("in").Add(1)
 			defer Global.gigaConnsAt("in").Add(-1)
diff --git a/sei-tendermint/internal/p2p/giga_router_validator.go b/sei-tendermint/internal/p2p/giga_router_validator.go
index 9891360919..19e86e9224 100644
--- a/sei-tendermint/internal/p2p/giga_router_validator.go
+++ b/sei-tendermint/internal/p2p/giga_router_validator.go
@@ -65,14 +65,6 @@ func (r *gigaValidatorRouter) Mempool() utils.Option[*producer.State] {
 
 func (r *gigaValidatorRouter) Run(ctx context.Context) error {
 	return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
-		// Validators dial every committee member in parallel — consensus
-		// voting needs fan-out, not stickiness. Same connections also
-		// serve block sync between committee peers. Self disables GetBlock:
-		// a loopback consumer always returns empty for missing catch-up
-		// heights and can starve the contiguous prefix while higher
-		// gap-fills keep retrying. Compare against the p2p node key
-		// (r.key.Public), not validatorKey (consensus signing key used by
-		// EvmProxy): GigaNodeAddr.Key is a NodePublicKey.
 		s.SpawnNamed("committeeMembers", func() error {
 			return r.runPerCommitteeMember(ctx, r.runCommitteePeer, r.runEvmProxy)
 		})
@@ -85,7 +77,12 @@ func (r *gigaValidatorRouter) Run(ctx context.Context) error {
 	})
 }
 
-// runCommitteePeer maintains an outbound connection to a committee member.
+// runCommitteePeer maintains an outbound giga connection to a committee member.
+// Self disables GetBlock: a loopback consumer always returns empty for missing
+// catch-up heights and can starve the contiguous prefix while higher gap-fills
+// keep retrying. Compare against the p2p node key (r.key.Public), not
+// validatorKey (consensus signing key used by EvmProxy): GigaNodeAddr.Key is a
+// NodePublicKey.
 func (r *gigaValidatorRouter) runCommitteePeer(ctx context.Context, validatorKey atypes.PublicKey, addr GigaNodeAddr) error {
 	getBlock := addr.Key != r.key.Public()
 	for {

From 087a28aebbf15ad5820cd8df58d2e4ac5bd79924 Mon Sep 17 00:00:00 2001
From: Wen <crocoxu@gmail.com>
Date: Thu, 3 Sep 2026 00:15:33 -0700
Subject: [PATCH 3/6] Keep Autobahn committee sessions until Anchor is within
 one epoch of commit.

Dropping N-1 members as soon as nextCommitEpoch advances cuts AppVote
paths that are still owed. Dial and prune the union of the commit and
Anchor committees, and only prune once Anchor is present and at most one
epoch behind.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../internal/p2p/giga_router_common.go        |  64 +++++--
 .../internal/p2p/giga_router_common_test.go   | 164 ++++++++++++++++--
 .../internal/p2p/giga_router_fullnode.go      |   1 +
 .../internal/p2p/giga_router_validator.go     |   1 +
 sei-tendermint/libs/utils/mutex.go            |  20 +++
 sei-tendermint/libs/utils/mutex_test.go       |  30 ++++
 6 files changed, 257 insertions(+), 23 deletions(-)

diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go
index 58a0dbe1c1..ab8e9e09b1 100644
--- a/sei-tendermint/internal/p2p/giga_router_common.go
+++ b/sei-tendermint/internal/p2p/giga_router_common.go
@@ -45,6 +45,9 @@ type gigaRouterCommon struct {
 	// nextCommitEpoch is data.NextCommitEpoch() cached at construction so
 	// EvmProxy can Load() without taking the data lock on every call.
 	nextCommitEpoch utils.AtomicRecv[*atypes.Epoch]
+	// anchor is data.Anchor() cached at construction: the AppQC/CommitQC covering
+	// the lowest row data.State still holds.
+	anchor utils.AtomicRecv[utils.Option[data.Anchor]]
 
 	// inboundFullnodeCount tracks inbound connections currently served the
 	// block-sync subset. Optimistic Add(1) + compare against cap;
@@ -528,12 +531,37 @@ type memberSession struct {
 	done   chan struct{}
 }
 
-// stopDepartingMembers stops sessions for validators outside committee.
-func stopDepartingMembers(ctx context.Context, live map[atypes.PublicKey]*memberSession, committee *atypes.Committee) error {
+// keepReplicas is commitEpoch's committee, plus the committee Anchor covers when present.
+func keepReplicas(anchor utils.Option[data.Anchor], commitEpoch *atypes.Epoch) map[atypes.PublicKey]struct{} {
+	keep := map[atypes.PublicKey]struct{}{}
+	for lane := range commitEpoch.Committee().Lanes().All() {
+		keep[lane.Validator] = struct{}{}
+	}
+	if a, ok := anchor.Get(); ok {
+		for lane := range a.Epoch.Committee().Lanes().All() {
+			keep[lane.Validator] = struct{}{}
+		}
+	}
+	return keep
+}
+
+// stopDepartingMembers stops sessions for validators outside keepReplicas once
+// Anchor is at most one epoch behind commitEpoch.
+func stopDepartingMembers(
+	ctx context.Context,
+	live map[atypes.PublicKey]*memberSession,
+	anchor utils.Option[data.Anchor],
+	commitEpoch *atypes.Epoch,
+) error {
+	a, ok := anchor.Get()
+	if !ok || commitEpoch.EpochIndex() > a.Epoch.EpochIndex()+1 {
+		return nil
+	}
+	keep := keepReplicas(anchor, commitEpoch)
 	var departing []*memberSession
 	// Cancel every departing session before waiting for any of them.
 	for validator, session := range live {
-		if committee.HasReplica(validator) {
+		if _, ok := keep[validator]; ok {
 			continue
 		}
 		session.cancel()
@@ -548,8 +576,7 @@ func stopDepartingMembers(ctx context.Context, live map[atypes.PublicKey]*member
 	return nil
 }
 
-// runPerCommitteeMember runs tasks for each reachable member of the
-// committee covering the next CommitQC.
+// runPerCommitteeMember runs tasks for each reachable member needed by the commit epoch or Anchor.
 func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...committeeMemberTask) error {
 	return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
 		live := map[atypes.PublicKey]*memberSession{}
@@ -559,13 +586,18 @@ func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...c
 				session.cancel()
 			}
 		}()
-		return r.nextCommitEpoch.Iter(ctx, func(_ context.Context, epoch *atypes.Epoch) error {
-			committee := epoch.Committee()
-			if err := stopDepartingMembers(ctx, live, committee); err != nil {
+		// Anchor is republished on every persist batch; only its epoch can
+		// change the keep set.
+		epochOf := func(opt utils.Option[data.Anchor]) utils.Option[atypes.EpochIndex] {
+			return utils.MapOpt(opt, func(a data.Anchor) atypes.EpochIndex { return a.Epoch.EpochIndex() })
+		}
+		for ctx.Err() == nil {
+			commitEpoch := r.nextCommitEpoch.Load()
+			anchor := r.anchor.Load()
+			if err := stopDepartingMembers(ctx, live, anchor, commitEpoch); err != nil {
 				return err
 			}
-			for lane := range committee.Lanes().All() {
-				validator := lane.Validator
+			for validator := range keepReplicas(anchor, commitEpoch) {
 				if _, ok := live[validator]; ok {
 					continue
 				}
@@ -587,8 +619,14 @@ func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...c
 					}))
 				})
 			}
-			return nil
-		})
+			if err := utils.WaitEither(ctx, r.nextCommitEpoch, r.anchor, func() bool {
+				return r.nextCommitEpoch.Load().EpochIndex() != commitEpoch.EpochIndex() ||
+					epochOf(r.anchor.Load()) != epochOf(anchor)
+			}); err != nil {
+				return err
+			}
+		}
+		return ctx.Err()
 	})
 }
 
@@ -620,6 +658,8 @@ func (r *gigaRouterCommon) runUntilMembershipChange(
 // RunInboundConn serves an inbound giga connection. Non-committee peers
 // get the block-sync subset (StreamFullCommitQCs + GetBlock). Committee peers
 // get the full RunServer on validators; on a fullnode the connection is refused.
+// Inbound role still follows nextCommitEpoch only: a departing peer is
+// downgraded even while outbound sessions keep it for AppVotes.
 //
 // The role and the fullnode cap are fixed for the lifetime of the connection: a
 // membership change ends it, and the peer's dialer reconnects into the role it
diff --git a/sei-tendermint/internal/p2p/giga_router_common_test.go b/sei-tendermint/internal/p2p/giga_router_common_test.go
index fb2a8f81d8..9752702fb8 100644
--- a/sei-tendermint/internal/p2p/giga_router_common_test.go
+++ b/sei-tendermint/internal/p2p/giga_router_common_test.go
@@ -253,17 +253,41 @@ func testEpoch(index atypes.EpochIndex, weights map[atypes.PublicKey]uint64) *at
 	return atypes.NewEpoch(index, atypes.RoadRange{}, time.Time{}, committee, 1)
 }
 
+func testAnchor(ep *atypes.Epoch) utils.Option[data.Anchor] {
+	return utils.Some(data.Anchor{Epoch: ep})
+}
+
+// settledEpochs drives both keep-set inputs to the same epoch. Tests of the
+// window between them send to nextCommitEpoch and anchor separately.
+type settledEpochs struct {
+	commitEpoch utils.AtomicSend[*atypes.Epoch]
+	anchor      utils.AtomicSend[utils.Option[data.Anchor]]
+}
+
+func newSettledEpochs(ep *atypes.Epoch) *settledEpochs {
+	return &settledEpochs{
+		commitEpoch: utils.NewAtomicSend(ep),
+		anchor:      utils.NewAtomicSend(testAnchor(ep)),
+	}
+}
+
+func (e *settledEpochs) store(ep *atypes.Epoch) {
+	e.commitEpoch.Store(ep)
+	e.anchor.Store(testAnchor(ep))
+}
+
 func TestGigaRouterCommon_RunPerCommitteeMemberFollowsCommittee(t *testing.T) {
 	rng := utils.TestRng()
 	a := atypes.GenSecretKey(rng).Public()
 	b := atypes.GenSecretKey(rng).Public()
-	nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{b: 1}))
+	epochs := newSettledEpochs(testEpoch(2, map[atypes.PublicKey]uint64{b: 1}))
 	router := &gigaRouterCommon{
 		cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{
 			a: {Key: makeKey(rng).Public()},
 			b: {Key: makeKey(rng).Public()},
 		}},
-		nextCommitEpoch: nextEpoch.Subscribe(),
+		nextCommitEpoch: epochs.commitEpoch.Subscribe(),
+		anchor:          epochs.anchor.Subscribe(),
 	}
 
 	startedA := make(chan struct{}, 1)
@@ -291,9 +315,9 @@ func TestGigaRouterCommon_RunPerCommitteeMemberFollowsCommittee(t *testing.T) {
 			return fmt.Errorf("a started while outside the committee")
 		default:
 		}
-		nextEpoch.Store(testEpoch(3, map[atypes.PublicKey]uint64{a: 1, b: 1}))
+		epochs.store(testEpoch(3, map[atypes.PublicKey]uint64{a: 1, b: 1}))
 		<-startedA
-		nextEpoch.Store(testEpoch(4, map[atypes.PublicKey]uint64{a: 1}))
+		epochs.store(testEpoch(4, map[atypes.PublicKey]uint64{a: 1}))
 		<-stoppedB
 		select {
 		case <-startedA:
@@ -305,17 +329,134 @@ func TestGigaRouterCommon_RunPerCommitteeMemberFollowsCommittee(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestGigaRouterCommon_RunPerCommitteeMemberKeepsLeaversWhileAnchorLags(t *testing.T) {
+	for _, tc := range []struct {
+		name           string
+		hasStartAnchor bool
+	}{
+		{"no AppQC yet", false},
+		{"AppQC still in the prior epoch", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			rng := utils.TestRng()
+			a := atypes.GenSecretKey(rng).Public()
+			b := atypes.GenSecretKey(rng).Public()
+			c := atypes.GenSecretKey(rng).Public()
+			ep1 := testEpoch(1, map[atypes.PublicKey]uint64{a: 1, b: 1})
+			ep2 := testEpoch(2, map[atypes.PublicKey]uint64{a: 1, c: 1})
+			startAnchor := utils.None[data.Anchor]()
+			if tc.hasStartAnchor {
+				startAnchor = testAnchor(ep1)
+			}
+			nextEpoch := utils.NewAtomicSend(ep1)
+			anchor := utils.NewAtomicSend(startAnchor)
+			router := &gigaRouterCommon{
+				cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{
+					a: {Key: makeKey(rng).Public()},
+					b: {Key: makeKey(rng).Public()},
+					c: {Key: makeKey(rng).Public()},
+				}},
+				nextCommitEpoch: nextEpoch.Subscribe(),
+				anchor:          anchor.Subscribe(),
+			}
+
+			started := make(chan atypes.PublicKey, 3)
+			stoppedB := make(chan struct{}, 1)
+			err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error {
+				s.SpawnBg(func() error {
+					return utils.IgnoreCancel(router.runPerCommitteeMember(ctx, func(ctx context.Context, validator atypes.PublicKey, _ GigaNodeAddr) error {
+						started <- validator
+						<-ctx.Done()
+						if validator.Compare(b) == 0 {
+							stoppedB <- struct{}{}
+						}
+						return ctx.Err()
+					}))
+				})
+				for range 2 {
+					<-started
+				}
+				nextEpoch.Store(ep2)
+				<-started // c joined, so the prune for ep2 already ran
+				select {
+				case <-stoppedB:
+					return fmt.Errorf("b stopped while Anchor was still catching up")
+				default:
+				}
+				anchor.Store(testAnchor(ep2))
+				<-stoppedB
+				return nil
+			})
+			require.NoError(t, err)
+		})
+	}
+}
+
+func TestGigaRouterCommon_RunPerCommitteeMemberDialsBothCommitteesUntilStable(t *testing.T) {
+	rng := utils.TestRng()
+	a := atypes.GenSecretKey(rng).Public()
+	b := atypes.GenSecretKey(rng).Public()
+	c := atypes.GenSecretKey(rng).Public()
+	d := atypes.GenSecretKey(rng).Public()
+	nextEpoch := utils.NewAtomicSend(testEpoch(4, map[atypes.PublicKey]uint64{a: 1, c: 1}))
+	anchor := utils.NewAtomicSend(testAnchor(testEpoch(1, map[atypes.PublicKey]uint64{a: 1, b: 1})))
+	lagAnchor := testEpoch(2, map[atypes.PublicKey]uint64{a: 1, c: 1, d: 1})
+	dropAnchor := testEpoch(3, map[atypes.PublicKey]uint64{a: 1, c: 1})
+	router := &gigaRouterCommon{
+		cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{
+			a: {Key: makeKey(rng).Public()},
+			b: {Key: makeKey(rng).Public()},
+			c: {Key: makeKey(rng).Public()},
+			d: {Key: makeKey(rng).Public()},
+		}},
+		nextCommitEpoch: nextEpoch.Subscribe(),
+		anchor:          anchor.Subscribe(),
+	}
+
+	started := make(chan atypes.PublicKey, 4)
+	stoppedB := make(chan struct{}, 1)
+	onStart := map[atypes.PublicKey]struct{}{}
+	err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error {
+		s.SpawnBg(func() error {
+			return utils.IgnoreCancel(router.runPerCommitteeMember(ctx, func(ctx context.Context, validator atypes.PublicKey, _ GigaNodeAddr) error {
+				started <- validator
+				<-ctx.Done()
+				if validator.Compare(b) == 0 {
+					stoppedB <- struct{}{}
+				}
+				return ctx.Err()
+			}))
+		})
+		for range 3 {
+			onStart[<-started] = struct{}{}
+		}
+		anchor.Store(testAnchor(lagAnchor))
+		<-started // d joined, so the prune for lagAnchor already ran
+		select {
+		case <-stoppedB:
+			return fmt.Errorf("b stopped while Anchor was still more than one epoch behind")
+		default:
+		}
+		anchor.Store(testAnchor(dropAnchor))
+		<-stoppedB
+		return nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, map[atypes.PublicKey]struct{}{a: {}, b: {}, c: {}}, onStart)
+}
+
 func TestGigaRouterCommon_RunPerCommitteeMemberRunsOneSessionPerMember(t *testing.T) {
 	rng := utils.TestRng()
 	a := atypes.GenSecretKey(rng).Public()
 	// b has no address-book entry, so an epoch containing only b starts nothing.
 	b := atypes.GenSecretKey(rng).Public()
-	nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{a: 1}))
+	epochs := newSettledEpochs(testEpoch(2, map[atypes.PublicKey]uint64{a: 1}))
 	router := &gigaRouterCommon{
 		cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{
 			a: {Key: makeKey(rng).Public()},
 		}},
-		nextCommitEpoch: nextEpoch.Subscribe(),
+		nextCommitEpoch: epochs.commitEpoch.Subscribe(),
+		anchor:          epochs.anchor.Subscribe(),
 	}
 
 	started := make(chan struct{}, 2)
@@ -338,9 +479,9 @@ func TestGigaRouterCommon_RunPerCommitteeMemberRunsOneSessionPerMember(t *testin
 			}))
 		})
 		<-started
-		nextEpoch.Store(testEpoch(3, map[atypes.PublicKey]uint64{b: 1}))
+		epochs.store(testEpoch(3, map[atypes.PublicKey]uint64{b: 1}))
 		<-canceled
-		nextEpoch.Store(testEpoch(4, map[atypes.PublicKey]uint64{a: 1}))
+		epochs.store(testEpoch(4, map[atypes.PublicKey]uint64{a: 1}))
 		close(release)
 		<-started
 		return nil
@@ -353,13 +494,14 @@ func TestGigaRouterCommon_RunPerCommitteeMemberCancelsAllDepartingBeforeAwait(t
 	a := atypes.GenSecretKey(rng).Public()
 	b := atypes.GenSecretKey(rng).Public()
 	c := atypes.GenSecretKey(rng).Public()
-	nextEpoch := utils.NewAtomicSend(testEpoch(2, map[atypes.PublicKey]uint64{a: 1, b: 1}))
+	epochs := newSettledEpochs(testEpoch(2, map[atypes.PublicKey]uint64{a: 1, b: 1}))
 	router := &gigaRouterCommon{
 		cfg: &GigaRouterCommonConfig{ValidatorAddrs: map[atypes.PublicKey]GigaNodeAddr{
 			a: {Key: makeKey(rng).Public()},
 			b: {Key: makeKey(rng).Public()},
 		}},
-		nextCommitEpoch: nextEpoch.Subscribe(),
+		nextCommitEpoch: epochs.commitEpoch.Subscribe(),
+		anchor:          epochs.anchor.Subscribe(),
 	}
 
 	started := make(chan atypes.PublicKey, 2)
@@ -378,7 +520,7 @@ func TestGigaRouterCommon_RunPerCommitteeMemberCancelsAllDepartingBeforeAwait(t
 		for range 2 {
 			<-started
 		}
-		nextEpoch.Store(testEpoch(3, map[atypes.PublicKey]uint64{c: 1}))
+		epochs.store(testEpoch(3, map[atypes.PublicKey]uint64{c: 1}))
 		for range 2 {
 			<-canceled
 		}
diff --git a/sei-tendermint/internal/p2p/giga_router_fullnode.go b/sei-tendermint/internal/p2p/giga_router_fullnode.go
index 70b05d2a36..c297806200 100644
--- a/sei-tendermint/internal/p2p/giga_router_fullnode.go
+++ b/sei-tendermint/internal/p2p/giga_router_fullnode.go
@@ -29,6 +29,7 @@ func NewGigaFullnodeRouter(cfg *GigaRouterCommonConfig, key NodeSecretKey, dataS
 			key:                key,
 			data:               dataState,
 			nextCommitEpoch:    dataState.NextCommitEpoch(),
+			anchor:             dataState.Anchor(),
 			service:            giga.NewFullNodeService(dataState),
 			poolIn:             giga.NewPool[NodePublicKey, rpc.Server[giga.API]](),
 			poolOut:            giga.NewPool[NodePublicKey, rpc.Client[giga.API]](),
diff --git a/sei-tendermint/internal/p2p/giga_router_validator.go b/sei-tendermint/internal/p2p/giga_router_validator.go
index 19e86e9224..b99cc84bf0 100644
--- a/sei-tendermint/internal/p2p/giga_router_validator.go
+++ b/sei-tendermint/internal/p2p/giga_router_validator.go
@@ -46,6 +46,7 @@ func NewGigaValidatorRouter(cfg *GigaValidatorConfig, key NodeSecretKey, dataSta
 			key:                key,
 			data:               dataState,
 			nextCommitEpoch:    dataState.NextCommitEpoch(),
+			anchor:             dataState.Anchor(),
 			service:            giga.NewService(consensusState),
 			poolIn:             giga.NewPool[NodePublicKey, rpc.Server[giga.API]](),
 			poolOut:            giga.NewPool[NodePublicKey, rpc.Client[giga.API]](),
diff --git a/sei-tendermint/libs/utils/mutex.go b/sei-tendermint/libs/utils/mutex.go
index ffd5952e85..2e8ee7c0b8 100644
--- a/sei-tendermint/libs/utils/mutex.go
+++ b/sei-tendermint/libs/utils/mutex.go
@@ -124,6 +124,26 @@ func (w *atomicWatch[T]) Wait(ctx context.Context, pred func(T) bool) (T, error)
 	}
 }
 
+// WaitEither waits for pred to hold, waking on updates to either watch.
+// pred takes no value and reads the watches itself, because a condition
+// spanning two of them cannot be expressed over one value.
+func WaitEither[A, B any](ctx context.Context, a AtomicRecv[A], b AtomicRecv[B], pred func() bool) error {
+	for {
+		// Both update channels are read before pred, so a Store landing between
+		// the two still wakes the select rather than being missed.
+		aUpdated, bUpdated := a.ptr.Load().updated, b.ptr.Load().updated
+		if pred() {
+			return nil
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-aUpdated:
+		case <-bUpdated:
+		}
+	}
+}
+
 // Iter executes sequentially the function f on each value of the atomic watch.
 // Context passed to f is canceled when the next value is available.
 // Exits when the returned error is different from nil and context.Canceled,
diff --git a/sei-tendermint/libs/utils/mutex_test.go b/sei-tendermint/libs/utils/mutex_test.go
index 73592fe0a1..eb7710e105 100644
--- a/sei-tendermint/libs/utils/mutex_test.go
+++ b/sei-tendermint/libs/utils/mutex_test.go
@@ -10,6 +10,36 @@ import (
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope"
 )
 
+func TestWaitEither(t *testing.T) {
+	ctx := t.Context()
+	nums := utils.NewAtomicSend(0)
+	strs := utils.NewAtomicSend("")
+	recvNums, recvStrs := nums.Subscribe(), strs.Subscribe()
+
+	require.NoError(t, utils.WaitEither(ctx, recvNums, recvStrs, func() bool { return true }))
+
+	for _, store := range []func(){
+		func() { nums.Store(1) },
+		func() { strs.Store("go") },
+	} {
+		require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
+			s.SpawnBg(func() error {
+				return utils.WaitEither(ctx, recvNums, recvStrs, func() bool {
+					return recvNums.Load() == 1 || recvStrs.Load() == "go"
+				})
+			})
+			store()
+			return nil
+		}))
+		nums.Store(0)
+		strs.Store("")
+	}
+
+	canceled, cancel := context.WithCancel(ctx)
+	cancel()
+	require.Equal(t, context.Canceled, utils.WaitEither(canceled, recvNums, recvStrs, func() bool { return false }))
+}
+
 func TestAtomicSend(t *testing.T) {
 	ctx := t.Context()
 	v := 5

From d82bc5db677e6b22b9e5c24e81aa9521c203c8c1 Mon Sep 17 00:00:00 2001
From: Wen <crocoxu@gmail.com>
Date: Thu, 3 Sep 2026 13:08:56 -0700
Subject: [PATCH 4/6] Document inbound AppVote direction and log an empty
 fullnode address book.

Inbound role stays nextCommitEpoch-only because votes are collected on the
outbound client stream. A fullnode with no committee member in ValidatorAddrs
now logs instead of sleeping silently.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 sei-tendermint/internal/p2p/giga_router_common.go   | 5 +++--
 sei-tendermint/internal/p2p/giga_router_fullnode.go | 1 +
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go
index ab8e9e09b1..02d5c7196a 100644
--- a/sei-tendermint/internal/p2p/giga_router_common.go
+++ b/sei-tendermint/internal/p2p/giga_router_common.go
@@ -658,8 +658,6 @@ func (r *gigaRouterCommon) runUntilMembershipChange(
 // RunInboundConn serves an inbound giga connection. Non-committee peers
 // get the block-sync subset (StreamFullCommitQCs + GetBlock). Committee peers
 // get the full RunServer on validators; on a fullnode the connection is refused.
-// Inbound role still follows nextCommitEpoch only: a departing peer is
-// downgraded even while outbound sessions keep it for AppVotes.
 //
 // The role and the fullnode cap are fixed for the lifetime of the connection: a
 // membership change ends it, and the peer's dialer reconnects into the role it
@@ -678,6 +676,9 @@ func (r *gigaRouterCommon) RunInboundConn(ctx context.Context, hConn *handshaked
 			break
 		}
 	}
+	// Inbound role follows nextCommitEpoch only. AppVotes are received on the
+	// outbound client stream, not this mux, so a departing peer can be
+	// downgraded here while outbound sessions still collect its votes.
 	isCommittee := false
 	if v, ok := validator.Get(); ok {
 		isCommittee = r.nextCommitEpoch.Load().Committee().HasReplica(v)
diff --git a/sei-tendermint/internal/p2p/giga_router_fullnode.go b/sei-tendermint/internal/p2p/giga_router_fullnode.go
index c297806200..23d0407b47 100644
--- a/sei-tendermint/internal/p2p/giga_router_fullnode.go
+++ b/sei-tendermint/internal/p2p/giga_router_fullnode.go
@@ -92,6 +92,7 @@ func (r *gigaFullnodeRouter) runFullnodeSubscriber(ctx context.Context) error {
 			}
 		}
 		if len(validators) == 0 {
+			logger.Error("no commit-committee member in the address book; not dialing", "epoch", ep.EpochIndex())
 			if err := utils.Sleep(ctx, r.cfg.DialInterval); err != nil {
 				return err
 			}

From 13ca03c1dd3be46833595b9342d1cc319d90ecf7 Mon Sep 17 00:00:00 2001
From: Wen <crocoxu@gmail.com>
Date: Thu, 3 Sep 2026 14:23:28 -0700
Subject: [PATCH 5/6] Replace WaitEither with WaitAny over any number of
 AtomicRecv watches.

The keep-set loop will need a third watch; a two-type helper cannot grow.
WaitAny loads every update channel before pred, then reflect.Select, so a
Store cannot be missed regardless of how many watches are supplied.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../internal/p2p/giga_router_common.go        |  4 +--
 sei-tendermint/libs/utils/mutex.go            | 32 ++++++++++++-------
 sei-tendermint/libs/utils/mutex_test.go       | 17 ++++++----
 3 files changed, 33 insertions(+), 20 deletions(-)

diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go
index 02d5c7196a..56b04a87f9 100644
--- a/sei-tendermint/internal/p2p/giga_router_common.go
+++ b/sei-tendermint/internal/p2p/giga_router_common.go
@@ -619,10 +619,10 @@ func (r *gigaRouterCommon) runPerCommitteeMember(ctx context.Context, tasks ...c
 					}))
 				})
 			}
-			if err := utils.WaitEither(ctx, r.nextCommitEpoch, r.anchor, func() bool {
+			if err := utils.WaitAny(ctx, func() bool {
 				return r.nextCommitEpoch.Load().EpochIndex() != commitEpoch.EpochIndex() ||
 					epochOf(r.anchor.Load()) != epochOf(anchor)
-			}); err != nil {
+			}, r.nextCommitEpoch, r.anchor); err != nil {
 				return err
 			}
 		}
diff --git a/sei-tendermint/libs/utils/mutex.go b/sei-tendermint/libs/utils/mutex.go
index 2e8ee7c0b8..e7e991208a 100644
--- a/sei-tendermint/libs/utils/mutex.go
+++ b/sei-tendermint/libs/utils/mutex.go
@@ -3,6 +3,7 @@ package utils
 import (
 	"context"
 	"iter"
+	"reflect"
 	"sync"
 	"sync/atomic"
 
@@ -124,26 +125,35 @@ func (w *atomicWatch[T]) Wait(ctx context.Context, pred func(T) bool) (T, error)
 	}
 }
 
-// WaitEither waits for pred to hold, waking on updates to either watch.
-// pred takes no value and reads the watches itself, because a condition
-// spanning two of them cannot be expressed over one value.
-func WaitEither[A, B any](ctx context.Context, a AtomicRecv[A], b AtomicRecv[B], pred func() bool) error {
+// WaitAny waits for pred to hold, waking on updates to any supplied watch.
+func WaitAny(ctx context.Context, pred func() bool, watches ...AtomicUpdate) error {
 	for {
-		// Both update channels are read before pred, so a Store landing between
-		// the two still wakes the select rather than being missed.
-		aUpdated, bUpdated := a.ptr.Load().updated, b.ptr.Load().updated
+		// Every update channel is read before pred, so a Store landing between
+		// these reads and pred still wakes the select rather than being missed.
+		cases := make([]reflect.SelectCase, 1, len(watches)+1)
+		cases[0] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ctx.Done())}
+		for _, watch := range watches {
+			cases = append(cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(watch.updated())})
+		}
 		if pred() {
 			return nil
 		}
-		select {
-		case <-ctx.Done():
+		chosen, _, _ := reflect.Select(cases)
+		if chosen == 0 {
 			return ctx.Err()
-		case <-aUpdated:
-		case <-bUpdated:
 		}
 	}
 }
 
+// AtomicUpdate is implemented by AtomicRecv of any T.
+type AtomicUpdate interface {
+	updated() <-chan struct{}
+}
+
+func (w *atomicWatch[T]) updated() <-chan struct{} {
+	return w.ptr.Load().updated
+}
+
 // Iter executes sequentially the function f on each value of the atomic watch.
 // Context passed to f is canceled when the next value is available.
 // Exits when the returned error is different from nil and context.Canceled,
diff --git a/sei-tendermint/libs/utils/mutex_test.go b/sei-tendermint/libs/utils/mutex_test.go
index eb7710e105..183b539e67 100644
--- a/sei-tendermint/libs/utils/mutex_test.go
+++ b/sei-tendermint/libs/utils/mutex_test.go
@@ -10,34 +10,37 @@ import (
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope"
 )
 
-func TestWaitEither(t *testing.T) {
+func TestWaitAny(t *testing.T) {
 	ctx := t.Context()
 	nums := utils.NewAtomicSend(0)
 	strs := utils.NewAtomicSend("")
-	recvNums, recvStrs := nums.Subscribe(), strs.Subscribe()
+	flags := utils.NewAtomicSend(false)
+	recvNums, recvStrs, recvFlags := nums.Subscribe(), strs.Subscribe(), flags.Subscribe()
 
-	require.NoError(t, utils.WaitEither(ctx, recvNums, recvStrs, func() bool { return true }))
+	require.NoError(t, utils.WaitAny(ctx, func() bool { return true }, recvNums, recvStrs, recvFlags))
 
 	for _, store := range []func(){
 		func() { nums.Store(1) },
 		func() { strs.Store("go") },
+		func() { flags.Store(true) },
 	} {
 		require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
 			s.SpawnBg(func() error {
-				return utils.WaitEither(ctx, recvNums, recvStrs, func() bool {
-					return recvNums.Load() == 1 || recvStrs.Load() == "go"
-				})
+				return utils.WaitAny(ctx, func() bool {
+					return recvNums.Load() == 1 || recvStrs.Load() == "go" || recvFlags.Load()
+				}, recvNums, recvStrs, recvFlags)
 			})
 			store()
 			return nil
 		}))
 		nums.Store(0)
 		strs.Store("")
+		flags.Store(false)
 	}
 
 	canceled, cancel := context.WithCancel(ctx)
 	cancel()
-	require.Equal(t, context.Canceled, utils.WaitEither(canceled, recvNums, recvStrs, func() bool { return false }))
+	require.Equal(t, context.Canceled, utils.WaitAny(canceled, func() bool { return false }, recvNums, recvStrs, recvFlags))
 }
 
 func TestAtomicSend(t *testing.T) {

From 06942880ba43d3787b35fb8644fe3bde11ec4df9 Mon Sep 17 00:00:00 2001
From: Wen <crocoxu@gmail.com>
Date: Fri, 4 Sep 2026 09:01:50 -0700
Subject: [PATCH 6/6] Document why WaitAny rebuilds select cases on each Store.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 sei-tendermint/libs/utils/mutex.go | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/sei-tendermint/libs/utils/mutex.go b/sei-tendermint/libs/utils/mutex.go
index e7e991208a..4f856d781c 100644
--- a/sei-tendermint/libs/utils/mutex.go
+++ b/sei-tendermint/libs/utils/mutex.go
@@ -97,7 +97,7 @@ func NewAtomicSend[T any](value T) (w AtomicSend[T]) {
 	return
 }
 
-// Store updates the value of the atomic watch.
+// Store publishes a new version and wakes waiters of the previous one.
 func (w *AtomicSend[T]) Store(value T) {
 	close(w.ptr.Swap(newVersion(value)).updated)
 }
@@ -128,8 +128,10 @@ func (w *atomicWatch[T]) Wait(ctx context.Context, pred func(T) bool) (T, error)
 // WaitAny waits for pred to hold, waking on updates to any supplied watch.
 func WaitAny(ctx context.Context, pred func() bool, watches ...AtomicUpdate) error {
 	for {
-		// Every update channel is read before pred, so a Store landing between
-		// these reads and pred still wakes the select rather than being missed.
+		// Store swaps in a new version and closes the previous updated
+		// channel, so these cases cannot be reused across iterations.
+		// They are also loaded before pred so a Store between the load and
+		// pred still wakes the select rather than being missed.
 		cases := make([]reflect.SelectCase, 1, len(watches)+1)
 		cases[0] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ctx.Done())}
 		for _, watch := range watches {