PBVex

Appearance

scheduler ​

go
import "github.com/nathabonfim59/pbvex/backend/internal/scheduler"

Index ​

Constants ​

go
const (
    JobTypeScheduled = "scheduled"

    JobStatusPending   = "pending"
    JobStatusRunning   = "running"
    JobStatusCanceled  = "canceled"
    JobStatusFailed    = "failed"
    JobStatusCompleted = "completed"
)

Variables ​

go
var (
    ErrJobNotFound                = errors.New("job not found")
    ErrJobNotCancelable           = errors.New("job cannot be canceled")
    ErrJobNotRetryable            = errors.New("job cannot be retried")
    ErrJobInvalidStatus           = errors.New("invalid job status filter")
    ErrDeploymentSnapshotNotFound = errors.New("deployment snapshot not found")
)

type Clock ​

Clock abstracts time for the scheduler so tests can control timing.

go
type Clock interface {
    Now() time.Time
    After(time.Duration) <-chan time.Time
    NewTicker(time.Duration) Ticker
}

func NewRealClock ​

go
func NewRealClock() Clock

type Config ​

Config controls the scheduler service.

go
type Config struct {
    PollInterval         time.Duration
    MaxConcurrency       int
    LeaseDuration        time.Duration
    RenewInterval        time.Duration
    MaxExecutionDuration time.Duration
    ClaimBatch           int
    RetryInitialDelay    time.Duration
    RetryMaxDelay        time.Duration
    MaxAttempts          int
    CleanupInterval      time.Duration
    CleanupBatch         int
    JobHistoryRetention  time.Duration
    Jitter               func(time.Duration) time.Duration
    Clock                Clock
}

func DefaultConfig ​

go
func DefaultConfig() Config

DefaultConfig returns sensible defaults for the scheduler.

type CronManager ​

CronManager mirrors the active deployment's recurring definitions into the PocketBase app-level cron registry. Cron ticks enqueue ordinary durable PBVex jobs rather than invoking application code in PocketBase's cron goroutine.

go
type CronManager struct {
    // contains filtered or unexported fields
}

func NewCronManager ​

go
func NewCronManager(app core.App, enqueuer cronEnqueuer) *CronManager

func (*CronManager) ActiveDeploymentChanged ​

go
func (m *CronManager) ActiveDeploymentChanged(deploymentID string, manifest deploy.DeploymentManifest)

ActiveDeploymentChanged implements deploy.ActivationObserver.

func (*CronManager) Clear ​

go
func (m *CronManager) Clear()

Clear removes only jobs registered by this manager, preserving PocketBase built-ins and jobs installed by Go or JS extensions.

type FakeClock ​

FakeClock is a test-only clock that can be advanced manually. It is safe for concurrent use by the worker and tests.

go
type FakeClock struct {
    // contains filtered or unexported fields
}

func NewFakeClock ​

go
func NewFakeClock(now time.Time) *FakeClock

func (*FakeClock) Advance ​

go
func (c *FakeClock) Advance(d time.Duration)

func (*FakeClock) After ​

go
func (c *FakeClock) After(time.Duration) <-chan time.Time

func (*FakeClock) NewTicker ​

go
func (c *FakeClock) NewTicker(d time.Duration) Ticker

func (*FakeClock) Now ​

go
func (c *FakeClock) Now() time.Time

type JobExecutor ​

JobExecutor runs a scheduled function and returns its wire-encoded result. Resolve exposes the deployment snapshot that the scheduler pins per job. Pin atomically increments/decrements the per-deployment job reference counter.

go
type JobExecutor interface {
    InvokeDeploymentSnapshot(ctx context.Context, deploymentID, bundleHash, functionName string, args any) (any, error)
    Resolve(ctx context.Context, deploymentID string) (deploy.DeploymentManifest, string, string, error)
    Pin(ctx context.Context, deploymentID string, delta int) error
}

type JobStatus ​

JobStatus is the operator-observable view of a job. It never includes the raw payload or args.

go
type JobStatus struct {
    ID             string    `json:"id"`
    DeploymentID   string    `json:"deploymentId"`
    Type           string    `json:"type"`
    Status         string    `json:"status"`
    ScheduledAt    time.Time `json:"scheduledAt"`
    Started        time.Time `json:"started,omitempty"`
    Finished       time.Time `json:"finished,omitempty"`
    Attempts       int       `json:"attempts"`
    Lease          string    `json:"-"`
    LeaseExpiresAt time.Time `json:"leaseExpiresAt,omitempty"`
    Result         any       `json:"result,omitempty"`
    Error          string    `json:"error,omitempty"`
    Metadata       any       `json:"metadata,omitempty"`
}

type ListResult ​

ListResult is returned by List.

go
type ListResult struct {
    Total      int         `json:"total"`
    Items      []JobStatus `json:"items"`
    Limit      int         `json:"limit"`
    NextCursor string      `json:"nextCursor,omitempty"`
    HasMore    bool        `json:"hasMore"`
}

type RealClock ​

RealClock uses the wall clock.

go
type RealClock struct{}

func (RealClock) After ​

go
func (RealClock) After(d time.Duration) <-chan time.Time

func (RealClock) NewTicker ​

go
func (RealClock) NewTicker(d time.Duration) Ticker

func (RealClock) Now ​

go
func (RealClock) Now() time.Time

type Service ​

Service is the durable scheduler implementation.

go
type Service struct {
    // contains filtered or unexported fields
}

func NewService ​

go
func NewService(app core.App, executor JobExecutor, config Config) *Service

NewService creates a new scheduler service. Call Start to begin processing.

func (*Service) Cancel ​

go
func (s *Service) Cancel(ctx context.Context, id string) error

Cancel marks a pending or running job as canceled. On lease-CAS miss (the lease changed between read and update due to a concurrent reclaim), it reloads and retries boundedly while the job is still cancelable.

func (*Service) Get ​

go
func (s *Service) Get(ctx context.Context, id string) (*JobStatus, error)

Get returns a single job by id.

func (*Service) List ​

go
func (s *Service) List(ctx context.Context, status string, limit int, cursor string) (*ListResult, error)

List returns jobs with optional status filter and keyset pagination. status is one of the JobStatus* constants or empty for all.

func (*Service) Retry ​

go
func (s *Service) Retry(ctx context.Context, id string) error

Retry requeues a terminal job. It rejects the request if the deployment snapshot the job is pinned to has been trimmed or replaced so that a worker cannot be asked to invoke a missing snapshot.

func (*Service) RunAfter ​

go
func (s *Service) RunAfter(ctx context.Context, delayMs int64, deploymentID, functionName string, args any) (string, error)

RunAfter schedules a function to run after delayMs.

func (*Service) RunAt ​

go
func (s *Service) RunAt(ctx context.Context, epochMs int64, deploymentID, functionName string, args any) (string, error)

RunAt schedules a function to run at a wall-clock time.

func (*Service) Start ​

go
func (s *Service) Start(ctx context.Context) error

Start begins the worker and cleanup loops.

func (*Service) Stop ​

go
func (s *Service) Stop()

Stop gracefully shuts down the worker and cleanup loops.

type Ticker ​

Ticker is a reusable time-driven signal that can be stopped.

go
type Ticker interface {
    Chan() <-chan time.Time
    Stop()
}

type Worker ​

Worker is the bounded background processor for scheduled jobs. It claims rows with a CAS on the observed status, lease token, and expiry so stale workers can never overwrite a new owner.

Once a job is claimed, a per-attempt lease token is assigned. A heartbeat goroutine renews the lease while the job runs. If the lease is lost (another worker steals the row), the running invocation is canceled. Completion, failure, retry, and release all require the same lease token.

This gives at-least-once delivery: a job may be run by multiple workers if a previous lease expires, but a stale worker cannot record a result after a new owner has taken over.

go
type Worker struct {
    // contains filtered or unexported fields
}

func NewWorker ​

go
func NewWorker(service *Service) *Worker

NewWorker creates a worker bound to the given service.

func (*Worker) Cancel ​

go
func (w *Worker) Cancel(id, leaseToken string) bool

Cancel requests that a running attempt stop. The leaseToken scopes the cancel to the current attempt, so a stale cancellation cannot accidentally remove the cancel func of a reclaimed attempt.

func (*Worker) Start ​

go
func (w *Worker) Start(ctx context.Context) error

Start begins the polling loop.

func (*Worker) Stop ​

go
func (w *Worker) Stop()

Stop cancels the worker and waits for all goroutines to finish.

func (*Worker) Wake ​

go
func (w *Worker) Wake()

Wake triggers an immediate poll attempt.

Generated by gomarkdoc

Generated API reference. Source of truth is the codebase.

Copyright © 2026 PBVex authors