test(integration): validate a real state-sync round-trip + SDK state-sync node support

sdk/sei/provider/k8s/render.go

@@ -111,6 +111,14 @@ func renderNode(spec sei.NodeSpec, namespace, networkNS string) *seiv1alpha1.Sei
 	if len(spec.Config) > 0 {
 		node.Spec.Overrides = maps.Clone(spec.Config)
 	}
+	if ss := spec.StateSync; ss != nil {
+		// Bootstrap via CometBFT state sync: the witnesses are the caller's bare
+		// host:port RpcServers; genesis block-sync leaves Snapshot nil (unchanged).
+		node.Spec.FullNode.Snapshot = &seiv1alpha1.SnapshotSource{
+			StateSync:  &seiv1alpha1.StateSyncSource{},
+			RpcServers: ss.RpcServers,
+		}
+	}
 	return node
 }
 

sdk/sei/provider/k8s/render_test.go

@@ -1,6 +1,7 @@
 package k8s
 
 import (
+	"slices"
 	"testing"
 
 	"github.com/sei-protocol/sei-k8s-controller/sdk/sei"
@@ -126,6 +127,31 @@ func TestRenderNode_CallerLabelsMergeUnderCanonical(t *testing.T) {
 	}
 }
 
+func TestRenderNode_StateSync(t *testing.T) {
+	witnesses := []string{"validator:26657", "validator:26657"}
+	spec := sei.NodeSpec{
+		Name: rpc0Name, Network: testNet, Image: testImage,
+		StateSync: &sei.NodeStateSync{RpcServers: witnesses},
+	}
+	node := renderNode(spec, testNS, testNS)
+
+	// StateSync maps to spec.fullNode.snapshot with the stateSync variant and the
+	// caller's bare host:port witnesses.
+	snap := node.Spec.FullNode.Snapshot
+	if snap == nil || snap.StateSync == nil {
+		t.Fatalf("state-sync node must render fullNode.snapshot.stateSync: %+v", node.Spec.FullNode)
+	}
+	if !slices.Equal(snap.RpcServers, witnesses) {
+		t.Errorf("snapshot.rpcServers = %v, want %v", snap.RpcServers, witnesses)
+	}
+
+	// Genesis block-sync (nil StateSync) leaves Snapshot unset — the unchanged path.
+	genesis := renderNode(sei.NodeSpec{Name: rpc0Name, Network: testNet, Image: testImage}, testNS, testNS)
+	if genesis.Spec.FullNode.Snapshot != nil {
+		t.Errorf("genesis node must leave snapshot nil, got %+v", genesis.Spec.FullNode.Snapshot)
+	}
+}
+
 func TestRenderNode_ConfigIntoOverrides(t *testing.T) {
 	spec := sei.NodeSpec{
 		Name: rpc0Name, Network: testNet, Image: testImage,

sdk/sei/readiness.go

@@ -34,8 +34,9 @@ type tendermintStatus struct {
 }
 
 type syncInfo struct {
-	LatestBlockHeight string `json:"latest_block_height"`
-	CatchingUp        bool   `json:"catching_up"`
+	LatestBlockHeight   string `json:"latest_block_height"`
+	EarliestBlockHeight string `json:"earliest_block_height"`
+	CatchingUp          bool   `json:"catching_up"`
 }
 
 func (s *tendermintStatus) sync() syncInfo {
@@ -92,6 +93,28 @@ func latestHeight(ctx context.Context, hc *http.Client, tmRPC string) (int64, bo
 	return h, true
 }
 
+// EarliestHeight reads tmRPC's earliest retained block height from /status.
+// ok=false on an unreachable endpoint or unparseable body. A state-synced node
+// reports an earliest > 1 (it started from a snapshot); a genesis-replayed node
+// reports 1 — so this is the point read that distinguishes the two. It accepts
+// both the enveloped and unwrapped /status shapes the Sei fork emits. Point read
+// only; there is no Wait wrapper.
+func EarliestHeight(ctx context.Context, hc *http.Client, tmRPC string) (int64, bool) {
+	body, ok := getJSON(ctx, hc, http.MethodGet, tmRPC+"/status", "")
+	if !ok {
+		return 0, false
+	}
+	var s tendermintStatus
+	if json.Unmarshal(body, &s) != nil {
+		return 0, false
+	}
+	h, err := strconv.ParseInt(s.sync().EarliestBlockHeight, 10, 64)
+	if err != nil {
+		return 0, false
+	}
+	return h, true
+}
+
 // WaitHeightAdvances blocks until tmRPC's committed height has risen by at least
 // delta from the height observed on the first successful read — proof the chain
 // is producing blocks, not merely reachable. Unlike WaitCaughtUp this catches a

sdk/sei/sei.go

@@ -218,6 +218,8 @@ func validateNodeSpec(s NodeSpec) error {
 		return usageErr("NodeSpec.Network is required (the peer-wire target)")
 	case strings.TrimSpace(s.Image) == "":
 		return usageErr("NodeSpec.Image is required")
+	case s.StateSync != nil && len(s.StateSync.RpcServers) < 2:
+		return usageErr("NodeSpec.StateSync.RpcServers must have >= 2 entries (the state-sync witness floor), got %d", len(s.StateSync.RpcServers))
 	}
 	return nil
 }

sdk/sei/spec.go

@@ -51,4 +51,20 @@ type NodeSpec struct {
 	Image            string            // -> spec.image
 	Config           map[string]string // -> spec.overrides (TOML-path keys)
 	Labels           map[string]string // extra labels on the SeiNode object, merged UNDER the canonical sei.io/role + sei.io/seinetwork (which win on key conflict)
+
+	// StateSync bootstraps the node through CometBFT state sync instead of
+	// genesis block-sync; nil => genesis block-sync (unchanged). Set it to bring
+	// a follower up from a peer-served snapshot rather than replaying from height
+	// 1. -> spec.fullNode.snapshot{stateSync{}, rpcServers}.
+	StateSync *NodeStateSync
+}
+
+// NodeStateSync configures a SeiNode to bootstrap via CometBFT state sync.
+type NodeStateSync struct {
+	// RpcServers are the light-client witnesses used for trust-point acquisition
+	// and verification. Entries are BARE host:port (no scheme) and there must be
+	// >= 2 (the CRD's fail-closed floor; CometBFT dedups identical entries).
+	// Network.TendermintRPC() returns a URL, so a caller deriving a witness from
+	// it must url.Parse(...).Host to strip the scheme. -> spec.fullNode.snapshot.rpcServers.
+	RpcServers []string
 }

sdk/sei/spec_test.go

@@ -36,6 +36,8 @@
 		{"missing name", func(s *NodeSpec) { s.Name = "" }, true},
 		{"missing network", func(s *NodeSpec) { s.Network = "" }, true},
 		{"missing image", func(s *NodeSpec) { s.Image = "" }, true},
+		{"statesync one witness", func(s *NodeSpec) { s.StateSync = &NodeStateSync{RpcServers: []string{"a:26657"}} }, true},
+		{"statesync two witnesses", func(s *NodeSpec) { s.StateSync = &NodeStateSync{RpcServers: []string{"a:26657", "b:26657"}} }, false},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {

test/integration/workflow_test.go

@@ -5,7 +5,9 @@ package integration
 import (
 	"context"
 	"net/http"
+	"net/url"
 	"os/signal"
+	"strconv"
 	"syscall"
 	"testing"
 	"time"
@@ -32,13 +34,16 @@ var snapshotProductionConfig = map[string]string{
 	"storage.snapshot_keep_recent": "3",
 }
 
-// TestWorkflowStateSync drives the SeiNodeTaskWorkflow StateSync recipe through
-// the SDK end to end: provision a chain + one caught-up RPC follower, then
-// CreateWorkflow(StateSync) against the follower and WaitWorkflowTerminal. The
-// recipe holds seid, wipes the follower's data directory, re-configures
-// state-sync against the witnesses, and releases — so a Complete workflow with
-// the follower caught up AGAIN is the re-bootstrap observable (wiped-then-synced,
-// no manual data-directory surgery). Acceptance criterion 5.
+// TestWorkflowStateSync drives the full state-sync procedure through the SDK end
+// to end: provision a snapshot-producing chain, wait past the snapshot interval,
+// then bring up a state-sync-bootstrapped follower and assert it started FROM a
+// snapshot (earliest retained height > 1, not a genesis replay's 1). It then runs
+// the SeiNodeTaskWorkflow StateSync recipe against THAT follower — the recipe
+// holds seid, wipes the data directory, re-configures state-sync against the
+// witnesses, and releases — and asserts a Complete workflow with the follower
+// caught up AGAIN and its earliest height advanced past the pre-wipe floor: proof
+// of a genuine wipe + fresh re-sync to a newer snapshot, not a no-op. Acceptance
+// criterion 5.
 //
 // Witnesses: the recipe's RpcServers are the network's aggregate TM RPC (the
 // validator service), passed as the >=2 the controller's fail-closed floor
@@ -80,14 +85,75 @@ func TestWorkflowStateSync(t *testing.T) {
 	if err != nil {
 		t.Fatalf("provision: %v", err)
 	}
-	node := ch.rpcNodes[0]
 	hc := &http.Client{Timeout: 10 * time.Second}
 
-	// Re-bootstrap the follower through the StateSync recipe.
+	// A follower can only state-sync from a snapshot that already exists. The
+	// witness validators emit one every storage.snapshot_interval blocks
+	// (snapshotProductionConfig); advance the aggregate RPC one interval + a
+	// margin to guarantee at least one snapshot boundary is crossed before we
+	// bootstrap from it. Derive the delta from the config so retuning the
+	// interval can't silently under-wait and make the state-sync assertion flaky.
+	interval, err := strconv.Atoi(snapshotProductionConfig["storage.snapshot_interval"])
+	if err != nil {
+		t.Fatalf("snapshot_interval not an int: %v", err)
+	}
+	if err := sei.WaitHeightAdvances(ctx, hc, ch.network.TendermintRPC(), int64(interval)+10); err != nil {
+		t.Fatalf("network did not advance past the snapshot interval: %v", err)
+	}
+
+	// Witnesses are the aggregate validator TM RPC as BARE host:port (the CRD
+	// SnapshotSource.RpcServers shape); TendermintRPC() is a URL, so strip the
+	// scheme. One aggregate witness passed twice satisfies the >=2 floor —
+	// CometBFT dedups identical entries; a distinct N-witness topology is a later
+	// refinement.
+	u, err := url.Parse(ch.network.TendermintRPC())
+	if err != nil {
+		t.Fatalf("parse network TM RPC %q: %v", ch.network.TendermintRPC(), err)
+	}
+	witness := u.Host
+
+	// Bring up a state-sync-bootstrapped follower: it must fetch a snapshot from a
+	// peer rather than replay from genesis. Appended to ch.rpcNodes so cleanupChain
+	// (registered above, evaluated at teardown) reaps it too.
+	ssNode, err := c.CreateNode(ctx, sei.NodeSpec{
+		Name:      "statesync-" + chainID,
+		Network:   chainID,
+		Namespace: ns,
+		Image:     seid,
+		Labels:    map[string]string{runLabelKey: chainID},
+		StateSync: &sei.NodeStateSync{RpcServers: []string{witness, witness}},
+	})
+	if ssNode != nil {
+		ch.rpcNodes = append(ch.rpcNodes, ssNode)
+	}
+	if err != nil {
+		t.Fatalf("create state-sync node: %v", err)
+	}
+	if err := ssNode.WaitReady(ctx); err != nil {
+		t.Fatalf("state-sync node %q running: %v", ssNode.Name(), err)
+	}
+	if err := sei.WaitCaughtUp(ctx, hc, ssNode.TendermintRPC()); err != nil {
+		t.Fatalf("state-sync node %q caught up: %v", ssNode.Name(), err)
+	}
+
+	// A state-synced node starts from a snapshot, so its earliest retained height
+	// is > 1; a genesis replay would report 1. This is the proof the bootstrap
+	// took the state-sync path, and the pre-wipe floor the post-workflow earliest
+	// must exceed.
+	preEarliest, ok := sei.EarliestHeight(ctx, hc, ssNode.TendermintRPC())
+	if !ok {
+		t.Fatalf("read earliest height of %q", ssNode.Name())
+	}
+	if preEarliest <= 1 {
+		t.Fatalf("state-sync node %q earliest height = %d, want > 1 (a genesis replay reports 1)", ssNode.Name(), preEarliest)
+	}
+	t.Logf("state-sync node %s: bootstrapped from snapshot (earliest height %d)", ssNode.Name(), preEarliest)
+
+	// Re-bootstrap that state-sync follower through the StateSync recipe.
 	wf, err := c.CreateWorkflow(ctx, sei.WorkflowSpec{
 		Name:      "resync-" + chainID,
 		Namespace: ns,
-		Node:      node.Name(),
+		Node:      ssNode.Name(),
 		Kind:      sei.WorkflowStateSync,
 		Labels:    map[string]string{runLabelKey: chainID},
 		StateSync: &sei.StateSyncWorkflow{
@@ -98,7 +164,7 @@ func TestWorkflowStateSync(t *testing.T) {
 		t.Fatalf("create workflow: %v", err)
 	}
 	cleanupWorkflow(t, wf)
-	t.Logf("workflow %s: created against %s", wf.Name(), node.Name())
+	t.Logf("workflow %s: created against %s", wf.Name(), ssNode.Name())
 
 	if err := wf.WaitTerminal(ctx); err != nil {
 		t.Fatalf("workflow %s did not complete: %v", wf.Name(), err)
@@ -111,13 +177,24 @@ func TestWorkflowStateSync(t *testing.T) {
 	// Re-bootstrap observable: after the wipe + resync the follower rejoins and
 	// catches up again (it cannot catch up to a chain it is not synced with, so
 	// this covers the wiped-then-synced round trip) and serves EVM again.
-	if err := sei.WaitCaughtUp(ctx, hc, node.TendermintRPC()); err != nil {
-		t.Fatalf("follower %s not caught up after resync: %v", node.Name(), err)
+	if err := sei.WaitCaughtUp(ctx, hc, ssNode.TendermintRPC()); err != nil {
+		t.Fatalf("follower %s not caught up after resync: %v", ssNode.Name(), err)
+	}
+	if err := sei.WaitEVMServing(ctx, hc, ssNode.EVMRPC()); err != nil {
+		t.Errorf("follower %s EVM not serving after resync: %v", ssNode.Name(), err)
+	}
+
+	// A genuine wipe + fresh re-sync lands on a NEW (higher) snapshot, so the
+	// earliest retained height must rise past the pre-workflow floor; a no-op
+	// recipe would leave earliest unchanged.
+	postEarliest, ok := sei.EarliestHeight(ctx, hc, ssNode.TendermintRPC())
+	if !ok {
+		t.Fatalf("read post-workflow earliest height of %q", ssNode.Name())
 	}
-	if err := sei.WaitEVMServing(ctx, hc, node.EVMRPC()); err != nil {
-		t.Errorf("follower %s EVM not serving after resync: %v", node.Name(), err)
+	if postEarliest <= preEarliest {
+		t.Fatalf("follower %s earliest height = %d after resync, want > %d (a genuine wipe + re-sync lands on a newer snapshot)", ssNode.Name(), postEarliest, preEarliest)
 	}
-	t.Logf("follower %s: caught up + EVM serving after state-sync re-bootstrap — TestWorkflowStateSync OK", node.Name())
+	t.Logf("follower %s: caught up + EVM serving, earliest advanced %d -> %d after state-sync re-bootstrap — TestWorkflowStateSync OK", ssNode.Name(), preEarliest, postEarliest)
 
 	// Interrupt-resume variant (deferred): kill the follower pod mid-recipe and
 	// assert the workflow resumes to Complete. Deferred here — it needs a