feat(replay): step a past slot range through dora as if it were live

clients/consensus/chainstate.go

@@ -2,16 +2,19 @@ package consensus
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha256"
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/ethpandaops/dora/utils"
 	"github.com/ethpandaops/ethwallclock"
 	v1 "github.com/ethpandaops/go-eth2-client/api/v1"
 	"github.com/ethpandaops/go-eth2-client/spec/phase0"
+	"github.com/sirupsen/logrus"
 )
 
 type ChainState struct {
@@ -22,8 +25,13 @@ type ChainState struct {
 	genesisMutex sync.Mutex
 	genesis      *v1.Genesis
 
-	wallclockMutex sync.Mutex
-	wallclock      *ethwallclock.EthereumBeaconChain
+	wallclockMutex   sync.Mutex
+	wallclockStarted bool
+	wallclock        *ethwallclock.EthereumBeaconChain
+
+	// replayClock replaces the wall clock with the virtual clock of a dora-replay
+	// control server, so a past slot range can be stepped through as if it were live.
+	replayClock atomic.Pointer[replayClock]
 
 	finalityMutex sync.RWMutex
 	finality      *v1.Finality
@@ -198,14 +206,23 @@ func (cs *ChainState) initWallclock() {
 	cs.wallclockMutex.Lock()
 	defer cs.wallclockMutex.Unlock()
 
-	if cs.wallclock != nil {
+	if cs.wallclockStarted {
 		return
 	}
 
 	if cs.specs == nil || cs.genesis == nil {
 		return
 	}
 
+	cs.wallclockStarted = true
+
+	// ethwallclock reads time.Now() internally and cannot be paused, so a replay run
+	// drives the slot/epoch dispatchers off the virtual clock instead.
+	if clock := cs.replayClock.Load(); clock != nil {
+		go cs.runReplayWallclock(clock)
+		return
+	}
+
 	cs.wallclock = ethwallclock.NewEthereumBeaconChain(cs.genesis.GenesisTime, time.Duration(cs.specs.SlotDurationMs)*time.Millisecond, cs.specs.SlotsPerEpoch)
 	cs.wallclock.OnEpochChanged(func(current ethwallclock.Epoch) {
 		cs.wallclockEpochDispatcher.Fire(&current)
@@ -215,6 +232,105 @@ func (cs *ChainState) initWallclock() {
 	})
 }
 
+// EnableReplayClock points the chain state at a dora-replay control server, so every
+// "now" in the explorer (current slot, wallclock ticks, page timestamps) follows the
+// replayed slot range instead of the real wall clock. It blocks until the control
+// server has answered once.
+func (cs *ChainState) EnableReplayClock(ctx context.Context, logger logrus.FieldLogger, controlURL string, pollInterval time.Duration) error {
+	clock, err := newReplayClock(ctx, logger, controlURL, pollInterval)
+	if err != nil {
+		return err
+	}
+
+	cs.replayClock.Store(clock)
+
+	logger.WithField("now", clock.now().UTC().Format(time.RFC3339)).Info("replay clock enabled")
+
+	return nil
+}
+
+// IsReplaying reports whether the chain state is driven by a replay clock.
+func (cs *ChainState) IsReplaying() bool {
+	return cs.replayClock.Load() != nil
+}
+
+// Now returns the current time as the chain sees it: the real wall clock normally,
+// or the virtual replay time while a replay is running.
+func (cs *ChainState) Now() time.Time {
+	if clock := cs.replayClock.Load(); clock != nil {
+		return clock.now()
+	}
+
+	return time.Now()
+}
+
+// maxReplayWallclockCatchup bounds how many slot ticks are replayed in one go after
+// the virtual clock jumped, so a large seek does not flood the dispatchers.
+const maxReplayWallclockCatchup = 16
+
+// runReplayWallclock fires the slot and epoch dispatchers from the virtual clock,
+// standing in for ethwallclock while a replay is running.
+func (cs *ChainState) runReplayWallclock(clock *replayClock) {
+	defer utils.HandleSubroutinePanic("clients.consensus.ChainState.runReplayWallclock", func() {
+		cs.runReplayWallclock(clock)
+	})
+
+	genesis := cs.genesis.GenesisTime
+	slotDuration := time.Duration(cs.specs.SlotDurationMs) * time.Millisecond
+	slotsPerEpoch := cs.specs.SlotsPerEpoch
+
+	interval := slotDuration / 20
+	if interval < 50*time.Millisecond {
+		interval = 50 * time.Millisecond
+	} else if interval > 250*time.Millisecond {
+		interval = 250 * time.Millisecond
+	}
+
+	lastSlot := uint64(cs.CurrentSlot())
+	lastEpoch := lastSlot / slotsPerEpoch
+
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-clock.ctx.Done():
+			return
+		case <-ticker.C:
+		}
+
+		slot := uint64(cs.CurrentSlot())
+		if slot <= lastSlot {
+			// the replay was rewound; resync without firing anything
+			lastSlot = slot
+			lastEpoch = slot / slotsPerEpoch
+
+			continue
+		}
+
+		firstSlot := lastSlot + 1
+		if slot-lastSlot > maxReplayWallclockCatchup {
+			firstSlot = slot
+		}
+
+		for tickSlot := firstSlot; tickSlot <= slot; tickSlot++ {
+			slotStart := genesis.Add(time.Duration(tickSlot) * slotDuration)
+
+			if epoch := tickSlot / slotsPerEpoch; epoch != lastEpoch {
+				epochStart := genesis.Add(time.Duration(epoch*slotsPerEpoch) * slotDuration)
+				wallclockEpoch := ethwallclock.NewEpoch(epoch, epochStart, epochStart.Add(slotDuration*time.Duration(slotsPerEpoch)))
+				cs.wallclockEpochDispatcher.Fire(&wallclockEpoch)
+				lastEpoch = epoch
+			}
+
+			wallclockSlot := ethwallclock.NewSlot(tickSlot, slotStart, slotStart.Add(slotDuration))
+			cs.wallclockSlotDispatcher.Fire(&wallclockSlot)
+		}
+
+		lastSlot = slot
+	}
+}
+
 func (cs *ChainState) setFinalizedCheckpoint(finality *v1.Finality) {
 	cs.finalityMutex.Lock()
 	if cs.finality != nil && finality.Justified.Epoch <= cs.finality.Justified.Epoch && finality.Finalized.Epoch <= cs.finality.Finalized.Epoch {
@@ -300,7 +416,7 @@ func (cs *ChainState) GetFinalizedSlot() phase0.Slot {
 }
 
 func (cs *ChainState) CurrentSlot() phase0.Slot {
-	return cs.TimeToSlot(time.Now())
+	return cs.TimeToSlot(cs.Now())
 }
 
 func (cs *ChainState) CurrentEpoch() phase0.Epoch {

clients/consensus/replayclock.go

@@ -0,0 +1,166 @@
+package consensus
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	// defaultReplayPollInterval is how often the virtual clock is refreshed from the
+	// control server. Between polls the clock is interpolated locally from the last
+	// reported rate, so a coarse interval does not make the clock jumpy.
+	defaultReplayPollInterval = 100 * time.Millisecond
+
+	// replayConnectTimeout bounds how long EnableReplayClock waits for the control
+	// server to become reachable before giving up.
+	replayConnectTimeout = 2 * time.Minute
+)
+
+// replayClockState is the payload served by a dora-replay control server at
+// GET /replay/clock. Rate is the number of virtual milliseconds that pass per real
+// millisecond; it is 0 while the replay is paused.
+type replayClockState struct {
+	TimeMs int64   `json:"time_ms"`
+	Rate   float64 `json:"rate"`
+}
+
+// replayClock mirrors the virtual clock of a dora-replay control server, so the whole
+// explorer sees a simulated "now" while a past slot range is stepped through.
+type replayClock struct {
+	ctx    context.Context
+	logger logrus.FieldLogger
+	url    string
+	client *http.Client
+
+	mutex      sync.RWMutex
+	anchorReal time.Time
+	anchorVirt time.Time
+	rate       float64
+}
+
+// newReplayClock connects to the control server and blocks until the first clock
+// state has been read, so callers never observe an uninitialized virtual time.
+func newReplayClock(ctx context.Context, logger logrus.FieldLogger, controlURL string, pollInterval time.Duration) (*replayClock, error) {
+	if controlURL == "" {
+		return nil, fmt.Errorf("no replay control url configured")
+	}
+
+	if pollInterval <= 0 {
+		pollInterval = defaultReplayPollInterval
+	}
+
+	clock := &replayClock{
+		ctx:    ctx,
+		logger: logger,
+		url:    strings.TrimSuffix(controlURL, "/") + "/replay/clock",
+		client: &http.Client{Timeout: 10 * time.Second},
+	}
+
+	if err := clock.awaitFirstPoll(); err != nil {
+		return nil, err
+	}
+
+	go clock.runPollLoop(pollInterval)
+
+	return clock, nil
+}
+
+// awaitFirstPoll retries the control server until it answers or the connect timeout
+// expires. The explorer cannot compute any slot before this succeeds.
+func (c *replayClock) awaitFirstPoll() error {
+	deadline := time.Now().Add(replayConnectTimeout)
+	lastLog := time.Time{}
+
+	for {
+		err := c.poll()
+		if err == nil {
+			return nil
+		}
+
+		if time.Now().After(deadline) {
+			return fmt.Errorf("could not reach replay control server at %v: %w", c.url, err)
+		}
+
+		if time.Since(lastLog) > 5*time.Second {
+			c.logger.WithError(err).Warnf("waiting for replay control server at %v", c.url)
+			lastLog = time.Now()
+		}
+
+		select {
+		case <-c.ctx.Done():
+			return c.ctx.Err()
+		case <-time.After(500 * time.Millisecond):
+		}
+	}
+}
+
+func (c *replayClock) runPollLoop(pollInterval time.Duration) {
+	ticker := time.NewTicker(pollInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		case <-ticker.C:
+			if err := c.poll(); err != nil {
+				c.logger.WithError(err).Debugf("failed polling replay clock")
+			}
+		}
+	}
+}
+
+func (c *replayClock) poll() error {
+	ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.url, http.NoBody)
+	if err != nil {
+		return err
+	}
+
+	rsp, err := c.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = rsp.Body.Close() }()
+
+	if rsp.StatusCode != http.StatusOK {
+		return fmt.Errorf("unexpected status %v", rsp.StatusCode)
+	}
+
+	state := replayClockState{}
+	if err := json.NewDecoder(rsp.Body).Decode(&state); err != nil {
+		return fmt.Errorf("error parsing replay clock response: %w", err)
+	}
+
+	c.mutex.Lock()
+	c.anchorReal = time.Now()
+	c.anchorVirt = time.UnixMilli(state.TimeMs)
+	c.rate = state.Rate
+	c.mutex.Unlock()
+
+	return nil
+}
+
+// now returns the current virtual time, interpolated from the last polled state so
+// the clock keeps moving smoothly between polls while the replay is playing.
+func (c *replayClock) now() time.Time {
+	c.mutex.RLock()
+	defer c.mutex.RUnlock()
+
+	if c.rate == 0 {
+		return c.anchorVirt
+	}
+
+	elapsed := time.Since(c.anchorReal)
+
+	return c.anchorVirt.Add(time.Duration(float64(elapsed) * c.rate))
+}

cmd/dora-explorer/main.go

@@ -63,6 +63,14 @@ func main() {
 
 	services.InitChainService(ctx, logger)
 
+	if cfg.Replay.Enabled {
+		chainState := services.GlobalBeaconService.GetChainState()
+		err = chainState.EnableReplayClock(ctx, logger.WithField("service", "replay-clock"), cfg.Replay.ControlUrl, cfg.Replay.PollInterval)
+		if err != nil {
+			logger.Fatalf("error connecting to replay control server: %v", err)
+		}
+	}
+
 	var webserver *http.Server
 	if cfg.Frontend.Enabled || cfg.Api.Enabled {
 		websrv, err := startWebserver(logger)