tasks/pkg/sdk/workflow/timer.go

// Package workflow — wall-clock Timer implementation.
//
// Phase 1: Timers are backed by a real time.Timer on a per-timer
// goroutine. That goroutine settles the Future when the deadline
// fires or the scope cancels. Duration is wall-clock; Now() advances
// only deterministically (via AdvanceClock / AutoAdvance) so there is
// no attempt to align "workflow time" with "timer fire time" yet.
//
// Phase 2 (event-time replay, durable timers): the env will persist
// a pending-timer record in the history log, and replay will refire
// the same Future from the log at the same point. The public
// workflow.NewTimer API does not change.
//
// Wall-clock-only v1 is explicitly fine — the read-team §5.2 bug
// was the opposite: NewTimer collapsed every duration to zero, so a
// 24h dunning timer fired on the next dispatch loop tick. Wall-clock
// correctness eliminates that class of bug entirely.

package workflow

import (
	"time"

	"github.com/hanzoai/tasks/pkg/sdk/temporal"
)

// NewTimer schedules a durable timer that fires after d of workflow
// time and returns the Future that represents its completion. The
// Future's Get settles with nil error when the timer fires, or with
// temporal.Canceled if the ctx is canceled before d elapses.
//
// Mirrors go.temporal.io/sdk/workflow.NewTimer.
func NewTimer(ctx Context, d time.Duration) Future {
	if ctx == nil {
		// Return an already-failed future so callers don't crash.
		f := NewFuture()
		f.Settle(nil, errNilContext)
		return f
	}
	if d < 0 {
		d = 0
	}
	return ctx.env().NewTimer(d)
}

// NewWallClockTimer is the shared helper used by both StubEnv and
// workerEnv to produce a real, cancellable Future driven by a
// time.Timer. Exported so the worker package (a sibling) can reuse
// the same implementation without reaching into package internals.
//
// - d <= 0 settles the Future with (nil, nil) immediately.
// - Otherwise, a per-timer goroutine parks on the timer's channel
//   alongside the supplied doneCh. If doneCh closes first, the Future
//   settles with errFn() — typically the scope's cancel error.
//
// Cancellation semantics: a closed doneCh always wins over a fired
// timer in the rare race where both are observable simultaneously.
func NewWallClockTimer(d time.Duration, doneCh <-chan struct{}, errFn func() error) Future {
	f := NewFuture()
	if d <= 0 {
		f.Settle(nil, nil)
		return f
	}
	go func() {
		t := time.NewTimer(d)
		defer t.Stop()
		select {
		case <-doneCh:
			err := errFn()
			if err == nil {
				err = temporal.NewCanceledError()
			}
			f.Settle(nil, err)
		case <-t.C:
			// Re-check cancellation: if both fired in the same window
			// prefer cancellation so a cancel never "loses" to a timer.
			select {
			case <-doneCh:
				err := errFn()
				if err == nil {
					err = temporal.NewCanceledError()
				}
				f.Settle(nil, err)
			default:
				f.Settle(nil, nil)
			}
		}
	}()
	return f
}

// SelectFanIn is the shared edge-triggered Select implementation.
// Exported for the worker env to reuse without duplication. User
// code should not call it directly — use workflow.NewSelector.
//
// It blocks until exactly one case in cases is ready, or until
// doneCh closes, and returns the ready case index (or -1 on
// cancellation / empty input).
//
// Implementation: each case parks on a readiness source —
//   - Future cases: the Future.ReadyCh() (closed on Settle).
//   - chanImpl cases: a single-buffer chan struct{} registered via
//     RegisterReadyWaker; the channel fires once when the chan
//     becomes readable (buffered value, parked unbuffered sender, or
//     closed).
//
// A small goroutine per case translates that signal into an index
// send on wakeCh. The main routine receives once from wakeCh and
// returns. On wake we re-scan for the earliest-ready case to preserve
// deterministic ordering across unrelated simultaneous ready events.
//
// No polling. No time.Ticker. No 1 ms spin.
func SelectFanIn(cases []SelectCase, doneCh <-chan struct{}) int {
	if len(cases) == 0 {
		return -1
	}

	// Fast path: if any case is already ready, fire it without spinning
	// up goroutines.
	if idx := readyIndex(cases); idx >= 0 {
		return idx
	}

	wakeCh := make(chan int, len(cases)+1)
	stopCh := make(chan struct{})
	defer close(stopCh)

	for i, c := range cases {
		i, c := i, c
		switch {
		case c.Future != nil:
			ready := c.Future.ReadyCh()
			go func() {
				select {
				case <-ready:
					select {
					case wakeCh <- i:
					case <-stopCh:
					}
				case <-stopCh:
				}
			}()
		case c.Channel != nil:
			if ch, ok := c.Channel.(*chanImpl); ok {
				w := make(chan struct{}, 1)
				ch.RegisterReadyWaker(w)
				go func() {
					select {
					case <-w:
						select {
						case wakeCh <- i:
						case <-stopCh:
						}
					case <-stopCh:
					}
				}()
			} else {
				// Unknown channel shape: fall back to a tiny poll
				// (should not occur — all channels flow through chanImpl).
				go func() {
					ticker := time.NewTicker(50 * time.Millisecond)
					defer ticker.Stop()
					for {
						select {
						case <-stopCh:
							return
						case <-ticker.C:
							if c.Channel != nil {
								if hv, ok := c.Channel.(interface{ HasValue() bool }); ok && hv.HasValue() {
									select {
									case wakeCh <- i:
									case <-stopCh:
									}
									return
								}
							}
						}
					}
				}()
			}
		}
	}

	select {
	case idx := <-wakeCh:
		if idx < 0 || idx >= len(cases) {
			return -1
		}
		// Prefer the earliest-ready case for deterministic ordering
		// when multiple cases became ready very close together.
		if better := readyIndex(cases); better >= 0 && better < idx {
			return better
		}
		return idx
	case <-doneCh:
		return -1
	}
}