[runner] fix MonotonicTimestamp

runner/internal/executor/timestamp.go

@@ -9,37 +9,52 @@ import (
 )
 
 type MonotonicTimestamp struct {
-	unix    int64
-	counter int
-	mu      sync.RWMutex
+	initial     time.Time
+	initialUnix int64 // seconds
+	elapsed     int64 // seconds since initial
+	counter     int   // surrogate milliseconds
+	overflow    bool
+	mu          sync.RWMutex
+	getNow      func() time.Time
 }
 
 func NewMonotonicTimestamp() *MonotonicTimestamp {
+	return newMonotonicTimestamp(time.Now)
+}
+
+func newMonotonicTimestamp(getNow func() time.Time) *MonotonicTimestamp {
+	// getNow must return time.Time with monotonic reading
+	now := getNow()
 	return &MonotonicTimestamp{
-		unix:    time.Now().Unix(),
-		counter: 0,
-		mu:      sync.RWMutex{},
+		initial:     now,
+		initialUnix: now.Unix(),
+		mu:          sync.RWMutex{},
+		getNow:      getNow,
 	}
 }
 
 func (t *MonotonicTimestamp) GetLatest() int64 {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
-	return t.unix*1000 + int64(t.counter)
+	return (t.initialUnix+t.elapsed)*1000 + int64(t.counter)
 }
 
 func (t *MonotonicTimestamp) Next() int64 {
-	// warning: time.Now() is not monotonic in general
 	t.mu.Lock()
-	now := time.Now().Unix()
-	if now == t.unix {
-		t.counter++
-		if t.counter == 1000 {
-			log.Warning(context.TODO(), "Monotonic timestamp counter overflowed", "timestamp", now)
+	now := t.getNow()
+	elapsed := int64(now.Sub(t.initial) / time.Second)
+	if elapsed == t.elapsed {
+		if t.counter < 999 {
+			t.counter++
+		} else if !t.overflow {
+			// warn only once per second to avoid log spamming
+			log.Warning(context.TODO(), "Monotonic timestamp counter overflowed", "unix", t.initialUnix+elapsed)
+			t.overflow = true
 		}
 	} else {
-		t.unix = now
+		t.elapsed = elapsed
 		t.counter = 0
+		t.overflow = false
 	}
 	t.mu.Unlock()
 	return t.GetLatest()

runner/internal/executor/timestamp_test.go

@@ -0,0 +1,43 @@
+package executor
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTimestamp_Counter(t *testing.T) {
+	now := time.Now()
+	ts := newMonotonicTimestamp(func() time.Time { return now })
+	initial := ts.GetLatest()
+	assert.Equal(t, int64(1), ts.Next()-initial)
+	assert.Equal(t, int64(2), ts.Next()-initial)
+	now = now.Add(999 * time.Millisecond)
+	assert.Equal(t, int64(3), ts.Next()-initial)
+	now = now.Add(100 * time.Millisecond)
+	assert.Equal(t, int64(1000), ts.Next()-initial)
+	assert.Equal(t, int64(1001), ts.Next()-initial)
+}
+
+func TestTimestamp_CounterOverflow(t *testing.T) {
+	now := time.Now()
+	ts := newMonotonicTimestamp(func() time.Time { return now })
+	initial := ts.GetLatest()
+	for i := 0; i < 997; i++ {
+		ts.Next()
+	}
+	assert.Equal(t, int64(998), ts.Next()-initial)
+	assert.False(t, ts.overflow)
+	assert.Equal(t, int64(999), ts.Next()-initial)
+	assert.False(t, ts.overflow)
+	assert.Equal(t, int64(999), ts.Next()-initial)
+	assert.True(t, ts.overflow)
+	assert.Equal(t, int64(999), ts.Next()-initial)
+	assert.True(t, ts.overflow)
+	now = now.Add(time.Second)
+	assert.Equal(t, int64(1000), ts.Next()-initial)
+	assert.False(t, ts.overflow)
+	assert.Equal(t, int64(1001), ts.Next()-initial)
+	assert.False(t, ts.overflow)
+}