diff --git a/mock.go b/mock.go index 6325cd8..aefd17d 100644 --- a/mock.go +++ b/mock.go @@ -33,6 +33,9 @@ type Mock struct { // writes holds the timers that have a goroutine sending their fired value on C. The cleanup // registered in NewMock ends those goroutines. writes map[*Timer]struct{} + // tickers holds the tickers whose runLoop goroutine is running. The cleanup registered in + // NewMock stops those goroutines. + tickers map[*Ticker]struct{} // cur is the current time cur time.Time @@ -71,10 +74,9 @@ func (m *Mock) TickerFunc(ctx context.Context, d time.Duration, f func() error, return t } -// NewTicker creates a mocked ticker attached to this Mock. The ticker stops sending ticks when the -// cleanup that NewTicker registers runs, and advancing the clock after that sends none. A Reset -// after that cleanup starts the ticker again, and only Stop ends it then. Best practice is to only -// manipulate the mock time in the main goroutine of the test. +// NewTicker creates a mocked ticker attached to this Mock. See NewMock for what happens to ticks +// when the test's cleanups run. Best practice is to only manipulate the mock time in the main +// goroutine of the test. func (m *Mock) NewTicker(d time.Duration, tags ...string) *Ticker { if d <= 0 { panic("NewTicker called with negative or zero duration") @@ -487,22 +489,21 @@ func (m *Mock) WithLogger(l Logger) *Mock { // You may re-set the time earlier than this, but only before timers or tickers // are created. // -// Cleanups registered after NewMock run before NewMock's own cleanup, so they can still receive -// the value of a timer that fired during the test. NewMock's cleanup drops timer values that have -// not been received, and a timer from NewTimer that fires after it sends only to a reader already -// waiting on C. Call NewMock before registering cleanups that wait for code reading the Mock's -// timers. Ticks from NewTicker stop earlier, unless Reset starts the ticker again; see -// [Mock.NewTicker]. +// Call NewMock before registering cleanups that wait for code reading the Mock's timers or +// tickers: NewMock's cleanup drops any timer value or tick that has not been received. After that +// cleanup, a timer from NewTimer or a ticker from NewTicker sends only to a reader already waiting +// on C. func NewMock(tb TestingT) *Mock { cur, err := time.Parse(time.RFC3339, "2024-01-01T00:00:00Z") if err != nil { panic(err) } m := &Mock{ - tb: tb, - logger: tb, - cur: cur, - writes: make(map[*Timer]struct{}), + tb: tb, + logger: tb, + cur: cur, + writes: make(map[*Timer]struct{}), + tickers: make(map[*Ticker]struct{}), } tb.Cleanup(func() { m.mu.Lock() @@ -511,6 +512,9 @@ func NewMock(tb TestingT) *Mock { for t := range m.writes { t.endWriteLocked() } + for t := range m.tickers { + t.stopRunLoopLocked() + } m.logger.Logf("Mock Clock - test cleanup; will no longer log clock events") }) return m diff --git a/mock_test.go b/mock_test.go index 5c7ac26..8669872 100644 --- a/mock_test.go +++ b/mock_test.go @@ -651,65 +651,87 @@ func TestTimerReset_NegativeDuration(t *testing.T) { } } -// TestTimer_ReadableDuringCleanup checks that a cleanup registered after NewMock, which runs before -// NewMock's cleanup, can still receive from a timer that fired during the test. Tests that stop a -// server in such a cleanup rely on this when a worker reads the timer on its way out. -func TestTimer_ReadableDuringCleanup(t *testing.T) { +// newTimer and newTicker create a timer or ticker that fires after one second and return its +// channel. +func newTimer(m *quartz.Mock) <-chan time.Time { return m.NewTimer(time.Second).C } +func newTicker(m *quartz.Mock) <-chan time.Time { return m.NewTicker(time.Second).C } + +// TestReadableDuringCleanup checks that a cleanup registered after NewMock, which runs before +// NewMock's cleanup, can still receive from a timer or ticker that fired during the test. Tests +// that stop a server in such a cleanup rely on this when a worker reads the channel on its way out. +func TestReadableDuringCleanup(t *testing.T) { t.Parallel() - synctest.Test(t, func(t *testing.T) { - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() + for _, tc := range []struct { + name string + start func(*quartz.Mock) <-chan time.Time + }{ + {name: "Timer", start: newTimer}, + {name: "Ticker", start: newTicker}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() - mClock := quartz.NewMock(t) - busy := make(chan struct{}) - quit := make(chan struct{}) - exited := make(chan struct{}) - // Stands in for a server's Close: it unblocks the worker and waits for it to exit. Bubble - // time moves once every goroutine is blocked, so a worker that never receives fails the - // test at once. quit then lets the worker exit, so the bubble can end. - t.Cleanup(func() { - close(busy) - select { - case <-exited: - case <-time.After(10 * time.Second): - t.Error("worker did not receive the timer value during cleanup") - } - close(quit) + mClock := quartz.NewMock(t) + busy := make(chan struct{}) + quit := make(chan struct{}) + exited := make(chan struct{}) + // Stands in for a server's Close: it unblocks the worker and waits for it to exit. + // It is registered before the timer or ticker is created, as when the server + // creates them. Bubble time moves once every goroutine is blocked, so a worker that + // never receives fails the test at once. quit then lets the worker exit, so the + // bubble can end. + t.Cleanup(func() { + close(busy) + select { + case <-exited: + case <-time.After(10 * time.Second): + t.Error("worker did not receive from the channel during cleanup") + } + close(quit) + }) + c := tc.start(mClock) + go func() { + defer close(exited) + <-busy + select { + case <-c: + case <-quit: + } + }() + mClock.Advance(time.Second).MustWait(ctx) + }) }) - tmr := mClock.NewTimer(time.Second) - go func() { - defer close(exited) - <-busy - select { - case <-tmr.C: - case <-quit: - } - }() - mClock.Advance(time.Second).MustWait(ctx) - }) + } } -// TestTimer_ReaderWaitingAtMockCleanup checks that a reader waiting on C when NewMock's cleanup -// runs receives the value of a timer that fires just before or after that cleanup. synctest.Wait -// makes sure the reader is waiting before the timer fires and that the send has ended before the -// check. -func TestTimer_ReaderWaitingAtMockCleanup(t *testing.T) { +// TestReaderWaitingAtMockCleanup checks that a reader waiting on C when NewMock's cleanup runs +// receives a timer value or tick sent just before or after that cleanup. synctest.Wait makes sure +// the reader is waiting before the clock fires and that the send has ended before the check. +func TestReaderWaitingAtMockCleanup(t *testing.T) { t.Parallel() for _, tc := range []struct { name string + start func(*quartz.Mock) <-chan time.Time fireAfterCleanup bool }{ - // The test fires the timer as its last step. In a bubble, the goroutine that sends the - // value often starts only after NewMock's cleanup is waiting to end it. - {name: "FiredBeforeCleanup"}, - {name: "FiredAfterCleanup", fireAfterCleanup: true}, + // These two fire as the test's last step. In a bubble, the goroutine that sends on C often + // reaches its send only after NewMock's cleanup is waiting to end it. + {name: "TimerFiredBeforeCleanup", start: newTimer}, + {name: "TickerFiredBeforeCleanup", start: newTicker}, + // These two fire from the cleanup registered before NewMock, which runs after NewMock's. + {name: "TimerFiredAfterCleanup", start: newTimer, fireAfterCleanup: true}, + {name: "TickerFiredAfterCleanup", start: newTicker, fireAfterCleanup: true}, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() - // When the sending goroutine starts late, select picks at random between the reader - // and the cleanup, so a send that does not try the reader first loses the value about - // half the time. Each round is a new bubble, so without the race detector, which - // changes scheduling, the test then fails in nearly every run. + // A send that reaches its select after NewMock's cleanup is waiting to end it finds the + // reader and the cleanup both ready, and select picks between them at random. A send + // that does not try the reader first then loses the value about half the time. Each + // round is a new bubble, so without the race detector, which changes scheduling, the + // test then fails in nearly every run. for range 10 { synctest.Test(t, func(t *testing.T) { var mClock *quartz.Mock @@ -723,14 +745,14 @@ func TestTimer_ReaderWaitingAtMockCleanup(t *testing.T) { synctest.Wait() close(quit) if len(received) == 0 { - t.Error("the waiting reader did not receive the timer value") + t.Error("the waiting reader did not receive the value") } }) mClock = quartz.NewMock(t) - tmr := mClock.NewTimer(time.Second) + c := tc.start(mClock) go func() { select { - case <-tmr.C: + case <-c: received <- struct{}{} case <-quit: } @@ -772,39 +794,144 @@ func TestNoLeak_Go123(t *testing.T) { } }) - // A test that stops waiting on an Advance can end before its timers fire. - t.Run("TimerFiredAfterTestEnds", func(t *testing.T) { + // Code that outlives a test, or an Advance it stopped waiting on, can use the Mock afterward. + t.Run("AfterTestEnds", func(t *testing.T) { t.Parallel() - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() + for _, tc := range []struct { + name string + during func(*quartz.Mock) (after func()) + }{ + { + name: "TimerFired", + during: func(m *quartz.Mock) func() { + _ = m.NewTimer(time.Second) + return func() {} + }, + }, + { + name: "TickerResetAfterStop", + during: func(m *quartz.Mock) func() { + tkr := m.NewTicker(time.Second) + tkr.Stop() + return func() { tkr.Reset(time.Second) } + }, + }, + { + name: "NewTicker", + during: func(m *quartz.Mock) func() { + return func() { _ = m.NewTicker(time.Second) } + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() - var mClock *quartz.Mock - t.Run("Inner", func(t *testing.T) { - mClock = quartz.NewMock(t) - _ = mClock.NewTimer(time.Second) - }) - if err := mClock.Advance(time.Second).Wait(ctx); err != nil { - t.Fatal(err) + var mClock *quartz.Mock + var after func() + t.Run("Inner", func(t *testing.T) { + mClock = quartz.NewMock(t) + after = tc.during(mClock) + }) + after() + if err := mClock.Advance(time.Second).Wait(ctx); err != nil { + t.Fatal(err) + } + }) } }) } -func TestTickerStop_Go123(t *testing.T) { +// TestTicker_ResetInBubble resets a running ticker from a synctest bubble that did not create its +// Mock. Reset must start no goroutine there: one waiting on the Mock's channels, which belong to no +// bubble, keeps synctest.Test from returning, and go test -timeout then stops the run with this +// test's synctest.Test call in the stack. +func TestTicker_ResetInBubble(t *testing.T) { t.Parallel() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() mClock := quartz.NewMock(t) + tkr := mClock.NewTicker(time.Second) + synctest.Test(t, func(*testing.T) { tkr.Reset(time.Hour) }) + mClock.Advance(time.Hour).MustWait(ctx) + select { + case <-tkr.C: + case <-ctx.Done(): + t.Fatal("timed out waiting for the tick after Reset") + } +} - tkr := mClock.NewTicker(1 * time.Second) - mClock.Advance(1 * time.Second).MustWait(ctx) - tkr.Stop() +// TestTicker_StopAndReset checks that no tick is received after Stop or Reset returns, and that +// Reset's period applies, as with a time.Ticker. +func TestTicker_StopAndReset(t *testing.T) { + t.Parallel() + ticked := func(ctx context.Context, m *quartz.Mock) *quartz.Ticker { + tkr := m.NewTicker(time.Second) + m.Advance(time.Second).MustWait(ctx) + return tkr + } + // Advance fires the ticker from a new goroutine, which can run before or after the call. + advancing := func(_ context.Context, m *quartz.Mock) *quartz.Ticker { + tkr := m.NewTicker(time.Second) + _ = m.Advance(time.Second) + return tkr + } + stop := func(tkr *quartz.Ticker) { tkr.Stop() } + reset := func(tkr *quartz.Ticker) { tkr.Reset(time.Hour) } + for _, tc := range []struct { + name string + start func(context.Context, *quartz.Mock) *quartz.Ticker + call func(*quartz.Ticker) + // wantNext is how long after the call the ticker ticks, or 0 if it does not. + wantNext time.Duration + }{ + {name: "StopTickedUnread", start: ticked, call: stop}, + {name: "ResetTickedUnread", start: ticked, call: reset, wantNext: time.Hour}, + {name: "StopDuringAdvance", start: advancing, call: stop}, + {name: "ResetDuringAdvance", start: advancing, call: reset, wantNext: time.Hour}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() - select { - case tme := <-tkr.C: - t.Fatalf("got channel read after stop: %s", tme) - default: - // OK! + mClock := quartz.NewMock(t) + tkr := tc.start(ctx, mClock) + tc.call(tkr) + // Lets any goroutine that could still send on C run before the checks. + synctest.Wait() + select { + case tme := <-tkr.C: + t.Errorf("received %s after the call returned", tme) + default: + } + if next, _ := mClock.Peek(); next != tc.wantNext { + t.Errorf("ticker ticks in %s, want %s", next, tc.wantNext) + } + }) + }) + } +} + +// TestTickerReset_NonPositiveDuration checks that Reset panics for a duration that is not positive, +// as time.Ticker.Reset and NewTicker do. A mock ticker with such a period made the next fire loop +// forever with the Mock's lock held. +func TestTickerReset_NonPositiveDuration(t *testing.T) { + t.Parallel() + for _, d := range []time.Duration{0, -time.Second} { + t.Run(d.String(), func(t *testing.T) { + t.Parallel() + tkr := quartz.NewMock(t).NewTicker(time.Second) + defer func() { + if recover() == nil { + t.Errorf("Reset(%s) did not panic", d) + } + }() + tkr.Reset(d) + }) } } diff --git a/ticker.go b/ticker.go index f4a4b06..3600ba8 100644 --- a/ticker.go +++ b/ticker.go @@ -13,24 +13,31 @@ type Ticker struct { mock *Mock // mock clock, if set stopped bool // true if the ticker is not running internalTicks chan time.Time // used to deliver ticks to the runLoop goroutine + // resets is how Reset tells runLoop to drop a tick from the old schedule. + resets chan struct{} - // As of Go 1.23, ticker channels are unbuffered and guaranteed to block forever after a call to stop. + // As of Go 1.23, ticker channels are unbuffered and guaranteed to block forever after a call to + // stop. // - // When a mocked ticker fires, we don't want to block on a channel write, because it's fine for the code under test - // not to be reading. That means we need to start a new goroutine to do the channel write (runLoop) if we are a - // channel-based ticker. + // When a mocked ticker fires, we don't want to block on a channel write, because it's fine for + // the code under test not to be reading. That means we need to start a new goroutine to do the + // channel write (runLoop) if we are a channel-based ticker. // - // They also are not supposed to leak even if they are never read or stopped (Go runtime can garbage collect them). - // We can't garbage-collect because we can't check if any other code besides the mock references, but we can ensure - // that we don't leak goroutines so that the garbage collector can do its job when the mock is no longer - // referenced. The channels below allow us to interrupt the runLoop goroutine. + // They also are not supposed to leak even if they are never read or stopped (Go runtime can + // garbage collect them). We can't garbage-collect because we can't check if any other code + // besides the mock references, but we can ensure that we don't leak goroutines so that the + // garbage collector can do its job when the mock is no longer referenced. interrupt lets Stop + // and the cleanup registered by NewMock end the runLoop goroutine. interrupt chan struct{} } func (t *Ticker) fire(tt time.Time) { t.mock.mu.Lock() defer t.mock.mu.Unlock() - if t.stopped { + // A fire that gets the lock after Stop or Reset belongs to the schedule they replaced: Stop + // stops the ticker, and Reset moves nxt. This happens when Stop or Reset gets the lock before + // a fire that Advance started. + if t.stopped || !t.nxt.Equal(tt) { return } for !t.nxt.After(t.mock.cur) { @@ -39,6 +46,13 @@ func (t *Ticker) fire(tt time.Time) { t.mock.recomputeNextLocked() if t.interrupt != nil { // implies runLoop is still going. t.internalTicks <- tt + return + } + // NewMock's cleanup has stopped runLoop, so only a reader that is already waiting gets the + // tick. + select { + case t.c <- tt: + default: } } @@ -61,11 +75,7 @@ func (t *Ticker) Stop(tags ...string) { defer close(c.complete) t.mock.removeEventLocked(t) t.stopped = true - // check if we've already fired, and if so, interrupt it. - if t.interrupt != nil { - <-t.interrupt - t.interrupt = nil - } + t.stopRunLoopLocked() } // Reset stops a ticker and resets its period to the specified duration. The @@ -76,6 +86,9 @@ func (t *Ticker) Reset(d time.Duration, tags ...string) { t.ticker.Reset(d) return } + if d <= 0 { + panic("Ticker.Reset called with negative or zero duration") + } t.mock.mu.Lock() defer t.mock.mu.Unlock() c := newCall(clockFunctionTickerReset, tags, withDuration(d)) @@ -91,7 +104,14 @@ func (t *Ticker) Reset(d time.Duration, tags ...string) { } if t.interrupt == nil { t.startRunLoopLocked() + return } + // Tell runLoop to drop a tick it holds for a reader, since that tick belongs to the old + // schedule. Starting a new runLoop instead would put a goroutine in the caller's synctest + // bubble that waits on channels created outside it, and synctest.Test would not return. The + // branch above still starts one, so Reset, called inside a bubble, of a stopped ticker created + // outside that bubble, hangs. + t.resets <- struct{}{} } func (t *Ticker) runLoop(interrupt chan struct{}) { @@ -100,16 +120,27 @@ outer: for { select { case tt := <-t.internalTicks: + // select picks at random among ready cases. Send first to a reader that is already + // waiting, so it gets the tick even when Stop or NewMock's cleanup is waiting on + // interrupt. + select { + case t.c <- tt: + continue outer + default: + } for { select { case t.c <- tt: continue outer case <-t.internalTicks: // Discard future ticks until we can send this one. + case <-t.resets: + continue outer case interrupt <- struct{}{}: return } } + case <-t.resets: case interrupt <- struct{}{}: return } @@ -121,11 +152,26 @@ func (t *Ticker) startRunLoopLocked() { if t.interrupt != nil { t.mock.tb.Error("called startRunLoopLocked when interrupt suggests we are already running") } + // NewMock's cleanup has run and would not stop the runLoop, so fire sends to C itself. + if t.mock.testOver { + return + } interrupt := make(chan struct{}) t.interrupt = interrupt + t.mock.tickers[t] = struct{}{} go t.runLoop(interrupt) } +// stopRunLoopLocked ends the runLoop goroutine, if it is running. +func (t *Ticker) stopRunLoopLocked() { + if t.interrupt == nil { + return + } + <-t.interrupt + t.interrupt = nil + delete(t.mock.tickers, t) +} + func newMockTickerLocked(m *Mock, d time.Duration) *Ticker { // no buffer follows Go 1.23+ behavior ticks := make(chan time.Time) @@ -136,16 +182,9 @@ func newMockTickerLocked(m *Mock, d time.Duration) *Ticker { nxt: m.cur.Add(d), mock: m, internalTicks: make(chan time.Time), + resets: make(chan struct{}), } m.addEventLocked(t) - m.tb.Cleanup(func() { - m.mu.Lock() - defer m.mu.Unlock() - if t.interrupt != nil { - <-t.interrupt - t.interrupt = nil - } - }) t.startRunLoopLocked() return t }