schedgate

package
v1.5.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Overview

Package schedgate is the shared "should this scheduled bot fire now?" gate used by all three scheduled-launch paths: pkg/cli/schedule (host crontab), pkg/trigger.Scheduler (in-process spine), and pkg/cloudsched (multi-replica cloud ticker). It provides three composable pieces — an overlap decision over live runs, a guard-command executor, and a tick-record shape — with no I/O of its own beyond running the guard subprocess. Each surface keeps its own persistence (JSONL locally, pkg/audit in cloud mode).

Index

Constants

View Source 
const (
	OverlapSkip      = "skip"
	OverlapAllow     = "allow"
	OverlapKeepalive = "keepalive"
)

Overlap policy values. An empty Overlap normalizes to OverlapSkip: firing while a previous run of the same schedule is still live is a latent-bug default (a nightly that overruns its window piles up concurrent runs on the same repo), so skip is the safe baseline and allow is the explicit opt-in.

OverlapKeepalive is the "always-on agent" policy: at-most-one-live (like skip), but a live run that has gone silent past StaleAfter is treated as dead — dropped from the live set so the tick relaunches a fresh run, and surfaced for reaping. It is what keeps a bot continuously alive across short, individually-budgeted runs rather than one immortal run.

View Source 
const (
	DefaultGuardTimeout = 30 * time.Second
	DefaultGuardVar     = "guard_output"
	// DefaultStaleAfter is the keepalive silence cutoff: a running run
	// whose last progress is older than this is treated as dead. Chosen
	// generously so a long-but-alive run is not falsely reaped; override
	// per schedule (ideally >= the bot's max_duration).
	DefaultStaleAfter = 5 * time.Minute
)

Guard defaults.

View Source 
const TailCap = 16 * 1024

TailCap bounds the stdout/stderr tails stored on audit records. The tail (not the head) is preserved: shell errors print last.

View Source 
const TickSchemaVersion = 1

TickSchemaVersion is the current TickRecord schema.

Variables

This section is empty.

Functions

func DefaultLocalAuditPath 

func DefaultLocalAuditPath() (string, error)

DefaultLocalAuditPath resolves the local tick-audit file the way the schedule CLI resolves its manifest (ITERION_SCHEDULES_FILE override, else ~/.iterion/schedules.yaml), so the in-process trigger scheduler writes to the SAME file `iterion schedule audit` reads.

func LiveAndStaleRunsForSchedule added in v1.0.0 

func LiveAndStaleRunsForSchedule(ctx context.Context, s ScheduleRunLister, scheduleID string, staleAfter time.Duration, now time.Time, logger *iterlog.Logger) (live, stale []string)

LiveAndStaleRunsForSchedule is the keepalive-aware liveness query. It returns two disjoint, order-preserving lists of non-terminal runs stamped with this schedule's provenance:

  • live: runs that count against the overlap policy;
  • stale: running runs whose last progress (UpdatedAt) is older than staleAfter — treated as dead so a keepalive tick relaunches, and returned so the caller can reap them.

When staleAfter <= 0 (the non-keepalive path) no run is ever considered stale, so this reduces exactly to the old LiveRunsForSchedule behavior. Staleness is gated on RunStatusRunning only: a paused run is legitimately idle and must never be reaped.

func LiveRunsForSchedule 

func LiveRunsForSchedule(ctx context.Context, s ScheduleRunLister, scheduleID string, logger *iterlog.Logger) []string

LiveRunsForSchedule returns the IDs of runs stamped with this schedule's provenance whose status is non-terminal, preserving the store's created_at-ascending order (so index 0 is the oldest — the deterministic "blocking run" for audit).

A run that fails to load is logged and skipped rather than blocking the tick: this is a gate, not a data-integrity boundary.

func LocalAuditPathFor 

func LocalAuditPathFor(manifestPath string) string

LocalAuditPathFor places the tick-audit JSONL next to the schedule manifest's cron logs: <manifest-dir>/logs/tick-audit.jsonl. Shared by the host-cron surface (which knows its manifest path exactly) and DefaultLocalAuditPath below.

func TruncTail 

func TruncTail(s string, max int) string

TruncTail returns s unchanged when it fits max bytes; otherwise the last max bytes prefixed with a truncation marker. The tail is kept because errors print last.

func Validate 

func Validate(p Policy) error

Validate reports whether p is coherent. Errors are user-facing and name the offending field. The guard command itself is not validated (any sh -lc snippet is legal).

Types

type Decision 

type Decision int

Decision is the outcome of EvaluateOverlap. 

const (
	// DecisionFire means the tick may proceed (to the guard, then launch).
	DecisionFire Decision = iota
	// DecisionSkipOverlap means firing would exceed the overlap policy.
	DecisionSkipOverlap
)

func EvaluateOverlap 

func EvaluateOverlap(liveIDs []string, p Policy) (Decision, string)

EvaluateOverlap decides whether a tick may fire given the schedule's currently-live run IDs. Pure. On skip, the returned blocking ID is the first live run (the store lists them created_at-ascending, so "first" is the oldest — deterministic across replicas) for the audit reason.

type GateInput 

type GateInput struct {
	Policy Policy
	// Lister powers the overlap check; nil skips it (a surface with no
	// store access degrades to guard-only, never to a hard error).
	Lister ScheduleRunLister
	// ScheduleID is the provenance key runs were stamped with
	// (RunSource.ScheduleID) — what LiveRunsForSchedule queries.
	ScheduleID string
	// Record is the surface-prefilled tick record (surface, ids,
	// tenant, cron, timestamp). Apply stamps the decision, reason and
	// guard fields on the copy it returns.
	Record TickRecord
	// GuardDir/GuardEnv shape the guard subprocess (see GuardSpec).
	GuardDir string
	GuardEnv []string
	Logger   *iterlog.Logger
	// Now overrides the clock for keepalive staleness detection (zero =
	// time.Now()). Tests inject a fixed instant; production leaves it zero.
	Now time.Time
}

GateInput bundles what a launch surface knows at tick time. The three scheduled-launch surfaces (host-cron, trigger spine, cloud ticker) all evaluate the SAME overlap→guard sequence; only the audit sink, the human reporting, and the guard's cwd/env differ — so those stay with the caller and everything else lives in Apply.

type GateOutcome 

type GateOutcome struct {
	// Proceed: launch the run. False: the slot passes, and Record
	// carries the audited reason (skipped_overlap / guard_blocked /
	// guard_error).
	Proceed bool
	// GuardRan is true when a guard command executed and passed —
	// callers then inject GuardStdout as vars[Policy.GuardVar], even
	// when the stdout is empty (the var's presence is the contract).
	GuardRan    bool
	GuardStdout string
	// Record is the input record stamped with the decision. On
	// Proceed it is left undecided — "fired" is only true after the
	// launch attempt, which the caller owns.
	Record TickRecord
	// ReapRunIDs lists keepalive runs found stale (silent past
	// StaleAfter) at this tick. The caller cancels them via its store so
	// the zombies free resources; schedgate stays I/O-free. Non-empty
	// only on the keepalive path, and only alongside Proceed (a stale
	// run no longer blocks, so the tick fires a fresh one).
	ReapRunIDs []string
}

GateOutcome is the gate's verdict for one consumed tick slot.

func Apply 

func Apply(ctx context.Context, in GateInput) GateOutcome

Apply runs the shared overlap→guard gate sequence. It never returns an error: a broken guard is a first-class audited outcome (guard_error), not a launch-path failure.

type GuardKind 

type GuardKind int

GuardKind classifies a guard execution outcome. 

const (
	// GuardOK: exit 0 — the run fires and Stdout becomes vars[guard_var].
	GuardOK GuardKind = iota
	// GuardBlocked: exit non-zero — the guard deliberately said "nothing
	// to do"; the tick is skipped.
	GuardBlocked
	// GuardError: the guard itself broke — spawn failure or timeout.
	// Distinct from GuardBlocked so an operator can tell "guard said no"
	// from "guard is broken".
	GuardError
)

func (GuardKind) String 

func (k GuardKind) String() string

type GuardResult 

type GuardResult struct {
	Kind GuardKind
	// ExitCode is the guard's exit status for GuardBlocked; -1 for
	// GuardError (no meaningful status).
	ExitCode int
	// Stdout is the full stdout on GuardOK (it may be structured input,
	// e.g. a JSON issue list — never truncated); the TruncTail'd tail
	// otherwise.
	Stdout string
	// StderrTail is always the TruncTail'd stderr.
	StderrTail string
	// Err is nil on GuardOK.
	Err error
	// Duration is the wall-clock guard runtime.
	Duration time.Duration
}

GuardResult is what the audit and the launch layer consume.

func RunGuard 

func RunGuard(parentCtx context.Context, spec GuardSpec) GuardResult

RunGuard executes spec.Command with `sh -lc` under its OWN timeout-bounded context derived from parentCtx. It never installs a deadline on parentCtx itself: the run has not started yet, and a guard timeout must not cascade into the caller's tick loop.

cmd.WaitDelay bounds the orphan-pipe wait when the timeout kills `sh` but a grandchild keeps the inherited pipes open (same discipline as dispatcher hooks).

type GuardSpec 

type GuardSpec struct {
	// Command is the sh -lc snippet.
	Command string
	// Dir is the working directory (the schedule's workdir, so `gh` /
	// `git` run in repo context).
	Dir string
	// Env entries are appended to the inherited os.Environ.
	Env []string
	// Timeout bounds the subprocess; <= 0 means DefaultGuardTimeout.
	Timeout time.Duration
}

GuardSpec is the resolved input to RunGuard.

type Policy 

type Policy struct {
	// Overlap is "skip" (default: don't fire while a previous run of
	// this schedule is live) or "allow".
	Overlap string `yaml:"overlap,omitempty" json:"overlap,omitempty" bson:"overlap,omitempty"`
	// MaxConcurrent caps live runs when Overlap is "allow", inclusive
	// of the run about to fire (2 = fire while fewer than 2 are live).
	// 0 with "allow" means unlimited. Invalid with "skip".
	MaxConcurrent int `yaml:"max_concurrent,omitempty" json:"max_concurrent,omitempty" bson:"max_concurrent,omitempty"`
	// Guard is an optional `sh -lc` snippet run before any launch:
	// exit 0 fires the run and the guard's stdout becomes the run's
	// vars[GuardVar]; non-zero skips the tick.
	Guard string `yaml:"guard,omitempty" json:"guard,omitempty" bson:"guard,omitempty"`
	// GuardTimeout bounds the guard subprocess (Go duration string,
	// default 30s).
	GuardTimeout string `yaml:"guard_timeout,omitempty" json:"guard_timeout,omitempty" bson:"guard_timeout,omitempty"`
	// GuardVar names the workflow var receiving the guard's stdout
	// (default "guard_output").
	GuardVar string `yaml:"guard_var,omitempty" json:"guard_var,omitempty" bson:"guard_var,omitempty"`
	// StaleAfter is the keepalive silence cutoff (Go duration string): a
	// running run whose last progress is older than this counts as dead,
	// so a fresh run relaunches and the zombie is reaped. Only meaningful
	// with Overlap=keepalive; empty normalizes to DefaultStaleAfter.
	StaleAfter string `yaml:"stale_after,omitempty" json:"stale_after,omitempty" bson:"stale_after,omitempty"`
}

Policy is the concurrency + guard contract shared by all three surfaces. Zero value is legal (skip, no guard); Normalize promotes it to explicit defaults. All fields are additive on their host schemas — old manifests / rows load unchanged.

func Normalize 

func Normalize(p Policy) Policy

Normalize returns p with defaults applied. Idempotent; never returns a Policy with empty Overlap, GuardTimeout or GuardVar.

func (Policy) GuardTimeoutDuration 

func (p Policy) GuardTimeoutDuration() time.Duration

GuardTimeoutDuration resolves the policy's guard timeout, falling back to the default on empty or unparseable values (Validate rejects the latter upstream; the fallback keeps runtime behavior total).

func (Policy) StaleAfterDuration added in v1.0.0 

func (p Policy) StaleAfterDuration() time.Duration

StaleAfterDuration resolves the keepalive silence cutoff, falling back to DefaultStaleAfter on empty or unparseable values (Validate rejects the latter upstream; the fallback keeps runtime behavior total).

type ScheduleRunLister 

type ScheduleRunLister interface {
	ListRunsBySchedule(ctx context.Context, scheduleID string) ([]string, error)
	LoadRun(ctx context.Context, id string) (*store.Run, error)
}

ScheduleRunLister is the narrow slice of store.RunStore the overlap gate needs. Both the filesystem and Mongo stores satisfy it (via ListRunsBySchedule); kept as a local interface so tests inject fakes and this package stays store-agnostic.

type Surface 

type Surface string

Surface names the scheduled-launch path that made a tick decision, so mixed audit streams can be filtered per-source. 

const (
	SurfaceHostCron Surface = "host-cron"
	SurfaceTrigger  Surface = "trigger"
	SurfaceCloud    Surface = "cloud"
)

type TickDecision 

type TickDecision string

TickDecision is the audit-level outcome of one tick. Exactly one tag lands on each record. 

const (
	// TickFired: overlap + guard passed and a launch was attempted.
	TickFired TickDecision = "fired"
	// TickSkippedOverlap: a live run of the same schedule held the slot.
	TickSkippedOverlap TickDecision = "skipped_overlap"
	// TickGuardBlocked: the guard exited non-zero ("nothing to do").
	TickGuardBlocked TickDecision = "guard_blocked"
	// TickGuardError: the guard failed to execute (spawn error/timeout).
	TickGuardError TickDecision = "guard_error"
)

type TickRecord 

type TickRecord struct {
	// Schema is the record format version (currently 1). Readers must
	// ignore unknown fields and reject unknown higher schemas loudly.
	Schema int `json:"schema"`
	// Surface is the launch path that made the decision.
	Surface Surface `json:"surface"`
	// ScheduleID is the stable schedule identity: ScheduleEntry.Name
	// (host-cron), Subscription.ID (trigger), ScheduledBot.ID (cloud).
	ScheduleID string `json:"schedule_id"`
	// ScheduleName is the human label when it differs from the ID.
	ScheduleName string       `json:"schedule_name,omitempty"`
	BotID        string       `json:"bot_id,omitempty"`
	TenantID     string       `json:"tenant_id,omitempty"`
	Cron         string       `json:"cron,omitempty"`
	At           time.Time    `json:"at"`
	Decision     TickDecision `json:"decision"`
	// Reason is a human sentence, e.g. "blocked by live run r_x".
	Reason string `json:"reason,omitempty"`
	// RunID is the launched run on TickFired.
	RunID string `json:"run_id,omitempty"`
	// BlockingRunID names the (oldest) live run on TickSkippedOverlap so
	// the operator can inspect or cancel it.
	BlockingRunID string `json:"blocking_run_id,omitempty"`
	// GuardExit is the guard's exit code — pointer so 0 is meaningful.
	GuardExit     *int          `json:"guard_exit,omitempty"`
	GuardDuration time.Duration `json:"guard_duration_ns,omitempty"`
	StdoutTail    string        `json:"stdout_tail,omitempty"`
	StderrTail    string        `json:"stderr_tail,omitempty"`
	Error         string        `json:"error,omitempty"`
}

TickRecord is the shared audit row written by every surface — as a JSONL line locally (~/.iterion/logs/tick-audit.jsonl), as audit.Event Meta in cloud mode. It answers "why didn't my scheduled bot fire last night?" deterministically. Field names are stable: operators grep the JSONL.

func NewTickRecord 

func NewTickRecord(surface Surface, scheduleID string, at time.Time, decision TickDecision) TickRecord

NewTickRecord builds a record with the invariant fields stamped.

func (*TickRecord) ApplyGuard 

func (t *TickRecord) ApplyGuard(res GuardResult)

ApplyGuard folds a guard result into the record (exit code, tails, duration, error text). The decision itself stays the caller's choice so a GuardOK record can still be TickFired.

func (TickRecord) ToAuditMeta 

func (t TickRecord) ToAuditMeta() map[string]any

ToAuditMeta projects the record onto the map shape pkg/audit stores in Event.Meta. Kept here — not in each surface — so the cloud rows and the local JSONL stay field-compatible.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.