tutti/packages/agent/daemon/runtime/process_replay_playback.go

package agentruntime

import (
	"context"
	"errors"
	"math"
	"sync"
	"time"
)

const DefaultReplayPlaybackSpeed = 1

var (
	ErrReplayPlaybackUnavailable = errors.New("replay playback control is unavailable")
	ErrReplayPlaybackSpeed       = errors.New("unsupported replay playback speed")
	// errReplayRecvInterrupted wakes recvContext so pending synthetic optional-probe
	// responses can drain while inbound delivery is paused at a checkpoint.
	errReplayRecvInterrupted = errors.New("replay recv interrupted")
)

type ReplayPlaybackState struct {
	Speed             float64
	PlaybackElapsedMS float64
	Drained           bool
	Paused            bool
	FastForward       bool
}

type replayPlaybackController struct {
	mu                sync.Mutex
	speed             float64
	paused            bool
	fastForward       bool
	wallAnchor        time.Time
	playbackElapsedMS float64
	changed           chan struct{}
}

type replayPlaybackCursor struct {
	mu                         sync.Mutex
	controller                 *replayPlaybackController
	recordedAnchorMS           float64
	controllerPlaybackAnchorMS float64
}

func newReplayPlaybackController() *replayPlaybackController {
	return &replayPlaybackController{
		speed:      DefaultReplayPlaybackSpeed,
		wallAnchor: time.Now(),
		changed:    make(chan struct{}),
	}
}

func (c *replayPlaybackController) state() ReplayPlaybackState {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.advanceClockLocked(time.Now())
	return ReplayPlaybackState{
		Speed:             c.speed,
		PlaybackElapsedMS: c.playbackElapsedMS,
		Paused:            c.paused,
		FastForward:       c.fastForward,
	}
}

func (c *replayPlaybackController) setSpeed(speed float64) error {
	if !isSupportedReplayPlaybackSpeed(speed) {
		return ErrReplayPlaybackSpeed
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	c.advanceClockLocked(time.Now())
	if c.speed == speed {
		return nil
	}
	c.speed = speed
	c.signalChangedLocked()
	return nil
}

func (c *replayPlaybackController) setPaused(paused bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.advanceClockLocked(time.Now())
	if c.paused == paused {
		return
	}
	c.paused = paused
	c.signalChangedLocked()
}

func (c *replayPlaybackController) setFastForward(fastForward bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.advanceClockLocked(time.Now())
	if c.fastForward == fastForward {
		return
	}
	c.fastForward = fastForward
	c.signalChangedLocked()
}

func (c *replayPlaybackController) snapshot() (
	ReplayPlaybackState,
	float64,
	<-chan struct{},
) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.advanceClockLocked(time.Now())
	return ReplayPlaybackState{
		Speed:             c.speed,
		PlaybackElapsedMS: c.playbackElapsedMS,
		Paused:            c.paused,
		FastForward:       c.fastForward,
	}, c.playbackElapsedMS, c.changed
}

func (c *replayPlaybackController) signalChangedLocked() {
	close(c.changed)
	c.changed = make(chan struct{})
}

func (c *replayPlaybackController) advanceClockLocked(now time.Time) {
	if !c.paused && !c.fastForward {
		c.playbackElapsedMS += float64(now.Sub(c.wallAnchor)) /
			float64(time.Millisecond) *
			c.speed
	}
	c.wallAnchor = now
}

func (c *replayPlaybackController) newCursor() *replayPlaybackCursor {
	_, playbackElapsedMS, _ := c.snapshot()
	return &replayPlaybackCursor{
		controller:                 c,
		controllerPlaybackAnchorMS: playbackElapsedMS,
	}
}

func (c *replayPlaybackCursor) waitUntil(
	ctx context.Context,
	targetElapsedMS int64,
	closed <-chan struct{},
	interrupt <-chan struct{},
) error {
	target := float64(targetElapsedMS)
	for {
		c.mu.Lock()
		state, playbackElapsedMS, changed := c.controller.snapshot()
		c.advanceToControllerTimeLocked(playbackElapsedMS)
		if state.Paused && !state.FastForward {
			c.mu.Unlock()
			if err := waitForReplayPlaybackChange(ctx, closed, changed, interrupt); err != nil {
				return err
			}
			continue
		}
		if state.FastForward {
			if target > c.recordedAnchorMS {
				c.recordedAnchorMS = target
			}
			c.mu.Unlock()
			return nil
		}
		remainingMS := target - c.recordedAnchorMS
		if remainingMS <= 0 {
			c.mu.Unlock()
			return nil
		}
		wait := time.Duration(math.Ceil(remainingMS/state.Speed)) * time.Millisecond
		c.mu.Unlock()
		timer := time.NewTimer(wait)
		select {
		case <-ctx.Done():
			stopReplayPlaybackTimer(timer)
			return ctx.Err()
		case <-closed:
			stopReplayPlaybackTimer(timer)
			return context.Canceled
		case <-changed:
			stopReplayPlaybackTimer(timer)
		case <-interrupt:
			stopReplayPlaybackTimer(timer)
			return errReplayRecvInterrupted
		case <-timer.C:
		}
	}
}

func (c *replayPlaybackController) beginInboundRelease(
	ctx context.Context,
	closed <-chan struct{},
	interrupt <-chan struct{},
) (func(), error) {
	for {
		c.mu.Lock()
		if !c.paused || c.fastForward {
			return c.mu.Unlock, nil
		}
		changed := c.changed
		c.mu.Unlock()
		if err := waitForReplayPlaybackChange(ctx, closed, changed, interrupt); err != nil {
			return nil, err
		}
	}
}

func waitForReplayPlaybackChange(
	ctx context.Context,
	closed <-chan struct{},
	changed <-chan struct{},
	interrupt <-chan struct{},
) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-closed:
		return context.Canceled
	case <-changed:
		return nil
	case <-interrupt:
		return errReplayRecvInterrupted
	}
}

func (c *replayPlaybackCursor) advanceTo(targetElapsedMS int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	_, playbackElapsedMS, _ := c.controller.snapshot()
	c.advanceToControllerTimeLocked(playbackElapsedMS)
	if target := float64(targetElapsedMS); target > c.recordedAnchorMS {
		c.recordedAnchorMS = target
	}
}

func (c *replayPlaybackCursor) advanceToControllerTimeLocked(playbackElapsedMS float64) {
	if elapsed := playbackElapsedMS - c.controllerPlaybackAnchorMS; elapsed > 0 {
		c.recordedAnchorMS += elapsed
	}
	c.controllerPlaybackAnchorMS = playbackElapsedMS
}

func stopReplayPlaybackTimer(timer *time.Timer) {
	if !timer.Stop() {
		select {
		case <-timer.C:
		default:
		}
	}
}

func isSupportedReplayPlaybackSpeed(speed float64) bool {
	switch speed {
	case 0.25, 0.5, 1, 2, 4:
		return true
	default:
		return false
	}
}