op-service: Support NewTimer in Clock

op-service/clock/clock.go

@@ -22,6 +22,10 @@ type Clock interface {
 	// The duration d must be greater than zero; if not, NewTicker will
 	// panic. Stop the ticker to release associated resources.
 	NewTicker(d time.Duration) Ticker
+
+	// NewTimer creates a new Timer that will send
+	// the current time on its channel after at least duration d.
+	NewTimer(d time.Duration) Timer
 }
 
 // A Ticker holds a channel that delivers "ticks" of a clock at intervals
@@ -42,9 +46,14 @@ type Ticker interface {
 
 // Timer represents a single event.
 type Timer interface {
+	// Ch returns the channel for the ticker. Equivalent to time.Timer.C
+	Ch() <-chan time.Time
+
 	// Stop prevents the Timer from firing.
 	// It returns true if the call stops the timer, false if the timer has already
 	// expired or been stopped.
+	// Stop does not close the channel, to prevent a read from the channel succeeding
+	// incorrectly.
 	//
 	// For a timer created with AfterFunc(d, f), if t.Stop returns false, then the timer
 	// has already expired and the function f has been started in its own goroutine;
@@ -80,10 +89,18 @@ func (s systemClock) NewTicker(d time.Duration) Ticker {
 	return &SystemTicker{time.NewTicker(d)}
 }
 
+func (s systemClock) NewTimer(d time.Duration) Timer {
+	return &SystemTimer{time.NewTimer(d)}
+}
+
 type SystemTimer struct {
 	*time.Timer
 }
 
+func (t *SystemTimer) Ch() <-chan time.Time {
+	return t.C
+}
+
 func (s systemClock) AfterFunc(d time.Duration, f func()) Timer {
 	return &SystemTimer{time.AfterFunc(d, f)}
 }

op-service/clock/deterministic.go

@@ -31,6 +31,7 @@ func (t task) fire(now time.Time) bool {
 
 type timer struct {
 	f       func()
+	ch      chan time.Time
 	due     time.Time
 	stopped bool
 	run     bool
@@ -53,6 +54,10 @@ func (t *timer) fire(now time.Time) bool {
 	return false
 }
 
+func (t *timer) Ch() <-chan time.Time {
+	return t.ch
+}
+
 func (t *timer) Stop() bool {
 	t.Lock()
 	defer t.Unlock()
@@ -171,6 +176,21 @@ func (s *DeterministicClock) NewTicker(d time.Duration) Ticker {
 	return t
 }
 
+func (s *DeterministicClock) NewTimer(d time.Duration) Timer {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	ch := make(chan time.Time, 1)
+	t := &timer{
+		f: func() {
+			ch <- s.now
+		},
+		ch:  ch,
+		due: s.now.Add(d),
+	}
+	s.addPending(t)
+	return t
+}
+
 func (s *DeterministicClock) addPending(t action) {
 	s.pending = append(s.pending, t)
 	select {

op-service/clock/deterministic_test.go

@@ -217,6 +217,46 @@ func TestNewTicker(t *testing.T) {
 	})
 }
 
+func TestNewTimer(t *testing.T) {
+	t.Run("FireOnceAfterDuration", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		timer := clock.NewTimer(5 * time.Second)
+
+		require.Len(t, timer.Ch(), 0, "should not fire immediately")
+
+		clock.AdvanceTime(4 * time.Second)
+		require.Len(t, timer.Ch(), 0, "should not fire before due")
+
+		clock.AdvanceTime(1 * time.Second)
+		require.Len(t, timer.Ch(), 1, "should fire when due")
+		require.Equal(t, clock.Now(), <-timer.Ch(), "should post current time")
+
+		clock.AdvanceTime(6 * time.Second)
+		require.Len(t, timer.Ch(), 0, "should not fire when due again")
+	})
+
+	t.Run("StopBeforeExecuted", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		timer := clock.NewTimer(5 * time.Second)
+
+		require.True(t, timer.Stop())
+
+		clock.AdvanceTime(10 * time.Second)
+		require.Len(t, timer.Ch(), 0, "should not fire after stop")
+	})
+
+	t.Run("StopAfterExecuted", func(t *testing.T) {
+		clock := NewDeterministicClock(time.UnixMilli(1000))
+		timer := clock.NewTimer(5 * time.Second)
+
+		clock.AdvanceTime(10 * time.Second)
+		require.Len(t, timer.Ch(), 1, "should fire when due")
+		require.Equal(t, clock.Now(), <-timer.Ch(), "should post current time")
+
+		require.False(t, timer.Stop())
+	})
+}
+
 func TestWaitForPending(t *testing.T) {
 	t.Run("DoNotBlockWhenAlreadyPending", func(t *testing.T) {
 		clock := NewDeterministicClock(time.UnixMilli(1000))