fix: reachable replay anchor

pkg/config/config.go

@@ -1282,6 +1282,14 @@ type BootstrapFCUConfig struct {
 	MaxRetries    int    `yaml:"max_retries" mapstructure:"max_retries" json:"max_retries"`
 	Backoff       string `yaml:"backoff" mapstructure:"backoff" json:"backoff"`
 	HeadBlockHash string `yaml:"head_block_hash" mapstructure:"head_block_hash" json:"head_block_hash,omitempty"`
+
+	// RootAnchorBlockHash is the block safe/finalized point at for the run. It
+	// must sit strictly below the block the fixtures replay from (the datadir
+	// head at bootstrap), since a client will not move its head back to a block
+	// at or below the one it considers finalized. Empty lets the runner derive
+	// one. For a datadir built by advancing a snapshot, the snapshot block is
+	// the natural value: far below every anchor and the same for every client.
+	RootAnchorBlockHash string `yaml:"root_anchor_block_hash,omitempty" mapstructure:"root_anchor_block_hash" json:"root_anchor_block_hash,omitempty"`
 }
 
 // DefaultOpcodeExtractionTimeout is the per-block trace timeout applied

pkg/eest/converter.go

@@ -68,9 +68,9 @@ func ConvertFixture(name string, fixture *Fixture) (*ConvertedTest, error) {
 // phase is the shared pre_run payloads (snapshot → start block, preRun may be
 // nil) followed by the fixture's own setupEngineNewPayloads (start block →
 // per-test pre-state). The fixture's engineNewPayloads (the benchmark block)
-// become the measured test step. Each payload still emits an
-// engine_newPayload + engine_forkchoiceUpdated pair, so the chain head
-// advances naturally and no separate forkchoice injection is needed.
+// become the measured test step. Each payload emits an engine_newPayload +
+// engine_forkchoiceUpdated pair, preceded by one forkchoiceUpdated returning
+// the head to the block the fixture starts from.
 func ConvertStatefulFixture(name string, fixture *Fixture, preRun *StatefulPreRun) (*ConvertedTest, error) {
 	if fixture == nil {
 		return nil, fmt.Errorf("fixture is nil")
@@ -98,6 +98,21 @@ func ConvertStatefulFixture(name string, fixture *Fixture, preRun *StatefulPreRu
 		PayloadCount: len(setupPayloads) + len(fixture.EngineNewPayloads),
 	}
 
+	// Every fixture replays from the same anchor, but the previous test leaves
+	// the head wherever its last payload landed and nothing puts it back. Ask
+	// for the anchor rather than assuming we are on it: otherwise the first
+	// newPayload names a parent whose state the client may no longer hold and
+	// is answered ACCEPTED, orphaning every payload after it.
+	//
+	// Sent unconditionally — when the head already matches, the client says so
+	// for free, which beats tracking the head across tests.
+	anchorLine, err := buildAnchorForkchoiceCall(setupPayloads, fixture.EngineNewPayloads)
+	if err != nil {
+		return nil, fmt.Errorf("building anchor forkchoiceUpdated call: %w", err)
+	}
+
+	result.SetupLines = append(result.SetupLines, anchorLine)
+
 	for i, payload := range setupPayloads {
 		lines, err := convertPayload(payload, i+1)
 		if err != nil {
@@ -217,12 +232,45 @@ func buildNewPayloadCall(payload *EngineNewPayload, id int) (string, error) {
 // ZeroHash is the zero hash used for forkchoice state.
 const ZeroHash = "0x0000000000000000000000000000000000000000000000000000000000000000"
 
-// buildForkchoiceUpdatedCall builds an engine_forkchoiceUpdatedVX JSON-RPC call.
+// buildAnchorForkchoiceCall returns the chain head to the block this fixture
+// replays from: the parent of its first payload. Taken from the payload rather
+// than startBlockHash so it stays correct when a pre_run is prepended, whose
+// first payload descends from the snapshot block instead.
+func buildAnchorForkchoiceCall(setupPayloads, benchmarkPayloads []*EngineNewPayload) (string, error) {
+	payloads := setupPayloads
+	if len(payloads) == 0 {
+		payloads = benchmarkPayloads
+	}
+
+	if len(payloads) == 0 || payloads[0].ExecutionPayload == nil {
+		return "", fmt.Errorf("no payload to derive the anchor from")
+	}
+
+	anchor := payloads[0].ExecutionPayload.ParentHash
+	if anchor == "" {
+		return "", fmt.Errorf("first payload has no parentHash")
+	}
+
+	// id 0 keeps the payload calls numbered from 1 as before.
+	return buildForkchoiceUpdatedCallForHash(anchor, payloads[0].ForkchoiceUpdatedVersion, 0)
+}
+
+// buildForkchoiceUpdatedCall builds an engine_forkchoiceUpdatedVX JSON-RPC call
+// setting the head to the payload's own block.
 func buildForkchoiceUpdatedCall(payload *EngineNewPayload, id int) (string, error) {
-	// Use the forkchoiceUpdated version from the fixture.
-	method := fmt.Sprintf("engine_forkchoiceUpdatedV%d", payload.ForkchoiceUpdatedVersion)
+	return buildForkchoiceUpdatedCallForHash(
+		payload.ExecutionPayload.BlockHash, payload.ForkchoiceUpdatedVersion, id)
+}
 
-	blockHash := payload.ExecutionPayload.BlockHash
+// buildForkchoiceUpdatedCallForHash builds an engine_forkchoiceUpdatedVX call
+// setting the head to blockHash.
+//
+// safe and finalized stay zero, which clients read as "no update" — replaying a
+// payload should not move either marker. Note the engine API only permits the
+// zero hash "unless transition block is finalized", which these mainnet-fork
+// datadirs do not satisfy; sending a real ancestor here is worth revisiting.
+func buildForkchoiceUpdatedCallForHash(blockHash string, version, id int) (string, error) {
+	method := fmt.Sprintf("engine_forkchoiceUpdatedV%d", version)
 
 	forkchoiceState := map[string]string{
 		"headBlockHash":      blockHash,

pkg/eest/converter_test.go

@@ -285,21 +285,28 @@ func TestConvertStatefulFixture(t *testing.T) {
 	assert.Equal(t, "0xbench", result.FinalHash)
 	// 3 pre_run + 1 setup + 1 benchmark = 5 payloads.
 	assert.Equal(t, 5, result.PayloadCount)
-	// Setup = (3 pre_run + 1 setup) * 2 lines (newPayload + fcU).
-	assert.Len(t, result.SetupLines, 8)
+	// Setup = 1 anchor fcU + (3 pre_run + 1 setup) * 2 lines (newPayload + fcU).
+	assert.Len(t, result.SetupLines, 9)
 	// Test = 1 benchmark * 2 lines.
 	assert.Len(t, result.TestLines, 2)
 
+	// The anchor forkchoiceUpdated leads, pointing at the parent of the first
+	// payload. With a pre_run prepended that is the snapshot block, not the
+	// fixture's startBlockHash.
+	assert.Equal(t, "engine_forkchoiceUpdatedV3", rpcMethod(t, result.SetupLines[0]))
+	assert.Equal(t, "0xsnapshot", forkchoiceHeadHash(t, result.SetupLines[0]),
+		"the anchor must be the parent of the first pre_run block")
+
 	// Ordering by CONTENT (not just method name): the shared pre_run blocks must
-	// precede the fixture's own setup block. SetupLines are (newPayload, fcU)
-	// pairs, so newPayload lines sit at even indices: [0]=pre_run#1, [2]=pre_run#2,
-	// [4]=pre_run#3 (start), [6]=setup.
-	assert.Equal(t, "engine_newPayloadV4", rpcMethod(t, result.SetupLines[0]))
-	assert.Equal(t, "0xb1", newPayloadBlockHash(t, result.SetupLines[0]),
+	// precede the fixture's own setup block. After the anchor line, SetupLines are
+	// (newPayload, fcU) pairs, so newPayload lines sit at odd indices:
+	// [1]=pre_run#1, [3]=pre_run#2, [5]=pre_run#3 (start), [7]=setup.
+	assert.Equal(t, "engine_newPayloadV4", rpcMethod(t, result.SetupLines[1]))
+	assert.Equal(t, "0xb1", newPayloadBlockHash(t, result.SetupLines[1]),
 		"first setup line must replay the first pre_run block, not the fixture's setup")
-	assert.Equal(t, "0xstart", newPayloadBlockHash(t, result.SetupLines[4]),
+	assert.Equal(t, "0xstart", newPayloadBlockHash(t, result.SetupLines[5]),
 		"third newPayload is the last pre_run block (start)")
-	assert.Equal(t, "0xsetup", newPayloadBlockHash(t, result.SetupLines[6]),
+	assert.Equal(t, "0xsetup", newPayloadBlockHash(t, result.SetupLines[7]),
 		"the fixture's own setup block comes AFTER the pre_run blocks")
 
 	// The benchmark newPayload is the test step.
@@ -338,6 +345,25 @@ func newPayloadBlockHash(t *testing.T, line string) string {
 	return hash
 }
 
+// forkchoiceHeadHash decodes a forkchoiceUpdated line and returns its
+// headBlockHash.
+func forkchoiceHeadHash(t *testing.T, line string) string {
+	t.Helper()
+
+	var call map[string]any
+	require.NoError(t, json.Unmarshal([]byte(line), &call))
+
+	params, ok := call["params"].([]any)
+	require.True(t, ok && len(params) > 0, "forkchoiceUpdated line must carry params")
+
+	state, ok := params[0].(map[string]any)
+	require.True(t, ok, "first param must be the forkchoice state object")
+
+	hash, _ := state["headBlockHash"].(string)
+
+	return hash
+}
+
 func TestConvertStatefulFixture_NilPreRun(t *testing.T) {
 	fixture := &Fixture{
 		Info:                   &FixtureInfo{FixtureFormat: SupportedStatefulFixtureFormat},
@@ -349,10 +375,70 @@ func TestConvertStatefulFixture_NilPreRun(t *testing.T) {
 	result, err := ConvertStatefulFixture("test_stateful", fixture, nil)
 	require.NoError(t, err)
 
-	// Without pre_run, only the fixture's own setup payload is replayed.
-	assert.Len(t, result.SetupLines, 2)
+	// Without pre_run: 1 anchor fcU + the fixture's own setup payload's 2 lines.
+	assert.Len(t, result.SetupLines, 3)
 	assert.Len(t, result.TestLines, 2)
 	assert.Equal(t, 2, result.PayloadCount)
+
+	// With no pre_run the anchor is the fixture's start block.
+	assert.Equal(t, "0xstart", forkchoiceHeadHash(t, result.SetupLines[0]))
+}
+
+// Every fixture replays from the same anchor, but nothing rewinds the client
+// between tests, so the replay has to ask for the anchor itself. Without this
+// line the first newPayload names a parent that is not the head, the client
+// answers ACCEPTED, and every later payload is orphaned.
+func TestConvertStatefulFixture_AnchorForkchoicePrecedesReplay(t *testing.T) {
+	fixture := &Fixture{
+		Info:                   &FixtureInfo{FixtureFormat: SupportedStatefulFixtureFormat},
+		SnapshotBlockHash:      "0xsnapshot",
+		StartBlockHash:         "0xstart",
+		SetupEngineNewPayloads: []*EngineNewPayload{statefulPayload("0x4", "0xsetup", "0xstart")},
+		EngineNewPayloads:      []*EngineNewPayload{statefulPayload("0x5", "0xbench", "0xsetup")},
+	}
+
+	result, err := ConvertStatefulFixture("test_stateful", fixture, nil)
+	require.NoError(t, err)
+
+	require.NotEmpty(t, result.SetupLines)
+
+	assert.Equal(t, "engine_forkchoiceUpdatedV3", rpcMethod(t, result.SetupLines[0]),
+		"the replay must open with a forkchoiceUpdated, not a newPayload")
+	assert.Equal(t, "0xstart", forkchoiceHeadHash(t, result.SetupLines[0]),
+		"it must point at the parent of the first payload")
+
+	// safe and finalized stay zero: replaying a payload must not move either
+	// marker.
+	var call map[string]any
+	require.NoError(t, json.Unmarshal([]byte(result.SetupLines[0]), &call))
+	state, _ := call["params"].([]any)[0].(map[string]any)
+	assert.Equal(t, ZeroHash, state["safeBlockHash"])
+	assert.Equal(t, ZeroHash, state["finalizedBlockHash"])
+}
+
+// With a pre_run prepended the anchor must come from the first payload actually
+// replayed, which descends from the snapshot block, not startBlockHash.
+func TestConvertStatefulFixture_AnchorIsFirstPayloadParent(t *testing.T) {
+	preRun := &StatefulPreRun{
+		EngineNewPayloads: []*EngineNewPayload{
+			statefulPayload("0x1", "0xb1", "0xsnapshot"),
+			statefulPayload("0x2", "0xstart", "0xb1"),
+		},
+	}
+
+	fixture := &Fixture{
+		Info:                   &FixtureInfo{FixtureFormat: SupportedStatefulFixtureFormat},
+		SnapshotBlockHash:      "0xsnapshot",
+		StartBlockHash:         "0xstart",
+		SetupEngineNewPayloads: []*EngineNewPayload{statefulPayload("0x4", "0xsetup", "0xstart")},
+		EngineNewPayloads:      []*EngineNewPayload{statefulPayload("0x5", "0xbench", "0xsetup")},
+	}
+
+	result, err := ConvertStatefulFixture("test_stateful", fixture, preRun)
+	require.NoError(t, err)
+
+	assert.Equal(t, "0xsnapshot", forkchoiceHeadHash(t, result.SetupLines[0]),
+		"with a pre_run the anchor is the snapshot block, not startBlockHash")
 }
 
 func TestConvertStatefulFixture_NoBenchmarkPayloads(t *testing.T) {

pkg/executor/executor.go

@@ -395,7 +395,7 @@ func (e *executor) RunPreRunSteps(ctx context.Context, opts *ExecuteOptions) (in
 		log.Info("Running pre-run step")
 
 		preRunResult := NewTestResult(step.Name)
-		if err := e.runStepFile(ctx, opts, step, preRunResult, false, opts.PreRunStepSleep); err != nil {
+		if err := e.runStepFile(ctx, opts, step, preRunResult, false, opts.PreRunStepSleep, StepTypePreRun); err != nil {
 			// FailFast: surface the error to the caller without writing partial results.
 			if opts.FailFast {
 				return 0, fmt.Errorf("pre-run step %q failed: %w", step.Name, err)
@@ -493,7 +493,7 @@ func (e *executor) ExecuteTests(ctx context.Context, opts *ExecuteOptions) (*Exe
 			log.Info("Running pre-run step")
 
 			preRunResult := NewTestResult(step.Name)
-			if err := e.runStepFile(ctx, opts, step, preRunResult, false, opts.PreRunStepSleep); err != nil {
+			if err := e.runStepFile(ctx, opts, step, preRunResult, false, opts.PreRunStepSleep, StepTypePreRun); err != nil {
 				log.WithError(err).Warn("Pre-run step failed")
 
 				// Check if the failure was due to context cancellation.
@@ -576,7 +576,7 @@ func (e *executor) ExecuteTests(ctx context.Context, opts *ExecuteOptions) (*Exe
 
 			setupResult := NewTestResult(test.Name)
 
-			if err := e.runStepFile(ctx, opts, test.Setup, setupResult, false, 0); err != nil {
+			if err := e.runStepFile(ctx, opts, test.Setup, setupResult, false, 0, StepTypeSetup); err != nil {
 				log.WithError(err).Error("Setup step failed")
 				testPassed = false
 
@@ -612,7 +612,7 @@ func (e *executor) ExecuteTests(ctx context.Context, opts *ExecuteOptions) (*Exe
 
 			testResult := NewTestResult(test.Name)
 
-			if err := e.runStepFile(ctx, opts, test.Test, testResult, true, 0); err != nil {
+			if err := e.runStepFile(ctx, opts, test.Test, testResult, true, 0, StepTypeTest); err != nil {
 				log.WithError(err).Error("Test step failed")
 				testPassed = false
 
@@ -668,7 +668,7 @@ func (e *executor) ExecuteTests(ctx context.Context, opts *ExecuteOptions) (*Exe
 
 			cleanupResult := NewTestResult(test.Name)
 
-			if err := e.runStepFile(ctx, opts, test.Cleanup, cleanupResult, false, 0); err != nil {
+			if err := e.runStepFile(ctx, opts, test.Cleanup, cleanupResult, false, 0, StepTypeCleanup); err != nil {
 				log.WithError(err).Error("Cleanup step failed")
 				testPassed = false
 
@@ -814,20 +814,23 @@ writeResults:
 // runStepFile executes a single step file or provider.
 // If captureBlockLogs is true, blockHashes from engine_newPayload calls are registered for log matching.
 // betweenLineSleep, when > 0, sleeps for that duration between each RPC call.
+// stepType decides whether resume skipping applies; see runStepLines.
 func (e *executor) runStepFile(
 	ctx context.Context,
 	opts *ExecuteOptions,
 	step *StepFile,
 	result *TestResult,
 	captureBlockLogs bool,
 	betweenLineSleep time.Duration,
+	stepType StepType,
 ) error {
 	// Use provider if available, otherwise read from file.
 	if step.Provider != nil {
-		return e.runStepLines(ctx, opts, step.Name, step.Provider.Lines(), result, captureBlockLogs, betweenLineSleep)
+		return e.runStepLines(ctx, opts, step.Name, step.Provider.Lines(), result,
+			captureBlockLogs, betweenLineSleep, stepType)
 	}
 
-	return e.runStepFromFile(ctx, opts, step, result, captureBlockLogs, betweenLineSleep)
+	return e.runStepFromFile(ctx, opts, step, result, captureBlockLogs, betweenLineSleep, stepType)
 }
 
 // runStepFromFile reads and executes lines from a file.
@@ -838,6 +841,7 @@ func (e *executor) runStepFromFile(
 	result *TestResult,
 	captureBlockLogs bool,
 	betweenLineSleep time.Duration,
+	stepType StepType,
 ) error {
 	file, err := os.Open(step.Path)
 	if err != nil {
@@ -867,7 +871,8 @@ func (e *executor) runStepFromFile(
 		}
 	}
 
-	return e.runStepLines(ctx, opts, step.Name, lines, result, captureBlockLogs, betweenLineSleep)
+	return e.runStepLines(ctx, opts, step.Name, lines, result, captureBlockLogs,
+		betweenLineSleep, stepType)
 }
 
 // runStepLines executes JSON-RPC lines.
@@ -882,6 +887,7 @@ func (e *executor) runStepLines(
 	result *TestResult,
 	captureBlockLogs bool,
 	betweenLineSleep time.Duration,
+	stepType StepType,
 ) error {
 	stepStart := time.Now()
 
@@ -897,7 +903,12 @@ func (e *executor) runStepLines(
 	// skipping is true when we're dropping already-applied lines at the
 	// start of the file (resume scenario). Cleared once we encounter the
 	// first engine_newPayload whose blockNumber > SkipUntilBlockNumber.
-	skipping := opts.SkipUntilBlockNumber > 0
+	//
+	// Only the pre-run replay resumes; it alone may be partly applied. A test's
+	// steps always start from the replay anchor and must be sent whole —
+	// skipping them silently dropped any leading line that is not a newPayload,
+	// such as a forkchoiceUpdated returning the head to the anchor.
+	skipping := opts.SkipUntilBlockNumber > 0 && stepType == StepTypePreRun
 	skippedCount := 0
 
 	for lineNum, line := range lines {

pkg/runner/lifecycle.go

@@ -1078,6 +1078,28 @@ func (r *runner) runContainerLifecycle(
 		}
 	}
 
+	// Move safe/finalized below the head before anything replays. Prestates
+	// arrive with their head block already finalized, which would make the
+	// replay anchor permanently unreachable. Best-effort: a client that will
+	// not lower its finalized marker still runs, it just keeps the old
+	// failure mode.
+	if blockHash != "" {
+		var configuredAnchor string
+		if r.cfg.FullConfig != nil {
+			if fcuCfg := r.cfg.FullConfig.GetBootstrapFCU(instance); fcuCfg != nil {
+				configuredAnchor = fcuCfg.RootAnchorBlockHash
+			}
+		}
+
+		if anchorErr := r.resetForkchoiceAnchor(
+			execCtx, log, containerIP, spec.EnginePort(), spec.RPCPort(),
+			blockHash, configuredAnchor,
+		); anchorErr != nil {
+			log.WithError(anchorErr).Warn(
+				"Could not reset safe/finalized; a deep rewind to the replay anchor may fail")
+		}
+	}
+
 	// Send bootstrap FCU if configured.
 	if r.cfg.FullConfig != nil {
 		if fcuCfg := r.cfg.FullConfig.GetBootstrapFCU(instance); fcuCfg != nil && fcuCfg.Enabled {
@@ -1092,7 +1114,7 @@ func (r *runner) runContainerLifecycle(
 
 			if fcuHash != "" {
 				if fcuErr := r.sendBootstrapFCU(
-					execCtx, log, containerIP, spec.EnginePort(), fcuHash, fcuCfg,
+					execCtx, log, containerIP, spec.EnginePort(), fcuHash, "", fcuCfg,
 				); fcuErr != nil {
 					log.WithError(fcuErr).Error("Bootstrap FCU failed")