feat: resume pre-run replay against existing datadir (#219)

## Summary
Adds two pieces that together make pre-run replay resumable after an
instance crashes mid-replay (e.g. reth dies on payload N, you don't want
to re-execute payloads 1..N from scratch):

- **`direct` datadir method** — bypasses copy/snapshot/clone and
bind-mounts `source_dir` as-is into the container. Cleanup is a no-op so
the directory survives the run. Intended to point at a ZFS clone left
behind by `--debug.stop-after-prerun`.
- **Skip already-applied blocks during pre-run** — after RPC-ready, the
lifecycle reads the client's latest block number and passes it as
`ExecuteOptions.SkipUntilBlockNumber`. `runStepLines` drops every
leading line until it sees the first `engine_newPayload` whose
`blockNumber` exceeds that target, then resumes normal processing. Wired
into both the `--debug.stop-after-prerun` path (resume) and the main
`ExecuteTests` path (idempotent normal runs).

Resume workflow:

1. Find the ZFS clone path the prior `--debug.stop-after-prerun` run
logged (`data_mount=...` field on the final exit log).
2. Edit your config:
   ```yaml
   datadirs:
     reth:
       source_dir: /path/to/benchmarkoor-clone-reth/...
       method: direct
   ```
3. Re-run with `--debug.stop-after-prerun --limit-instance-id=reth`. The
container mounts the existing clone, the latest applied block is
detected, and pre-run resumes from the next block.

## Test plan
- [x] `go build`, `go test ./pkg/executor/... ./pkg/datadir/...
./pkg/config/...`, and `golangci-lint run
--new-from-rev="origin/master"` clean
- [x] End-to-end: trigger a mid-pre-run failure (reth crash),
reconfigure with `method: direct` pointing at the surviving clone,
re-run with `--debug.stop-after-prerun`, verify the "Resuming pre-run
replay" log fires with the expected `resumed_at_block` and the replay
continues from the next payload

## Notes
- The `runner-level` rollback strategy paths (`strategy_container.go`
ZFS-snapshot path, `strategy_checkpoint.go`) build their own
`ExecuteOptions` and call `RunPreRunSteps` directly. They don't yet
receive `SkipUntilBlockNumber` — would need `blockNum` threaded through
the strategy signatures. Out of scope here since the resume case goes
through `--debug.stop-after-prerun`.

Author: skylenet

pkg/config/config.go

@@ -701,9 +701,9 @@ func (d *DataDirConfig) Validate(prefix string) error {
 		return fmt.Errorf("%s: source_dir %q is not a directory", prefix, d.SourceDir)
 	}
 
-	validMethods := map[string]bool{"": true, "copy": true, "overlayfs": true, "fuse-overlayfs": true, "zfs": true}
+	validMethods := map[string]bool{"": true, "copy": true, "overlayfs": true, "fuse-overlayfs": true, "zfs": true, "direct": true}
 	if !validMethods[d.Method] {
-		return fmt.Errorf("%s: invalid method %q, must be: copy, overlayfs, fuse-overlayfs, zfs", prefix, d.Method)
+		return fmt.Errorf("%s: invalid method %q, must be: copy, overlayfs, fuse-overlayfs, zfs, direct", prefix, d.Method)
 	}
 
 	return nil

pkg/datadir/direct.go

@@ -0,0 +1,46 @@
+package datadir
+
+import (
+	"context"
+
+	"github.com/sirupsen/logrus"
+)
+
+// DirectProvider implements Provider by mounting source_dir directly,
+// without copying, snapshotting, or cloning. The container writes
+// directly to the user-supplied path, so any changes persist after
+// the run.
+//
+// Intended for inspection / resume workflows (e.g. continuing a
+// pre-run replay against an existing ZFS clone left behind by
+// --debug.stop-after-prerun). Not suitable for normal benchmarking.
+type DirectProvider interface {
+	Provider
+}
+
+// NewDirectProvider creates a new direct provider.
+func NewDirectProvider(log logrus.FieldLogger) DirectProvider {
+	return &directProvider{
+		log: log.WithField("component", "datadir-direct"),
+	}
+}
+
+type directProvider struct {
+	log logrus.FieldLogger
+}
+
+var _ DirectProvider = (*directProvider)(nil)
+
+// Prepare returns the source directory unchanged. No copy or snapshot.
+// Cleanup is a no-op so the directory survives the run.
+func (p *directProvider) Prepare(_ context.Context, cfg *ProviderConfig) (*PreparedDir, error) {
+	p.log.WithFields(logrus.Fields{
+		"source":   cfg.SourceDir,
+		"instance": cfg.InstanceID,
+	}).Warn("Using datadir method=direct: container will write directly to source_dir, changes will persist")
+
+	return &PreparedDir{
+		MountPath: cfg.SourceDir,
+		Cleanup:   func() error { return nil },
+	}, nil
+}

pkg/datadir/provider.go

@@ -27,7 +27,7 @@ type PreparedDir struct {
 }
 
 // NewProvider creates a new Provider based on the method.
-// Supported methods: "copy" (default), "overlayfs", "fuse-overlayfs", "zfs".
+// Supported methods: "copy" (default), "overlayfs", "fuse-overlayfs", "zfs", "direct".
 func NewProvider(log logrus.FieldLogger, method string) (Provider, error) {
 	switch method {
 	case "", "copy":
@@ -38,6 +38,8 @@ func NewProvider(log logrus.FieldLogger, method string) (Provider, error) {
 		return NewFuseOverlayFSProvider(log), nil
 	case "zfs":
 		return NewZFSProvider(log), nil
+	case "direct":
+		return NewDirectProvider(log), nil
 	default:
 		return nil, fmt.Errorf("unknown datadir method: %q", method)
 	}

pkg/executor/executor.go

@@ -113,6 +113,7 @@ type ExecuteOptions struct {
 	PostTestSleepDuration         time.Duration                         // Sleep duration after each test (0 = disabled).
 	FailFast                      bool                                  // If true, return an error from runStepLines on the first failed RPC call.
 	PreRunStepSleep               time.Duration                         // Sleep between each RPC call within pre-run step files (0 = disabled).
+	SkipUntilBlockNumber          uint64                                // Skip pre-run RPC lines until the first engine_newPayload with blockNumber > this. 0 = no skipping.
 }
 
 // ExecutionResult contains the overall execution summary.
@@ -830,6 +831,12 @@ func (e *executor) runStepLines(
 ) error {
 	stepStart := time.Now()
 
+	// 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
+	skippedCount := 0
+
 	for lineNum, line := range lines {
 		select {
 		case <-ctx.Done():
@@ -859,6 +866,35 @@ func (e *executor) runStepLines(
 			continue
 		}
 
+		// Skip already-applied lines if requested. Stay in skipping mode
+		// until we see a newPayload past the target; everything before
+		// (other newPayloads, FCUs, etc.) is for a block already on
+		// disk and would just be redundant work.
+		if skipping {
+			drop := true
+
+			if strings.HasPrefix(method, "engine_newPayload") {
+				if bn, ok := extractBlockNumber(line); ok && bn > opts.SkipUntilBlockNumber {
+					skipping = false
+					drop = false
+
+					e.log.WithFields(logrus.Fields{
+						"step":             stepName,
+						"resumed_at_line":  lineNum + 1,
+						"resumed_at_block": bn,
+						"skipped_lines":    skippedCount,
+						"skip_until_block": opts.SkipUntilBlockNumber,
+					}).Info("Resuming pre-run replay")
+				}
+			}
+
+			if drop {
+				skippedCount++
+
+				continue
+			}
+		}
+
 		// Register blockHash BEFORE the RPC call for engine_newPayload methods.
 		if captureBlockLogs && strings.HasPrefix(method, "engine_newPayload") &&
 			opts.BlockLogCollector != nil && result != nil {

pkg/executor/results.go

@@ -307,6 +307,41 @@ func extractGasUsed(request string) (uint64, error) {
 	return strconv.ParseUint(strings.TrimPrefix(payload.GasUsed, "0x"), 16, 64)
 }
 
+// extractBlockNumber extracts blockNumber from an engine_newPayload request.
+// Returns the parsed block number and ok=true on success. Returns ok=false
+// for any parse failure or missing field — callers should treat this as
+// "unknown" rather than an error.
+func extractBlockNumber(request string) (uint64, bool) {
+	var req struct {
+		Params []json.RawMessage `json:"params"`
+	}
+	if err := json.Unmarshal([]byte(request), &req); err != nil {
+		return 0, false
+	}
+
+	if len(req.Params) == 0 {
+		return 0, false
+	}
+
+	var payload struct {
+		BlockNumber string `json:"blockNumber"`
+	}
+	if err := json.Unmarshal(req.Params[0], &payload); err != nil {
+		return 0, false
+	}
+
+	if payload.BlockNumber == "" {
+		return 0, false
+	}
+
+	n, err := strconv.ParseUint(strings.TrimPrefix(payload.BlockNumber, "0x"), 16, 64)
+	if err != nil {
+		return 0, false
+	}
+
+	return n, true
+}
+
 // extractBlockHash extracts blockHash from an engine_newPayload request.
 func extractBlockHash(request string) (string, error) {
 	var req struct {

pkg/runner/lifecycle.go

@@ -1013,10 +1013,11 @@ func (r *runner) runContainerLifecycle(
 				EngineEndpoint: fmt.Sprintf(
 					"http://%s:%d", containerIP, spec.EnginePort(),
 				),
-				JWT:             r.cfg.JWT,
-				ResultsDir:      runResultsDir,
-				FailFast:        true,
-				PreRunStepSleep: r.cfg.PreRunStepSleep,
+				JWT:                  r.cfg.JWT,
+				ResultsDir:           runResultsDir,
+				FailFast:             true,
+				PreRunStepSleep:      r.cfg.PreRunStepSleep,
+				SkipUntilBlockNumber: blockNum,
 			}
 
 			if n, err := r.executor.RunPreRunSteps(execCtx, preRunOpts); err != nil {
@@ -1122,6 +1123,7 @@ func (r *runner) runContainerLifecycle(
 				PostTestRPCCalls:              r.cfg.FullConfig.GetPostTestRPCCalls(instance),
 				PostTestSleepDuration:         r.cfg.FullConfig.GetPostTestSleepDuration(instance),
 				PreRunStepSleep:               r.cfg.PreRunStepSleep,
+				SkipUntilBlockNumber:          blockNum,
 			}
 
 			result, execErr = r.executor.ExecuteTests(execCtx, execOpts)