store

package
v0.40.0 Latest Latest
Warning

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

 Go to latest Published: Jul 21, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrJobHandlerNameEmpty indicates a registered handler without a job name.
	ErrJobHandlerNameEmpty = errors.New("cron store: job handler name is empty")
	// ErrJobHandlerNameTooLong indicates a job name beyond the persisted width.
	ErrJobHandlerNameTooLong = errors.New("cron store: job handler name exceeds 128 characters")
	// ErrJobHandlerDuplicate indicates two handlers registered for one job name.
	ErrJobHandlerDuplicate = errors.New("cron store: duplicate job handler")
	// ErrScheduleNameRequired indicates a schedule spec without a name.
	ErrScheduleNameRequired = errors.New("cron store: schedule name is required")
	// ErrScheduleNameTooLong indicates a schedule name beyond the column width.
	ErrScheduleNameTooLong = errors.New("cron store: schedule name exceeds 128 characters")
	// ErrScheduleTimeoutNegative indicates a negative per-run timeout.
	ErrScheduleTimeoutNegative = errors.New("cron store: schedule timeout must not be negative")
	// ErrScheduleTimeoutTooLong indicates milliseconds outside time.Duration.
	ErrScheduleTimeoutTooLong = errors.New("cron store: schedule timeout exceeds the supported duration")
	// ErrScheduleTimeoutPrecision indicates a timeout that cannot be persisted
	// exactly in whole milliseconds.
	ErrScheduleTimeoutPrecision = errors.New("cron store: schedule timeout must use whole milliseconds")
	// ErrScheduleWindowInverted indicates EndsAt at or before StartsAt.
	ErrScheduleWindowInverted = errors.New("cron store: schedule window must end after it starts")
	// ErrScheduleNeverFires indicates an enabled schedule whose trigger
	// yields no occurrence from now (a past one-shot, an expired window).
	ErrScheduleNeverFires = errors.New("cron store: schedule trigger yields no future occurrence")
	// ErrAbandonTakeoverIncomplete indicates a takeover that could not
	// finalize every locked stale run — a broken invariant, since the rows
	// were selected and locked in the same transaction.
	ErrAbandonTakeoverIncomplete = errors.New("cron store: abandoned-run takeover left locked runs unmarked")
	// ErrJobPanicked wraps a recovered handler panic into the run's failure.
	ErrJobPanicked = errors.New("cron store: job panicked")
	// ErrRunTimedOut marks a run that outlived its timeout, whatever its
	// handler returned.
	ErrRunTimedOut = errors.New("cron store: run timed out")
)
View Source 
var Module = fx.Module(
	"vef:cron:store",

	fx.Provide(
		fx.Annotate(NewRegistry, fx.ParamTags(`group:"vef:cron:job_handlers"`)),
		fx.Private,
	),
	fx.Provide(NewRunEventPublisher, fx.Private),
	fx.Provide(newEngine, fx.Private),
	fx.Provide(newScheduleManager),
	fx.Provide(
		fx.Annotate(NewScheduleResource, fx.ResultTags(`group:"vef:api:resources"`)),
		fx.Annotate(NewRunResource, fx.ResultTags(`group:"vef:api:resources"`)),
	),

	fx.Invoke(registerLifecycle),
)

Module provides the durable schedule store: the handler registry, the claim/execute engine, the management surface, and the sys/cron resources. While vef.cron.store.enabled is false, the engine stays inert, the manager returns ErrStoreDisabled, and the resources mount no operations. Enabling the store activates migration, seeding, and execution through its lifecycle.

Functions

func NewRunResource 

func NewRunResource(cfg *config.CronConfig) api.Resource

NewRunResource creates the run journal resource. With the store disabled the resource mounts no operations.

func NewScheduleManager 

func NewScheduleManager(db orm.DB, enabled bool, registry *Registry, engine *Engine) cron.ScheduleManager

NewScheduleManager builds the management surface. With the store disabled it degrades to a stub that fails every call with cron.ErrStoreDisabled — the dependency stays injectable, the capability reports itself off.

func NewScheduleResource 

func NewScheduleResource(cfg *config.CronConfig, manager cron.ScheduleManager, registry *Registry) api.Resource

NewScheduleResource creates the schedule management resource. With the store disabled the resource mounts no operations — a feature that is off exposes no surface.

func SeedDefaultSchedules 

func SeedDefaultSchedules(ctx context.Context, manager cron.ScheduleManager, registry *Registry) error

SeedDefaultSchedules inserts every handler-shipped default schedule that does not exist yet. Existing rows — operator-tuned or created by a peer booting concurrently — are left untouched: shipping a default is an insert-if-absent, never an overwrite. An invalid shipped spec fails start-up; it is a programming error, not runtime input.

Types

type Engine 

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

Engine is the durable scheduler: it polls the store adaptively, claims due fires, executes them on a bounded pool, heartbeats running rows, recovers abandoned ones, and prunes the journal by retention. Every node runs one engine; the claim transaction is what keeps each fire single-noded.

func NewEngine 

func NewEngine(db orm.DB, cfg *config.CronStoreConfig, registry *Registry, publisher *RunEventPublisher) *Engine

NewEngine builds the engine; Start launches it.

func (*Engine) Start 

func (e *Engine) Start()

Start launches the claim loop and the heartbeat runner.

func (*Engine) Stop 

func (e *Engine) Stop(ctx context.Context) (stopErr error)

Stop drains gracefully: claiming stops immediately, running handlers get the drain window to finish, stragglers are canceled and journaled as canceled (recoverable schedules re-queue them; see complete). Heartbeats outlive handlers so draining runs stay owned. The caller's deadline bounds the whole sequence; when necessary, graceful drain is shortened to reserve the journal completion window.

func (*Engine) Wake 

func (e *Engine) Wake()

Wake nudges the loop to re-read the store now — called after local schedule mutations so a nearer fire does not wait out the current sleep.

type FiresPreview 

type FiresPreview struct {
	NextFiresUnixMs []int64 `json:"nextFiresUnixMs"`
}

FiresPreview is the preview_fires response: the trigger's upcoming fire times from now; empty when it yields no occurrence inside its window.

type PreviewFiresParams 

type PreviewFiresParams struct {
	api.P

	Trigger        TriggerParams `json:"trigger"`
	StartsAtUnixMs *int64        `json:"startsAtUnixMs"`
	EndsAtUnixMs   *int64        `json:"endsAtUnixMs"`
}

PreviewFiresParams carries an unsaved trigger whose upcoming fire times the editor wants to preview before persisting.

type Registry 

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

Registry indexes the registered JobHandlers by job name. It is immutable after construction; the engine claims only jobs this node can execute, so heterogeneous deployments (rolling updates, split job fleets) stay safe.

func NewRegistry 

func NewRegistry(handlers []cron.JobHandler) (*Registry, error)

NewRegistry builds the registry from the DI-collected handlers. Duplicate job names fail construction — and with it application start-up.

func (*Registry) All 

func (r *Registry) All() []cron.JobHandler

All returns every registered handler in name order.

func (*Registry) IsEmpty 

func (r *Registry) IsEmpty() bool

IsEmpty reports whether no handler is registered.

func (*Registry) Lookup 

func (r *Registry) Lookup(name string) (cron.JobHandler, bool)

Lookup returns the handler serving the job name.

func (*Registry) Names 

func (r *Registry) Names() []string

Names returns the registered job names, sorted. Callers must not mutate the returned slice.

type RunEventPublisher 

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

RunEventPublisher emits the store's operational notifications. Publishing is best-effort by design: the run journal is the durable truth, events only feed alerting. A bounded worker isolates transport back-pressure from claiming and executor slots; a full queue drops the notification.

func NewRunEventPublisher 

func NewRunEventPublisher(bus event.Bus) *RunEventPublisher

NewRunEventPublisher builds the publisher over the application event bus.

func (*RunEventPublisher) RunAbandoned 

func (p *RunEventPublisher) RunAbandoned(run *cron.Run)

RunAbandoned reports an abandoned run.

func (*RunEventPublisher) RunFailed 

func (p *RunEventPublisher) RunFailed(run *cron.Run)

RunFailed reports a failed run.

func (*RunEventPublisher) Start 

func (p *RunEventPublisher) Start()

Start launches the single publish worker.

func (*RunEventPublisher) Stop 

func (p *RunEventPublisher) Stop(ctx context.Context) error

Stop cancels the in-flight publish, flushes what the queue still holds, and waits within the caller's budget.

type RunResource 

type RunResource struct {
	api.Resource

	crud.FindPage[cron.Run, RunSearch]
	crud.FindOne[cron.Run, RunSearch]
}

RunResource exposes the run journal read-only: the paged view for browsing and the single-record view for the full error text.

type RunSearch 

type RunSearch struct {
	crud.Sortable

	// ID addresses one journal row; find_one has no other way to name the
	// record the caller means.
	ID                    string `json:"id" search:"eq,column=id"`
	ScheduleName          string `json:"scheduleName" search:"eq,column=schedule_name"`
	JobName               string `json:"jobName" search:"eq,column=job_name"`
	Status                string `json:"status" search:"eq"`
	NodeID                string `json:"nodeId" search:"eq,column=node_id"`
	ScheduledAtFromUnixMs *int64 `json:"scheduledAtFromUnixMs" search:"gte,column=scheduled_at_unix_ms"`
	ScheduledAtToUnixMs   *int64 `json:"scheduledAtToUnixMs" search:"lte,column=scheduled_at_unix_ms"`
}

RunSearch contains the search parameters for the run journal.

type ScheduleDetail 

type ScheduleDetail struct {
	Schedule *cron.Schedule `json:"schedule"`
	// NextFiresUnixMs previews the next few exact fire times from now. An
	// overdue cursor is projected through the schedule's misfire policy; a
	// paused or spent schedule returns an empty list.
	NextFiresUnixMs []int64 `json:"nextFiresUnixMs"`
}

ScheduleDetail is the get response: the schedule plus a preview of its upcoming fire times.

type ScheduleNameParams 

type ScheduleNameParams struct {
	api.P

	Name string `json:"name" validate:"required"`
}

ScheduleNameParams addresses one schedule by name.

type ScheduleParams 

type ScheduleParams struct {
	api.P

	Name              string                 `json:"name" validate:"required"`
	NewName           string                 `json:"newName"`
	JobName           string                 `json:"jobName" validate:"required"`
	Trigger           TriggerParams          `json:"trigger"`
	Params            json.RawMessage        `json:"params"`
	StartsAtUnixMs    *int64                 `json:"startsAtUnixMs"`
	EndsAtUnixMs      *int64                 `json:"endsAtUnixMs"`
	MisfirePolicy     cron.MisfirePolicy     `json:"misfirePolicy"`
	ConcurrencyPolicy cron.ConcurrencyPolicy `json:"concurrencyPolicy"`
	Recover           bool                   `json:"recover"`
	TimeoutMs         int64                  `json:"timeoutMs"`
	Enabled           *bool                  `json:"enabled"`
}

ScheduleParams contains the create/update parameters of a schedule. On update, Name addresses the schedule and NewName optionally renames it.

type ScheduleResource 

type ScheduleResource struct {
	api.Resource

	crud.FindPage[cron.Schedule, ScheduleSearch]
	// contains filtered or unexported fields
}

ScheduleResource manages durable schedules: paged browsing plus the control operations, all delegated to the ScheduleManager so API mutations and programmatic ones share one validation and wake path.

func (*ScheduleResource) Create 

func (r *ScheduleResource) Create(ctx fiber.Ctx, params ScheduleParams) error

Create persists a new schedule.

func (*ScheduleResource) Delete 

func (r *ScheduleResource) Delete(ctx fiber.Ctx, params ScheduleNameParams) error

Delete removes the schedule; its journaled runs are kept.

func (*ScheduleResource) Get 

func (r *ScheduleResource) Get(ctx fiber.Ctx, params ScheduleNameParams) error

Get returns one schedule with its upcoming-fire preview.

func (*ScheduleResource) ListJobs 

func (r *ScheduleResource) ListJobs(ctx fiber.Ctx) error

ListJobs returns the job names registered on this node — the vocabulary the schedule editor's job picker offers. Heterogeneous deployments may register different sets per node; the answering node's view is returned.

func (*ScheduleResource) Pause 

func (r *ScheduleResource) Pause(ctx fiber.Ctx, params ScheduleNameParams) error

Pause stops fire claiming until resume.

func (*ScheduleResource) PreviewFires 

func (r *ScheduleResource) PreviewFires(ctx fiber.Ctx, params PreviewFiresParams) error

PreviewFires projects the upcoming fire times of an unsaved trigger, so the editor validates an expression against the real parser before saving.

func (*ScheduleResource) Resume 

func (r *ScheduleResource) Resume(ctx fiber.Ctx, params ScheduleNameParams) error

Resume re-enables the schedule under its misfire policy.

func (*ScheduleResource) TriggerNow 

func (r *ScheduleResource) TriggerNow(ctx fiber.Ctx, params ScheduleNameParams) error

TriggerNow persists one immediate fire without moving the regular cursor.

func (*ScheduleResource) Update 

func (r *ScheduleResource) Update(ctx fiber.Ctx, params ScheduleParams) error

Update reshapes the named schedule; NewName renames it.

type ScheduleSearch 

type ScheduleSearch struct {
	crud.Sortable

	Name      string `json:"name" search:"contains"`
	JobName   string `json:"jobName" search:"eq,column=job_name"`
	Kind      string `json:"kind" search:"eq"`
	IsEnabled *bool  `json:"isEnabled" search:"eq,column=is_enabled"`
}

ScheduleSearch contains the search parameters for schedules.

type TriggerParams 

type TriggerParams struct {
	Kind     cron.TriggerKind `json:"kind" validate:"required"`
	Expr     *string          `json:"expr"`
	Timezone *string          `json:"timezone"`
	EveryMs  *int64           `json:"everyMs"`
	AtUnixMs *int64           `json:"atUnixMs"`
}

TriggerParams is the trigger section of a schedule mutation.

Source Files

View all Source files

Directories

Path Synopsis

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.