supervisor.go

package chronos

import (

	"context"

	"fmt"

	"log/slog"

	"math/rand/v2"

	"runtime/debug"

	"sort"

	"sync"

	"sync/atomic"

	"time"

	"go.bigb.es/auxilia/async"

	"go.bigb.es/auxilia/culpa"

	"go.bigb.es/auxilia/scribe"

)

// jobHandle is the supervisor's per-job bookkeeping: the [Job] itself,

// the cancel func for its scheduling loop (cancelled by [Supervisor.Deactivate],

// IV13), and a flag tracking whether a run is in flight (for [OverlapSkip]).

type jobHandle struct {

	job        *Job

	loopCancel context.CancelFunc

	running    atomic.Bool

}

// Supervisor owns a set of [Job]s and runs each on its own scheduling

// loop. It bounds total concurrency with a shared semaphore (IV9), drains

// in-flight runs on [Supervisor.Stop] (IV5), and never blocks a loop on a

// run, slot acquisition, or timeout (IV3). The zero value is not usable;

// construct one with [New].

type Supervisor struct {

	opts supervisorOpts

	sem  *async.Semaphore

	mu   sync.Mutex

	jobs map[string]*jobHandle

	root   context.Context

	cancel context.CancelFunc

	wg     sync.WaitGroup

	started bool

	stopped bool

	nameSeq atomic.Uint64

}

// New builds a [Supervisor] from the given [Option]s. Defaults: location

// [time.Local], logger [slog.Default], the package wall [Clock], no

// timeout, unlimited concurrency, and slot-wait 0 (shed when full).

func New(opts ...Option) *Supervisor {

	cfg := supervisorOpts{

		loc:         time.Local,

		logger:      slog.Default(),

		clock:       defaultClock,

		timeout:     0,

		maxInflight: 0,

		slotWait:    0,

		jitter:      func(max time.Duration) time.Duration { return time.Duration(rand.Int64N(int64(max))) },

	}

	for _, opt := range opts {

		opt(&cfg)

	}

	return &Supervisor{

		opts: cfg,

		sem:  async.NewSemaphore(cfg.maxInflight),

		jobs: map[string]*jobHandle{},

	}

}

// Add registers a [Job], surfacing its stashed build error (IV6) and

// rejecting duplicate names. If the supervisor is already running, the

// job's loop starts immediately (IV12).

func (s *Supervisor) Add(j *Job) error { return s.add(j) }

// Deactivate retires the named job, stopping its future fires without

// cancelling any in-flight run (IV13). It reports whether the job

// existed and is idempotent.

func (s *Supervisor) Deactivate(name string) bool { return s.deactivate(name) }

// add is the unexported seam behind [Supervisor.Add] and

// [ChronosContext.Add] (IV6, IV12).

func (s *Supervisor) add(j *Job) error {

	s.mu.Lock()

	defer s.mu.Unlock()

	if s.stopped {

		return culpa.New("chronos: supervisor stopped")

	}

	if j.buildErr != nil {

		return j.buildErr

	}

	name := j.name

	if name == "" {

		name = fmt.Sprintf("job-%d", s.nameSeq.Add(1))

		j.name = name

	}

	if _, dup := s.jobs[name]; dup {

		return culpa.Errorf("chronos: duplicate job name %q", name)

	}

	h := &jobHandle{job: j}

	s.jobs[name] = h

	s.opts.logger.Debug("chronos: job added", "job", name)

	if s.started {

		loopCtx, lc := context.WithCancel(s.root)

		h.loopCancel = lc

		s.wg.Add(1)

		go s.runLoop(h, loopCtx)

	}

	return nil

}

// deactivate is the unexported seam behind [Supervisor.Deactivate] and

// [ChronosContext.Deactivate]/[ChronosContext.DeactivateJob] (IV13).

func (s *Supervisor) deactivate(name string) bool {

	s.mu.Lock()

	h, ok := s.jobs[name]

	if !ok {

		s.mu.Unlock()

		return false

	}

	delete(s.jobs, name)

	if h.loopCancel != nil {

		h.loopCancel()

	}

	s.mu.Unlock()

	s.opts.logger.Debug("chronos: job deactivated", "job", name)

	return true

}

// removeIfSelf drops the handle from the registry only if it is still the

// one registered under its name. Called when a loop exits on a finite

// schedule, so it does not clobber a re-Added job of the same name.

func (s *Supervisor) removeIfSelf(h *jobHandle) {

	s.mu.Lock()

	if s.jobs[h.job.name] == h {

		delete(s.jobs, h.job.name)

	}

	s.mu.Unlock()

}

// logger returns the supervisor's logger; it satisfies the supervisor seam.

func (s *Supervisor) logger() *slog.Logger { return s.opts.logger }

// Jobs returns the names of the registered jobs, sorted.

func (s *Supervisor) Jobs() []string {

	s.mu.Lock()

	names := make([]string, 0, len(s.jobs))

	for name := range s.jobs {

		names = append(names, name)

	}

	s.mu.Unlock()

	sort.Strings(names)

	return names

}

// Start begins scheduling every registered job, deriving the run root

// from ctx (PC2). It errors if already started.

func (s *Supervisor) Start(ctx context.Context) error {

	s.mu.Lock()

	defer s.mu.Unlock()

	if s.started {

		return culpa.New("chronos: supervisor already started")

	}

	s.started = true

	s.root, s.cancel = context.WithCancel(ctx)

	for _, h := range s.jobs {

		loopCtx, lc := context.WithCancel(s.root)

		h.loopCancel = lc

		s.wg.Add(1)

		go s.runLoop(h, loopCtx)

	}

	return nil

}

// Stop cancels every loop and run, then waits for in-flight runs to drain,

// bounded by ctx (IV5, PC2). It is idempotent and returns ctx.Err() if the

// drain outlasts ctx.

func (s *Supervisor) Stop(ctx context.Context) error {

	s.mu.Lock()

	if !s.started || s.stopped {

		s.mu.Unlock()

		return nil

	}

	s.stopped = true

	s.cancel()

	s.mu.Unlock()

	done := make(chan struct{})

	go func() {

		s.wg.Wait()

		close(done)

	}()

	select {

	case <-done:

		return nil

	case <-ctx.Done():

		return ctx.Err()

	}

}

// runLoop schedules and dispatches fires for a single job until its loop

// context is cancelled or its schedule is exhausted. It dispatches each

// fire on its own goroutine so the loop never blocks (IV3, AS3).

// GPC2: exceeds ~10 lines but is inherently one cohesive scheduling loop.

func (s *Supervisor) runLoop(h *jobHandle, loopCtx context.Context) {

	defer s.wg.Done()

	prev := s.opts.clock.Now().In(s.opts.loc)

	for {

		next := h.job.sched.Next(prev)

		if next.IsZero() {

			h.loopCancel() // schedule exhausted — release the loop context (IV5)

			s.removeIfSelf(h)

			return

		}

		if !next.After(prev) {

			// Defensive IV1 guard: a Schedule that fails to advance would

			// busy-loop. Stop the job loudly rather than spin.

			s.opts.logger.Error("chronos: schedule did not advance; stopping job",

				"job", h.job.name, "prev", prev, "next", next)

			h.loopCancel()

			s.removeIfSelf(h)

			return

		}

		delay := next.Sub(s.opts.clock.Now())

		if h.job.delta > 0 {

			delay += s.opts.jitter(h.job.delta)

		}

		if delay < 0 {

			delay = 0

		}

		timer := s.opts.clock.NewTimer(delay)

		select {

		case <-loopCtx.Done():

			timer.Stop()

			return

		case <-timer.C():

		}

		s.wg.Add(1)

		go s.fire(h, next)

		prev = next // anchor on the scheduled (not jittered) time, so no drift

	}

}

// fire executes one run of a job: it enforces the overlap policy, acquires

// a global slot per the resolved slot-wait, applies the resolved timeout,

// runs the function, and logs the outcome. Every acquired resource is

// released on every path, including panic (IV7, IV9).

// GPC2: exceeds ~10 lines but is inherently one cohesive run pipeline.

func (s *Supervisor) fire(h *jobHandle, scheduled time.Time) {

	defer s.wg.Done()

	log := s.opts.logger.With("job", h.job.name, "fire", scheduled)

	if h.job.overlap == OverlapSkip {

		if !h.running.CompareAndSwap(false, true) {

			log.Warn("chronos: overlap skip")

			return

		}

		defer h.running.Store(false)

	}

	// Acquire a global slot, interruptible by root cancellation (Stop) (IV9, IV10).

	wait := h.job.slotWait

	if !h.job.slotWaitSet {

		wait = s.opts.slotWait

	}

	var acquired bool

	switch {

	case wait == 0:

		acquired = s.sem.TryAcquire()

	case wait < 0:

		acquired = s.sem.Acquire(s.root) == nil

	default:

		actx, c := context.WithTimeout(s.root, wait)

		acquired = s.sem.Acquire(actx) == nil

		c()

	}

	if !acquired {

		log.Warn("chronos: cap skip")

		return

	}

	defer s.sem.Release()

	// The timeout clock starts only after the slot is acquired (IV8).

	to := h.job.timeout

	if !h.job.timeoutSet {

		to = s.opts.timeout

	}

	var runCtx context.Context

	var cancel context.CancelFunc

	if to > 0 {

		runCtx, cancel = context.WithTimeout(s.root, to)

	} else {

		runCtx, cancel = context.WithCancel(s.root)

	}

	defer cancel()

	start := s.opts.clock.Now()

	defer func() {

		if r := recover(); r != nil {

			log.Error("chronos: job panic", "panic", r, "stack", string(debug.Stack())) // IV7

		}

	}()

	err := h.job.fn(&ChronosContext{Context: runCtx, sup: s, name: h.job.name})

	switch {

	case runCtx.Err() == context.DeadlineExceeded:

		// Build a non-nil base even when the run returned nil, then tag it (IV8).

		base := culpa.Wrapf(err, "job %q timed out", h.job.name)

		if base == nil {

			base = culpa.Errorf("job %q timed out", h.job.name)

		}

		log.Warn("chronos: job timeout", scribe.Err(culpa.WithTimeout(base)))

	case err != nil:

		log.Error("chronos: job error", scribe.Err(culpa.Wrap(err, "job failed"))) // PC4

	default:

		log.Debug("chronos: job complete", "dur", s.opts.clock.Now().Sub(start))

	}

}