cloudsched

package
v1.9.4 Latest Latest
Warning

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

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

Documentation

Overview

Package cloudsched is the cloud-mode recurring-bot scheduler: a per-org store of cron-scheduled bots and a multi-replica-safe ticker that fires each due schedule exactly once (CAS on the next-fire time, no leader election needed). The self-hosted equivalent is `iterion schedule` (host crontab); this is its cloud counterpart.

Index

Constants

View Source 
const Collection = "scheduled_bots"

Collection is the Mongo collection name for scheduled bots.

Variables

View Source 
var ErrNotFound = fmt.Errorf("cloudsched: scheduled bot not found")

ErrNotFound is returned by Get/Delete for an unknown id.

Functions

func EnsureSchema 

func EnsureSchema(ctx context.Context, db *mongo.Database) error

EnsureSchema creates the indexes. Idempotent.

func NextFire 

func NextFire(expr string, after time.Time) (time.Time, error)

NextFire returns the next instant after `after` at which expr fires.

func NextFireForBot added in v1.0.0 

func NextFireForBot(sb ScheduledBot, after time.Time) (time.Time, error)

NextFireForBot computes a schedule's next-fire instant: a fixed interval for keepalive (IntervalSeconds > 0), else the cron expression. The single seam every ticker/store path uses so keepalive and cron stay consistent.

func ValidateCron 

func ValidateCron(expr string) error

ValidateCron reports whether expr is a valid 5-field standard cron.

Types

type GateFunc added in v0.50.0 

type GateFunc func(ctx context.Context, sb ScheduledBot) (proceed bool, guardStdout string, rec schedgate.TickRecord)

GateFunc decides, AFTER this replica won the slot's CAS, whether the launch proceeds (overlap policy + guard, pkg/schedgate). On proceed=false, rec is the skip record for the audit sink; on proceed=true rec is ignored (the ticker audits the fired outcome itself, with the launch error) and guardStdout carries the guard's stdout ("" when no guard) for injection into the launch vars. Positioned after the CAS on purpose: the slot must be consumed exactly once across replicas regardless of the gate's verdict, and only the CAS winner may run the guard (a pre-CAS gate would run it on every replica).

type LaunchFunc 

type LaunchFunc func(ctx context.Context, sb ScheduledBot) error

LaunchFunc fires one scheduled bot run. The cloud bootstrap wires it to the run publisher (resolve the bot, build a LaunchSpec, SubmitLaunch).

type MemoryStore 

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

MemoryStore is an in-memory Store for tests + single-process use. ClaimTick enforces the same CAS-on-next_fire_at semantics as the Mongo store.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore returns an empty in-memory store.

func (*MemoryStore) ClaimTick 

func (s *MemoryStore) ClaimTick(_ context.Context, id string, expectedNext, newNext, firedAt time.Time) (bool, error)

func (*MemoryStore) Create 

func (s *MemoryStore) Create(_ context.Context, sb ScheduledBot) error

func (*MemoryStore) Delete 

func (s *MemoryStore) Delete(_ context.Context, id string) error

func (*MemoryStore) DeleteByIntegration 

func (s *MemoryStore) DeleteByIntegration(_ context.Context, tenantID, integrationID string) error

func (*MemoryStore) Get 

func (s *MemoryStore) Get(_ context.Context, id string) (ScheduledBot, error)

func (*MemoryStore) ListByIntegration 

func (s *MemoryStore) ListByIntegration(_ context.Context, tenantID, integrationID string) ([]ScheduledBot, error)

func (*MemoryStore) ListByTenant added in v0.50.0 

func (s *MemoryStore) ListByTenant(_ context.Context, tenantID string) ([]ScheduledBot, error)

func (*MemoryStore) ListDue 

func (s *MemoryStore) ListDue(_ context.Context, now time.Time, limit int) ([]ScheduledBot, error)

func (*MemoryStore) Update added in v0.50.0 

func (s *MemoryStore) Update(_ context.Context, id string, patch SchedulePatch) (ScheduledBot, error)

type MongoStore 

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

MongoStore is the Mongo-backed Store.

func NewMongoStore 

func NewMongoStore(db *mongo.Database) *MongoStore

NewMongoStore builds a Mongo-backed scheduled-bot store.

func (*MongoStore) ClaimTick 

func (s *MongoStore) ClaimTick(ctx context.Context, id string, expectedNext, newNext, firedAt time.Time) (bool, error)

ClaimTick is the CAS: the update matches only while next_fire_at still equals expectedNext, so the first replica to advance it wins and the rest get (false, nil). exactly-once per slot, no leader.

func (*MongoStore) Create 

func (s *MongoStore) Create(ctx context.Context, sb ScheduledBot) error

func (*MongoStore) Delete 

func (s *MongoStore) Delete(ctx context.Context, id string) error

func (*MongoStore) DeleteByIntegration 

func (s *MongoStore) DeleteByIntegration(ctx context.Context, tenantID, integrationID string) error

func (*MongoStore) Get 

func (s *MongoStore) Get(ctx context.Context, id string) (ScheduledBot, error)

func (*MongoStore) ListByIntegration 

func (s *MongoStore) ListByIntegration(ctx context.Context, tenantID, integrationID string) ([]ScheduledBot, error)

func (*MongoStore) ListByTenant added in v0.50.0 

func (s *MongoStore) ListByTenant(ctx context.Context, tenantID string) ([]ScheduledBot, error)

func (*MongoStore) ListDue 

func (s *MongoStore) ListDue(ctx context.Context, now time.Time, limit int) ([]ScheduledBot, error)

func (*MongoStore) Update added in v0.50.0 

func (s *MongoStore) Update(ctx context.Context, id string, patch SchedulePatch) (ScheduledBot, error)

Update applies a partial mutation. Reads the current row, mutates it via applySchedulePatch, and writes back via ReplaceOne — the atomicity that matters here is exactly-once fire (ClaimTick's CAS), not multi-writer serialisation on the mutable payload (Cron/Vars/…), so a full replace is safe and matches the semantics operators expect from a REST PATCH.

type SchedulePatch added in v0.50.0 

type SchedulePatch struct {
	Cron            *string
	IntervalSeconds *int
	NextFireAt      *time.Time
	Vars            *map[string]string
	RepoURL         *string
	RepoRef         *string
	Disabled        *bool
	Overlap         *string
	MaxConcurrent   *int
	Guard           *string
	GuardTimeout    *string
	GuardVar        *string
	StaleAfter      *string
	UpdatedAt       time.Time
}

SchedulePatch describes the mutable slice of ScheduledBot the manual CRUD endpoint exposes. Nil field = leave untouched. Cron carries the new expression when set (already validated by ValidateCron); the store recomputes NextFireAt through NextFire(cron, now).

type ScheduledBot 

type ScheduledBot struct {
	ID                string `bson:"_id" json:"id"`
	TenantID          string `bson:"tenant_id" json:"tenant_id"`
	RepoIntegrationID string `bson:"repo_integration_id,omitempty" json:"repo_integration_id,omitempty"`
	BotID             string `bson:"bot_id" json:"bot_id"`
	Cron              string `bson:"cron" json:"cron"` // 5-field standard cron
	// IntervalSeconds drives an always-on (keepalive) schedule instead of Cron:
	// the ticker relaunches the bot every IntervalSeconds (sub-minute allowed,
	// bounded by the ticker's own Interval). Exactly one of Cron/IntervalSeconds
	// is set. Overlap=keepalive gives at-most-one-live + staleness reaping.
	IntervalSeconds int               `bson:"interval_seconds,omitempty" json:"interval_seconds,omitempty"`
	Vars            map[string]string `bson:"vars,omitempty" json:"vars,omitempty"`
	RepoURL         string            `bson:"repo_url,omitempty" json:"repo_url,omitempty"`
	RepoRef         string            `bson:"repo_ref,omitempty" json:"repo_ref,omitempty"`
	Disabled        bool              `bson:"disabled,omitempty" json:"disabled,omitempty"`

	// Overlap policy + pre-launch guard (pkg/schedgate). Overlap ""
	// normalizes to "skip": a slot whose previous run is still live is
	// consumed without launching (audited), instead of piling up runs.
	Overlap       string `bson:"overlap,omitempty" json:"overlap,omitempty"`
	MaxConcurrent int    `bson:"max_concurrent,omitempty" json:"max_concurrent,omitempty"`
	Guard         string `bson:"guard,omitempty" json:"guard,omitempty"`
	GuardTimeout  string `bson:"guard_timeout,omitempty" json:"guard_timeout,omitempty"`
	GuardVar      string `bson:"guard_var,omitempty" json:"guard_var,omitempty"`
	// StaleAfter is the keepalive silence cutoff (Go duration); empty defaults
	// to schedgate.DefaultStaleAfter. Only meaningful with Overlap=keepalive.
	StaleAfter string `bson:"stale_after,omitempty" json:"stale_after,omitempty"`

	// NextFireAt is the next UTC instant this schedule is due. The ticker
	// CAS-advances it the moment it claims a tick, so a second replica racing
	// on the same row finds next_fire_at already moved and backs off — exactly
	// one fire per slot without a leader.
	NextFireAt time.Time  `bson:"next_fire_at" json:"next_fire_at"`
	LastFireAt *time.Time `bson:"last_fire_at,omitempty" json:"last_fire_at,omitempty"`

	CreatedBy string    `bson:"created_by,omitempty" json:"created_by,omitempty"`
	CreatedAt time.Time `bson:"created_at" json:"created_at"`
	UpdatedAt time.Time `bson:"updated_at" json:"updated_at"`
}

ScheduledBot is one cron-scheduled bot run. When RepoURL is set the cloud runner clones that repo before the bot starts (RepoRef pins the ref); stateful bots that persist to git (e.g. feed-watch state_commit=true) need this to have a workspace to push to. When RepoURL is empty the run executes against the runner pod's base WorkDir, matching the pre-repo behaviour.

func (ScheduledBot) Policy added in v0.50.0 

func (sb ScheduledBot) Policy() schedgate.Policy

Policy projects the schedule's schedgate fields into a normalized overlap/guard policy.

type Store 

type Store interface {
	Create(ctx context.Context, sb ScheduledBot) error
	Get(ctx context.Context, id string) (ScheduledBot, error)
	ListByIntegration(ctx context.Context, tenantID, integrationID string) ([]ScheduledBot, error)
	// ListByTenant returns every schedule owned by tenantID (including rows
	// with an empty RepoIntegrationID — the manual CRUD path). Ordered by
	// CreatedAt ascending for stable rendering.
	ListByTenant(ctx context.Context, tenantID string) ([]ScheduledBot, error)
	// ListDue returns enabled schedules whose next_fire_at <= now (capped by
	// limit, 0 = no cap), oldest-due first.
	ListDue(ctx context.Context, now time.Time, limit int) ([]ScheduledBot, error)
	// ClaimTick atomically advances a schedule's next_fire_at from expectedNext
	// to newNext (and stamps last_fire_at = firedAt), returning true only when
	// THIS caller won the CAS. A losing replica gets (false, nil). This is the
	// exactly-once primitive — no leader election.
	ClaimTick(ctx context.Context, id string, expectedNext, newNext, firedAt time.Time) (bool, error)
	// Update applies a partial mutation to an existing schedule. Only the
	// non-nil fields of patch are written; NextFireAt is recomputed from the
	// new Cron when Cron is set.
	Update(ctx context.Context, id string, patch SchedulePatch) (ScheduledBot, error)
	Delete(ctx context.Context, id string) error
	DeleteByIntegration(ctx context.Context, tenantID, integrationID string) error
}

Store persists scheduled bots. Mongo (cloud) and an in-memory impl (tests) satisfy it.

type Ticker 

type Ticker struct {
	Store    Store
	Launch   LaunchFunc
	Interval time.Duration // default 1 minute
	Logger   *iterlog.Logger
	// Gate, when set, runs the overlap/guard policy between the CAS win
	// and the launch. Audit, when set, receives every gate decision AND
	// the fired outcome. Both nil-safe (nil = fire unconditionally,
	// unaudited — the pre-gate behavior).
	Gate  GateFunc
	Audit func(rec schedgate.TickRecord)
	// Now is injectable for tests; defaults to time.Now().UTC().
	Now func() time.Time
}

Ticker fires due schedules. It is multi-replica-safe WITHOUT leader election: every replica may run a Ticker; the CAS in ClaimTick guarantees each slot fires exactly once (the first replica to advance next_fire_at wins; the rest see the moved value and skip).

func (*Ticker) Run 

func (t *Ticker) Run(ctx context.Context)

Run loops Tick every Interval until ctx is cancelled. Start one per replica.

func (*Ticker) Tick 

func (t *Ticker) Tick(ctx context.Context) (int, error)

Tick fires every due schedule this caller wins the CAS for, returning the count it fired. Exposed for tests + a manual kick.

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.