scheduler

README

scheduler

Scheduling scaffold for containerized job runners

A standalone Go library of small, composable primitives for a container that runs a job on an interval or an external trigger: interval parsing with the standard sentinels, a startup-plus-ticker run loop with jitter that drains on shutdown, an advisory flock overlap guard, a SIGTERM-graceful subprocess runner, and the trigger subpackage's single-owner trigger broker (bounded FIFO queue, owner-only unix-socket server, thin synchronous client, opt-in executor loop) for daemons where PID 1 owns every run. Standard library only (test dependency: pgregory.net/rapid). Unix-only (the overlap guard is flock(2)).

It is a toolbox, not a framework: each primitive is independent, and the composition root wires the ones it needs. The library says nothing about what a job does, how health is signaled, or how logging is configured; those stay in the app.

Install

go get github.com/cplieger/scheduler/v4@latest

Usage

A typical composition root reads an interval variable, picks a mode, and drives the job, guarding overlap and shutting down gracefully:

package main

import (
	"context"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/cplieger/scheduler/v4"
)

const lockPath = "/tmp/.myjob.lock"

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	sched := scheduler.ParseInterval(os.Getenv("JOB_INTERVAL"), 6*time.Hour,
		scheduler.WithName("JOB_INTERVAL"))

	switch sched.Mode {
	case scheduler.ModeBuiltin:
		// Fire once now, then every interval (with ±10% jitter), draining on SIGTERM.
		scheduler.RunLoop(ctx, runPass, scheduler.LoopOptions{
			Interval:    sched.Interval,
			FireOnStart: true,
			Jitter:      0.10,
		})
	case scheduler.ModeExternal:
		// Idle: runs are triggered out-of-band (an Ofelia docker-exec of a
		// one-shot subcommand); the lock below keeps them from overlapping. A
		// daemon that must itself wait out or cancel externally triggered runs
		// should own execution instead; see the trigger subpackage.
		<-ctx.Done()
	case scheduler.ModeOnce:
		runPass(ctx) // run exactly once, then exit
	}
}

// run builds context-cancellable subprocesses that get SIGTERM (not SIGKILL)
// on shutdown, with a grace period before the kill.
var run = scheduler.NewCommandRunner(scheduler.DefaultGrace)

func runPass(ctx context.Context) {
	lock, ok, err := scheduler.TryLock(lockPath)
	if err != nil {
		return // could not acquire; mark unhealthy in a real app
	}
	if !ok {
		return // another run already in flight; the overlap guard skips this one
	}
	defer lock.Unlock()

	cmd := run(ctx, "rsync", "-a", "/src/", "remote:/dst/")
	cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
	_ = cmd.Run()
}

Interval parsing

ParseInterval applies the standard sentinel and fallback rules to a *_INTERVAL environment value and returns a Schedule (cadence + Mode):

Raw value	Result
"30m", "1h30m" (positive Go duration)	ModeBuiltin, that cadence (clamped by WithBounds)
"" (unset)	ModeBuiltin, the default cadence
"off", "disabled" (case-insensitive across ASCII letters only)	ModeExternal
"0", "0s" (zero)	ModeExternal, or ModeOnce with WithZeroAsOnce(true)
"-1h" (negative)	ModeBuiltin at default + a warning (a likely typo)
"banana" (unparseable)	ModeBuiltin at default + a warning

Options: WithZeroAsOnce(zeroAsOnce) (true treats a zero duration as run-once; false keeps the default), WithBounds(low, high) (clamp a positive cadence), WithName(env) (name the variable in warnings), WithIntervalLogger(l) (route warnings to a specific logger; defaults to slog.Default()), and WithRedactedValue(redacted) (true keeps the supplied raw value out of every warning; false keeps the default echo). Repeated applications of a bool option resolve last-wins. Pass WithRedactedValue(true) when the interval passes through secret-capable config expansion, where a typo could place an expanded secret in the field; plain env-var reads should keep the default echo, which is useful diagnostics.

Overlap guard and coalescing

TryLock / Unlock serialize runs across both the in-process loop and an out-of-band docker exec trigger. ReadHolder reads when the current holder acquired the lock (observability only). A trigger that arrives mid-run is not dropped: Exclusive (next section) queues it as a coalesced rerun.

Run coalescing across processes

Exclusive packages the lock + queue pattern into cross-process run coalescing for a whole app: at most one cycle runs at a time per instance, across every entry point (the resident daemon's tick, a poll subcommand exec'd by an operator or an external scheduler). A request that arrives while a cycle runs is queued without blocking the requester: it records a rerun request in a counter file and exits immediately, and the active runner executes the queued demand when the current run finishes. Requests beyond the queue capacity (default 1, set with WithQueueCapacity) are discarded, because the queued rerun already guarantees a run starts after they arrived.

The two entry points pair as queue mode for demand-driven callers and skip mode for time-driven ticks:

ex := scheduler.NewExclusive("/config", logger)

// Daemon: RunLoop ticks use skip mode: a busy lock means the job is already
// running, and the next tick provides freshness; never queue a tick.
scheduler.RunLoop(ctx, func(ctx context.Context) {
	_, _ = ex.RunOrSkip(func() error { return runCycle(ctx) })
}, scheduler.LoopOptions{Interval: sched.Interval, FireOnStart: true})

// Poll subcommand (exec'd by an operator or an external scheduler): queue
// mode: the request must be satisfied by a run that starts after it arrived.
outcome, err := ex.Run(func() error { return runCycle(ctx) })
switch outcome {
case scheduler.OutcomeQueued, scheduler.OutcomeDiscarded:
	os.Exit(0) // the in-flight runner covers this request; nothing to wait for
default:
	if err != nil {
		os.Exit(1)
	}
}

The lock is a flock(2) (cycle.lock in the directory), so the kernel releases it when the holding process dies: a crashed run never wedges the scheduler, and a queue counter orphaned by a crash is cleared at the next acquisition. Pending reports the queued-request count for observability, and ReadHolder on ExclusiveLockName reports when the current cycle started.

Three policy edges are deliberate, and all deferral is demand-preserving (the queue counter survives; the next run satisfies it):

• A failed run does not stop queued demand: each queued request is owed a run, succeed or fail, so the consume loop continues through job errors (bounded by the rerun cap below) instead of dropping demand against a failing job.
• WithGate(func() bool) puts the composition root's shutdown signal (typically the shutdown context's Err) in front of every run start: a gated initial run returns OutcomeGated (cycle gate closed; skipping run), and queued demand behind a closed gate defers (cycle gate closed; deferring queued demand). An in-flight run is never interrupted, and a stop request is never followed by a fresh run.
• A holder executes at most 8 queued reruns per acquisition: past that cap it retires (warning rerun cap reached; deferring queued demand), so a relentless trigger source cannot pin one holder indefinitely. Each rerun's running queued cycle request line carries an attempt ordinal for log attribution.

The storage under the queue counter is exported as SlotFile: a single-slot byte payload shared across processes through one file, mutated by atomic read-modify-write transactions under a short exclusive flock on the file itself. Build on it when your coalescing state needs a payload the counter cannot carry. The bytes' meaning, how concurrent demands merge, and when recorded demand counts as served stay the caller's policy; SlotFile owns only the transaction (create-on-first-use, blocking lock, skip-if-unchanged write, never unlink a live slot).

Single-owner trigger broker (trigger subpackage)

Where Exclusive coordinates runs across processes, the trigger subpackage is the in-process alternative for daemons that own execution outright: PID 1 executes every run as its own child, and triggers (the built-in ticker, each docker exec'd subcommand) only submit requests. One bounded FIFO trigger.Queue feeds one executor goroutine (mutual exclusion is that loop), a trigger.Server accepts requests on an owner-only in-container unix socket and streams queued/started/done events back, and trigger.Submit is the thin synchronous client a trigger subcommand wraps. No coalescing: every accepted request gets its own run, its own arguments, and its own true result, in arrival order.

The request payload is a type parameter: a daemon whose runs take arguments declares a struct (repo slugs plus a forwarded environment, say); an argless daemon uses struct{}, which frames as {} on the wire.

// Daemon side: one queue, one executor goroutine, one socket server.
// trigger.Execute owns the Start/Finish lifecycle, so the exactly-one-result
// contract is structural: the callback just returns the outcome.
queue := trigger.NewQueue[payload](16)
go func() {
	trigger.Execute(ctx, queue, func(ctx context.Context, trig string, p payload) trigger.Outcome {
		ok, elapsed := runPass(ctx, p) // the app's real work
		return trigger.Outcome{OK: ok, Duration: elapsed}
	})
}()
ln, err := trigger.Listen("/tmp/myapp.sock") // owner-only, stale file unlinked
srv := &trigger.Server[payload]{Queue: queue}
srv.Serve(ln)
// shutdown: ln.Close(); queue.Close(); Execute drains and returns; srv.Wait()

// Trigger subcommand: submit one run, wait for its own result.
// Pass signal.NotifyContext so Ctrl-C unwinds the wait instead of killing
// the process with the connection half-open.
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
final, err := trigger.Submit(ctx, "/tmp/myapp.sock", payload{Repos: repos}, nil)
// map final.OK / errors.Is(err, trigger.ErrUnreachable) to the exit code

The queue rejects fast when full or closing (ErrFull, ErrClosed; their messages travel the wire as the rejection reason). An accepted job is guaranteed exactly one result: Execute finishes jobs received after shutdown with CancelledReason instead of dropping them, and delivers a panicking run's failure result before propagating the panic, so a waiting client is never stranded. A daemon whose executor policy diverges (running jobs outside the shutdown context, halting admission on an app state, its own cancellation vocabulary) writes the ~7-line loop by hand instead; Execute is an opt-in mechanism, not a required framework. The server never logs payload contents (a forwarded environment can carry secrets; the OnAccepted/OnRejected hooks exist so the app logs acceptance in its own vocabulary). What a job does, how its outcome maps to health, and the exact wording of lifecycle log lines stay in the app, the same mechanism-vs-policy split as SlotFile.

API

• Mode: ModeBuiltin, ModeExternal, ModeOnce (implements fmt.Stringer).
• Schedule: {Interval, Mode} returned by ParseInterval.
• ParseInterval(raw string, def time.Duration, opts ...IntervalOption) Schedule.
• WithZeroAsOnce(zeroAsOnce bool), WithBounds(low, high), WithName(name), WithIntervalLogger(l), WithRedactedValue(redacted bool): interval options.
• Job: func(ctx context.Context), one unit of scheduled work.
• LoopOptions: {Interval, Jitter, FireOnStart}.
• RunLoop(ctx, job, opts): sequential startup-plus-ticker loop; drains on cancellation.
• JitteredDelay(interval, fraction) time.Duration: the pure ±band jitter core.
• Lock, TryLock(path) (*Lock, bool, error), (*Lock).Unlock(), ReadHolder(path) (time.Time, bool).
• Exclusive, NewExclusive(dir, logger, opts...), .Run(job) (Outcome, error) (queue mode), .RunOrSkip(job) (Outcome, error) (skip mode), .Pending() (int, error): cross-process run coalescing.
• WithQueueCapacity(n), WithGate(func() bool): Exclusive options for queue depth (default 1) and a pre-run shutdown gate.
• SlotFile, NewSlotFile(path), .Mutate(fn func(before []byte) []byte) ([]byte, error): the flock'd single-slot read-modify-write transaction behind Exclusive's counter, exported for app-defined coalescing payloads.
• Outcome: OutcomeRan, OutcomeRanQueued, OutcomeQueued, OutcomeDiscarded, OutcomeSkipped, OutcomeGated, OutcomeNone (implements fmt.Stringer).
• ExclusiveLockName, ExclusiveQueueName: the file names Exclusive maintains inside its directory.
• CommandRunner, NewCommandRunner(grace) CommandRunner, DefaultGrace.

Subpackage trigger (the single-owner broker):

• Queue[P], NewQueue[P](capacity), .Submit(*Job[P]) error, .Jobs() <-chan *Job[P], .Close(): the bounded FIFO; ErrFull, ErrClosed.
• Job[P], NewJob[P](trigger, payload), .Start(), .Started(), .Finish(Outcome), .Result(), TriggerExternal: one request and its exactly-one-result lifecycle.
• Execute[P](ctx, queue, run func(ctx, trigger, payload) Outcome): the opt-in executor loop that owns Start/Finish structurally; CancelledReason is the outcome reason for jobs cancelled by shutdown before starting.
• Outcome: {OK, Reason, Duration}, a job's final result.
• Listen(path) (net.Listener, error): owner-only unix socket with stale-file hygiene.
• Server[P]: {Queue, OnAccepted, OnRejected}, .Serve(ln), .Wait(); streams Event lines per connection.
• Event: {Kind, Reason, DurationMs, OK}; kinds EventQueued, EventStarted, EventDone.
• Submit[P](ctx, socketPath, payload, onEvent) (Event, error): the synchronous client; ErrUnreachable, ErrSend, ErrConnectionLost; DialTimeout. Cancelling ctx stops the wait.

Unsupported by Design

These are deliberate non-goals, not a TODO list. The library is one cohesive concept (schedule a container job, guard its overlap, run and drain it) and stays small on purpose.

Feature	Rationale
Logging setup (slog handler, UTC time attr)	The composition root owns logging. The library logs interval warnings through slog.Default() (or WithIntervalLogger) and Exclusive's coalescing lines through its injected logger (nil falls back to slog.Default()); it never configures a handler.
Health signaling	Set(healthy) is the app's call inside its job. Use the companion health library for the marker; the two compose.
What a job does / its outcome type	Job is func(ctx). Exit codes, health flips, and log lines are the app's policy, wired inside the closure.
Cron expressions / calendar schedules	This is interval + external-trigger scheduling. For 0 2 * * * semantics, use an external scheduler (Ofelia, cron) in ModeExternal.
Distributed / multi-node coordination	The flock guard is single-host. Cross-node leader election is a different abstraction (a lease store), out of scope.
Concurrent in-process runs	RunLoop is sequential by design (two runs never overlap in-process); the flock guards the cross-process case. Run a job concurrently yourself if you must.
Retry / backoff of a failed run	Retrying outbound work belongs to httpx; a failed pass is reported by the job and retried on the next tick or trigger.

Contributing

Issues and PRs are welcome. See CONTRIBUTING.md for the conventions and how to run the checks locally.

Disclaimer

This project is built with care and follows security best practices, but it is intended for personal / self-hosted use. No guarantees of fitness for production environments. Use at your own risk.

This project was built with AI-assisted tooling using Claude, GPT, and Kiro. The human maintainer defines architecture, supervises implementation, and makes all final decisions.

License

Apache-2.0. See LICENSE.

Documentation

Overview

Package scheduler is the scheduling scaffold shared by the fleet's containerized job runners.

It provides small, orthogonal primitives — not a framework — that a composition root wires together:

• ParseInterval turns a *_INTERVAL environment value into a Schedule (a cadence plus a Mode: built-in, external, or once), applying the standard off/disabled/0 sentinel and fallback rules.
• RunLoop drives the built-in mode: a startup fire plus a jittered interval ticker that drains on context cancellation. JitteredDelay is its pure, testable core.
• TryLock / Unlock / ReadHolder are an advisory flock(2) overlap guard so a run and an out-of-band trigger never execute two jobs at once (a trigger arriving mid-run is queued by Exclusive, below).
• Exclusive composes the pieces above into packaged cross-process run coalescing for whole cycles: at most one cycle runs at a time across processes, a requester that finds a run in flight queues a rerun request (bounded, no blocked waiters) or skips its tick, and the runner executes the queued demand when the current run finishes — continuing through job errors (a queued request is owed a run, succeed or fail) and retiring at a rerun cap that defers a storm's residue to the next run. WithGate wires the composition root's shutdown signal in front of every run start, so a stop request is never followed by a fresh run.
• SlotFile is the storage mechanism under Exclusive's counter — a single-slot byte payload mutated by flock'd read-modify-write transactions — exported for apps whose coalescing state carries a payload (the payload's meaning and merge/claim policy stay app-side).
• NewCommandRunner builds context-cancellable subprocesses that shut down gracefully (SIGTERM with a grace period before SIGKILL).

The trigger subpackage is the in-process alternative to the flock-based coordination above, for single-owner daemons: PID 1 owns every run, and triggers submit requests through a bounded FIFO queue served over an owner-only in-container unix socket (see package trigger).

The package is deliberately silent about what a job does, how health is signaled, and how logging is configured; those belong to the consuming app (see the companion health library for the marker pattern). It carries no runtime dependencies beyond the standard library, and its flock-based primitives are Unix-only.

Index

• Constants
• func JitteredDelay(interval time.Duration, fraction float64) time.Duration
• func ReadHolder(path string) (since time.Time, known bool)
• func RunLoop(ctx context.Context, job Job, opts LoopOptions)
• type CommandRunner
  • func NewCommandRunner(grace time.Duration) CommandRunner
• type Exclusive
  • func NewExclusive(dir string, log *slog.Logger, opts ...ExclusiveOption) *Exclusive
  • func (e *Exclusive) Pending() (int, error)
  • func (e *Exclusive) Run(job func() error) (Outcome, error)
  • func (e *Exclusive) RunOrSkip(job func() error) (Outcome, error)
• type ExclusiveOption
  • func WithGate(gate func() bool) ExclusiveOption
  • func WithQueueCapacity(n int) ExclusiveOption
• type IntervalOption
  • func WithBounds(low, high time.Duration) IntervalOption
  • func WithIntervalLogger(l *slog.Logger) IntervalOption
  • func WithName(name string) IntervalOption
  • func WithRedactedValue(redacted bool) IntervalOption
  • func WithZeroAsOnce(zeroAsOnce bool) IntervalOption
• type Job
• type Lock
  • func TryLock(path string) (l *Lock, ok bool, err error)
  • func (l *Lock) Unlock()
• type LoopOptions
• type Mode
  • func (m Mode) String() string
• type Outcome
  • func (o Outcome) String() string
• type Schedule
  • func ParseInterval(raw string, def time.Duration, opts ...IntervalOption) Schedule
• type SlotFile
  • func NewSlotFile(path string) *SlotFile
  • func (s *SlotFile) Mutate(fn func(before []byte) []byte) (before []byte, err error)

Examples

• Exclusive
• ParseInterval
• TryLock

Constants

const (
	// ExclusiveLockName is the flock(2) file serializing cycle runs.
	ExclusiveLockName = "cycle.lock"
	// ExclusiveQueueName is the counter file holding the number of queued
	// rerun requests.
	ExclusiveQueueName = "cycle.queued"
)

Names of the two files Exclusive maintains inside its directory. They are exported so a consumer can point observability tooling at them — for example ReadHolder(filepath.Join(dir, ExclusiveLockName)) to report how long the current cycle has been running.

const DefaultGrace = 5 * time.Second

DefaultGrace is the graceful-shutdown grace period the scheduling apps use: on context cancellation the child is sent SIGTERM and given this long to exit before os/exec escalates to SIGKILL.

Functions

func JitteredDelay

func JitteredDelay(interval time.Duration, fraction float64) time.Duration

JitteredDelay returns a delay drawn uniformly from [interval−fraction×interval, interval+fraction×interval). It is the pure core of RunLoop's jitter, exported so the ±band can be tested directly and reused. A non-positive fraction or interval returns interval unchanged.

func ReadHolder

func ReadHolder(path string) (since time.Time, known bool)

ReadHolder reads the acquisition timestamp the current holder recorded in the lock file. known is false when the timestamp could not be read — the holder had not written it yet, the line was torn mid-write, or the file is absent — in which case since is the zero time. The value is observability-only (a contender reporting how long the holder has run) and never affects locking correctness; it is meaningful only while the lock is actually held.

func RunLoop

func RunLoop(ctx context.Context, job Job, opts LoopOptions)

RunLoop runs job on a schedule until ctx is cancelled. The job runs sequentially in the loop, so two invocations never overlap in-process; guard cross-process overlap (an external trigger racing the loop) with TryLock inside the job. RunLoop blocks until ctx is cancelled and the in-flight job (if any) returns, so a caller can treat its return as a completed drain.

RunLoop covers the built-in scheduling mode only. A one-shot (ModeOnce) job is run directly by the caller; an idle (ModeExternal) container simply waits on ctx.Done. RunLoop returns immediately if Interval is not positive.

Types

type CommandRunner

type CommandRunner func(ctx context.Context, name string, args ...string) *exec.Cmd

CommandRunner constructs a configured *exec.Cmd for a context and argument vector. It decouples job orchestration from subprocess construction so tests can inject a fake runner; NewCommandRunner returns the production one.

func NewCommandRunner

func NewCommandRunner(grace time.Duration) CommandRunner

NewCommandRunner returns a CommandRunner that builds a context-cancellable command with graceful shutdown: on cancellation the child is sent SIGTERM (rather than os/exec's default SIGKILL) and given grace before the SIGKILL escalation. The caller wires Stdout/Stderr on the returned command — capture them into buffers, or stream them to os.Stdout/os.Stderr — before calling Run. A non-positive grace uses DefaultGrace.

type Exclusive

type Exclusive struct {
	// contains filtered or unexported fields
}

Exclusive coordinates cycle runs across processes so that at most one runs at a time, with a small queue of pending rerun requests instead of blocked waiters — packaged cross-process run coalescing for callers that are themselves short-lived processes (a poll subcommand exec'd by an operator or an external scheduler racing the resident daemon).

The mechanics: a runner holds a flock(2) on dir/cycle.lock for the whole job; a requester that finds the lock busy increments the counter in dir/cycle.queued and exits immediately (never blocking for the job's duration); the runner consumes the counter at job end, rerunning once per queued request until none remain. Because the lock is an flock, the kernel releases it when the holding process dies — there is no stale-lock state — and a queue counter orphaned by a crash is cleared at the next acquisition (the run about to start satisfies the demand it recorded).

A holder executes at most maxCoalescedReruns (8) queued reruns per acquisition; demand still pending past that cap is deferred — it stays in the counter file and the next acquisition's run satisfies it — so a relentless trigger source cannot pin one holder indefinitely. WithGate bounds the holder further: it stops runs from starting once the composition root's shutdown signal trips. Deferral is always demand-preserving.

Both files live in dir and are created on first use; they are never deleted (clearing the queue writes a zero count — unlinking a locked file would let a concurrent opener land on a different inode and break mutual exclusion). Place dir where untrusted local users cannot write, per the same symlink-following caveat as TryLock.

Example

package main

import (
	"fmt"
	"log/slog"
	"os"

	"github.com/cplieger/scheduler/v4"
)

func main() {
	dir, err := os.MkdirTemp("", "scheduler_example_exclusive")
	if err != nil {
		return
	}
	defer func() { _ = os.RemoveAll(dir) }()

	ex := scheduler.NewExclusive(dir, slog.New(slog.DiscardHandler))
	outcome, _ := ex.Run(func() error {
		fmt.Println("cycle ran")
		return nil
	})
	fmt.Println("outcome:", outcome)
}

Output:
cycle ran
outcome: ran

func NewExclusive

func NewExclusive(dir string, log *slog.Logger, opts ...ExclusiveOption) *Exclusive

NewExclusive returns an Exclusive coordinating runs through lock and queue files inside dir (which must exist). A nil log falls back to slog.Default() at call time. Options: WithQueueCapacity, WithGate.

func (*Exclusive) Pending

func (e *Exclusive) Pending() (int, error)

Pending reports the number of currently queued rerun requests. It is observability-only: the value can change the moment it is read.

func (*Exclusive) Run

func (e *Exclusive) Run(job func() error) (Outcome, error)

Run executes job under the cycle lock, queueing the request if a run is already in flight (queue mode — for demand-driven callers such as a poll subcommand, where the caller's request must be satisfied by a run that starts after it arrived).

• Lock free: run the job now, then execute any rerun requests queued during it (OutcomeRan, or OutcomeRanQueued if reruns happened).
• Lock busy, queue below capacity: record a rerun request and return immediately (OutcomeQueued) — the active runner executes it when the current run finishes. The requester never blocks for the job's duration.
• Lock busy, queue full: drop the request (OutcomeDiscarded) — the queued rerun(s) already guarantee a run starts after this request arrived.

The returned error carries the job's own error(s) when it ran (joined across reruns), or the infrastructure error that prevented the request from being recorded (OutcomeNone). OutcomeQueued may also accompany an error: the request was recorded but the post-enqueue re-probe failed — the demand stands, and a current or next runner consumes it. Queued and Discarded outcomes are success for the requesting process: log-and-exit-0 is the intended caller behavior.

A failed run does not stop queued demand: each queued request is owed a run that starts after it arrived, succeed or fail, so the consume loop continues through job errors (bounded by the rerun cap). WithGate stops runs from STARTING once the composition root's shutdown (or other) signal trips: a gated initial run returns OutcomeGated, and demand queued behind a closed gate waits for the next run.

func (*Exclusive) RunOrSkip

func (e *Exclusive) RunOrSkip(job func() error) (Outcome, error)

RunOrSkip executes job under the cycle lock, skipping when a run is already in flight (skip mode — for time-driven callers such as a RunLoop tick, where the next tick provides freshness and queueing would only pile on the process already doing the work):

scheduler.RunLoop(ctx, func(ctx context.Context) {
	_, _ = ex.RunOrSkip(func() error { return runCycle(ctx) })
}, opts)

A skipped tick logs a warning (the job is overrunning its interval) and returns (OutcomeSkipped, nil). When the lock is free it behaves exactly like Run's acquired path, including executing queued rerun requests at job end.

type ExclusiveOption

type ExclusiveOption func(*Exclusive)

ExclusiveOption configures NewExclusive.

func WithGate

func WithGate(gate func() bool) ExclusiveOption

WithGate installs a pre-run gate consulted every time the runner is about to START the job: on acquisition, before each queued rerun, and before each post-release handoff. A false return stops runs from starting — an initial run is skipped (OutcomeGated) and queued demand is deferred to the next run — while the in-flight run is never interrupted. Wire it to the composition root's shutdown signal (typically the shutdown context's Err) so a stop request is never followed by a fresh run. Requests are still recorded while the gate is closed: the gate decides what runs, not what queues.

func WithQueueCapacity

func WithQueueCapacity(n int) ExclusiveOption

WithQueueCapacity sets how many rerun requests may queue while a cycle is running (the default is 1, the single-slot coalescing model: one queued rerun satisfies all demand that arrived before it starts). Each queued request is consumed by exactly one rerun, so a capacity of n allows up to n back-to-back reruns to accumulate. A value below 1 is treated as 1.

type IntervalOption

type IntervalOption func(*intervalConfig)

IntervalOption configures ParseInterval.

func WithBounds

func WithBounds(low, high time.Duration) IntervalOption

WithBounds clamps a positive built-in interval to [low, high], logging a warning when it adjusts the value. A non-positive bound is ignored, so WithBounds(time.Minute, 0) enforces only a floor. If both bounds are positive and high is lower than low, the pair is normalized so a swapped argument order cannot produce an interval outside the intended band. Bounds never apply to the external or once modes.

func WithIntervalLogger

func WithIntervalLogger(l *slog.Logger) IntervalOption

WithIntervalLogger routes ParseInterval's warnings to a specific logger instead of slog.Default().

func WithName

func WithName(name string) IntervalOption

WithName sets the environment-variable name used in warning logs (for example "SYNC_INTERVAL"), so an operator can tell which setting was rejected. It defaults to "interval".

func WithRedactedValue

func WithRedactedValue(redacted bool) IntervalOption

WithRedactedValue selects whether ParseInterval's warnings echo the raw interval value: true keeps it out, making them field-name-only, while false keeps the default echo, exactly as if the option were absent. Repeated applications resolve last-wins. Pass true when the value passes through secret-capable config expansion (a YAML file with ${VAR} references): a config typo can place an expanded secret in the interval field, and the default unparseable-value warning would echo it to the startup log. Only the unparseable warning can ever carry such a value — a negative or clamped value necessarily parsed as a duration — but with true every warning omits the supplied value (the clamp warning keeps the resulting bound), so the redaction contract is uniform rather than per-branch.

func WithZeroAsOnce

func WithZeroAsOnce(zeroAsOnce bool) IntervalOption

WithZeroAsOnce selects what a zero duration ("0"/"0s") means: true makes it select ModeOnce — for a job that supports a run-once mode (a batch or one-shot context) in addition to a resident daemon — while false keeps the default ModeExternal, exactly as if the option were absent. Repeated applications resolve last-wins.

type Job

type Job func(ctx context.Context)

Job is one unit of scheduled work. It receives the loop's context, which is cancelled when RunLoop is asked to stop; a job that must run to completion past a shutdown signal should derive its own context (context.WithoutCancel) internally. A Job reports its outcome through its own closure (setting a health marker, logging); RunLoop does not inspect a return value.

type Lock

type Lock struct {
	// contains filtered or unexported fields
}

Lock is an advisory exclusive lock backed by flock(2). It is the overlap guard for a scheduled job: it serializes runs both in-process (a startup run racing a tick) and cross-process (an external trigger — a docker exec — racing the built-in loop), because flock associates the lock with the open file description, so two independent OpenFile calls contend even within one process.

func TryLock

func TryLock(path string) (l *Lock, ok bool, err error)

TryLock attempts a non-blocking exclusive lock on path, creating the file if absent. ok is false without error when another holder currently owns the lock (a run is already in flight); the caller must release an acquired lock with Unlock. On acquisition it records the current time in the file so a later contender can read the holder's age via ReadHolder.

Place path in a directory not writable by untrusted local users (e.g. a container-private /tmp or a service-owned dir, not a world-writable host /tmp shared with other accounts): the file is opened following symlinks and its holder timestamp is written with Truncate, so a pre-planted symlink at path would be clobbered. Callers that must harden further can place path under a 0700 service-owned directory.

Example

package main

import (
	"fmt"
	"os"
	"path/filepath"

	"github.com/cplieger/scheduler/v4"
)

func main() {
	path := filepath.Join(os.TempDir(), "scheduler_example.lock")
	defer func() { _ = os.Remove(path) }()

	lock, ok, _ := scheduler.TryLock(path)
	fmt.Println("first acquire:", ok)

	_, contended, _ := scheduler.TryLock(path)
	fmt.Println("second acquire while held:", contended)

	lock.Unlock()
}

Output:
first acquire: true
second acquire while held: false

func (*Lock) Unlock

func (l *Lock) Unlock()

Unlock releases the lock and closes the underlying file. The lock file is left on disk; its only content is the last holder's acquisition timestamp, reused across runs and irrelevant while the lock is free.

type LoopOptions

type LoopOptions struct {
	// Interval is the gap between ticks. It must be positive; RunLoop returns
	// immediately otherwise (the built-in mode ParseInterval selects always
	// carries a positive interval).
	Interval time.Duration
	// Jitter spreads each tick uniformly across ±(Jitter × Interval) so
	// restarts across many instances do not synchronize into a thundering herd on a
	// shared upstream. It is a fraction in [0, 1); 0 disables jitter. The
	// startup fire is never jittered.
	Jitter float64
	// FireOnStart runs the job immediately as the first iteration, before the
	// first interval elapses, so a freshly-deployed container does work at once
	// instead of waiting a full interval.
	FireOnStart bool
}

LoopOptions configures RunLoop. Interval must be positive; Jitter and FireOnStart are optional.

type Mode

type Mode int

Mode is how a container job is scheduled, derived from an interval environment variable by ParseInterval.

const (
	// ModeBuiltin runs the job once at startup, then on every interval tick. It
	// is selected by a positive interval duration and is the fallback when the
	// value is empty, unparseable, or (by default) negative.
	ModeBuiltin Mode = iota
	// ModeExternal idles: the built-in loop is disabled and runs are triggered
	// out-of-band (for example an Ofelia docker-exec of a one-shot subcommand).
	// It is selected by the "off" and "disabled" sentinels, and by a zero
	// duration unless WithZeroAsOnce(true) is passed.
	ModeExternal
	// ModeOnce runs the job exactly once, then exits. It is selected by a zero
	// duration ("0"/"0s") only when WithZeroAsOnce(true) is passed; otherwise a
	// zero duration selects ModeExternal.
	ModeOnce
)

func (Mode) String

func (m Mode) String() string

String returns the lowercase mode name for logging.

type Outcome

type Outcome int

Outcome reports what an Exclusive.Run or RunOrSkip call did with the request. It stays meaningful alongside a non-nil error: check the error first, then the outcome — OutcomeRan with an error means the job ran and failed, OutcomeNone with an error means an infrastructure failure prevented the request from running or queueing at all, and OutcomeQueued with an error means the request WAS recorded but the post-enqueue re-probe failed (the queued demand still stands; a current or next runner consumes it).

const (
	// OutcomeNone is the zero value: the request produced neither a run nor a
	// queue entry. It is returned only with an infrastructure error (the lock
	// or queue file could not be used).
	OutcomeNone Outcome = iota
	// OutcomeRan means this call acquired the cycle lock and ran the job.
	OutcomeRan
	// OutcomeRanQueued means this call ran the job and then executed at least
	// one queued rerun request on top of it.
	OutcomeRanQueued
	// OutcomeQueued means a cycle was already in flight, so this request was
	// queued; the active runner executes it when the current run finishes (or,
	// if that runner retires at its rerun cap, the next run to acquire the
	// lock satisfies it).
	OutcomeQueued
	// OutcomeDiscarded means a cycle was in flight and the rerun queue was
	// already full; the request was dropped because the queued rerun(s)
	// already guarantee a run starts after this request arrived.
	OutcomeDiscarded
	// OutcomeSkipped means a cycle was in flight and the caller chose skip
	// mode (RunOrSkip): the tick was dropped without queueing.
	OutcomeSkipped
	// OutcomeGated means the run gate (WithGate) was closed when the runner
	// was about to start the job: nothing ran, and the queue counter was left
	// untouched (a request this call had already queued on the re-probe path
	// stays queued; the next run satisfies it).
	OutcomeGated
)

func (Outcome) String

func (o Outcome) String() string

String returns the lowercase outcome name for logging.

type Schedule

type Schedule struct {
	Interval time.Duration
	Mode     Mode
}

Schedule is the parsed result of an interval environment variable: the built-in cadence and the selected Mode. Interval is meaningful only in ModeBuiltin; the other modes carry the default for reference.

func ParseInterval

func ParseInterval(raw string, def time.Duration, opts ...IntervalOption) Schedule

ParseInterval interprets a raw interval environment value into a Schedule, applying the standard sentinel and fallback rules shared by every scheduled container job:

• empty -> def, ModeBuiltin
• "off" / "disabled" -> def, ModeExternal (case-insensitive)
• a positive Go duration -> that duration (clamped by WithBounds), ModeBuiltin
• a zero duration ("0"/"0s") -> def, ModeExternal (or ModeOnce with WithZeroAsOnce(true))
• a negative duration -> def, ModeBuiltin, with a warning (a likely typo; falling back to the default cadence beats silently disabling the job)
• anything unparseable -> def, ModeBuiltin, with a warning

def is the fallback cadence used for every non-positive outcome; it is also carried on the returned Schedule in the external and once modes for reference. def must be positive and ParseInterval panics otherwise (a programmer error in the composition root, caught at first boot — the same contract as time.NewTicker): def becomes the Interval of every ModeBuiltin result (empty, negative, or unparseable input), and the library's invariant that a ModeBuiltin Schedule always carries a positive Interval -- which a consumer relies on when it passes the Interval straight to time.NewTicker -- holds only when def > 0. (RunLoop itself also guards defensively, since a hand-built LoopOptions can carry any Interval.) Warnings are logged via slog.Default() unless WithIntervalLogger is set.

Example

package main

import (
	"fmt"
	"time"

	"github.com/cplieger/scheduler/v4"
)

func main() {
	built := scheduler.ParseInterval("30m", time.Hour)
	fmt.Printf("%s every %s\n", built.Mode, built.Interval)

	fmt.Println(scheduler.ParseInterval("off", time.Hour).Mode)
	fmt.Println(scheduler.ParseInterval("0", time.Hour, scheduler.WithZeroAsOnce(true)).Mode)
}

Output:
built-in every 30m0s
external
once

type SlotFile

type SlotFile struct {
	// contains filtered or unexported fields
}

SlotFile is a single-slot byte payload shared across processes through one file, mutated by atomic read-modify-write transactions under a short exclusive flock(2) on the file itself. It is the storage mechanism behind Exclusive's rerun counter, exported so an app can build its own coalescing state on the same transaction — for example a payload-carrying demand slot whose merge and claim semantics are app policy (docker-renovate-scheduler records WHICH repos a queued trigger wants, not just that one arrived).

The transaction is content-agnostic: what the bytes mean, how concurrent demands merge, and when a slot counts as satisfied are the caller's parser and policy. Parsers must self-heal on unparseable content (treat torn or garbage bytes as the zero value): a crash between Truncate and WriteAt can leave a torn slot, and the library's own counter and every other slot user recover by reading garbage as empty.

The slot file is created on first use and never unlinked — by the library or the caller — while contenders may exist: unlinking a locked file lets a concurrent opener land on a different inode and breaks mutual exclusion. "Clear" is writing an empty payload. Place the path in a directory not writable by untrusted local users, per the same symlink-following caveat as TryLock.

func NewSlotFile

func NewSlotFile(path string) *SlotFile

NewSlotFile returns a SlotFile backed by the file at path (see the type documentation for the trust and lifecycle rules).

func (*SlotFile) Mutate

func (s *SlotFile) Mutate(fn func(before []byte) []byte) (before []byte, err error)

Mutate applies fn to the slot's current content under an exclusive flock on the slot file and returns the content fn saw. fn receives the current bytes (empty on first use) and returns the bytes to store; returning content byte-equal to before (returning before itself is the idiom) leaves the file untouched, so a read is a Mutate whose fn returns its argument. A nil return stores an empty payload (the clear idiom).

The lock is blocking: contenders serialize, and the critical section is the read, fn, and write of one short payload — microseconds, never a job's duration — so keep fn small and non-blocking (it runs under the flock). before stays meaningful alongside a non-nil error when the failure happened after the read (a truncate or write error).