jobs

package
v0.0.6 Latest Latest
Warning

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

 Go to latest Published: Aug 19, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// ComponentName is the framework component name for the valkey-jobs
	// component. It is the identifier other components use in GetDependencies
	// to require it.
	ComponentName = "valkey-jobs"

	// ComponentStage is the stage data-layer components initialize in. It is
	// not a built-in bootstrap stage; AddComponent registers it automatically
	// the first time a component declares it.
	ComponentStage = cf.Stage("data")
)

Variables

View Source 
var (
	// ErrJobNotDead means the id is not a member of the dead ZSET (already
	// replayed, purged, or never dead-lettered).
	ErrJobNotDead = errors.New("cf_valkey_jobs: job is not in the dead-letter set")
	// ErrJobMissing means the id was in dead but the payload hash is gone
	// (retention expired). Replay cannot restore it.
	ErrJobMissing = errors.New("cf_valkey_jobs: job payload is gone")
	// ErrAlreadyEnqueued means WithID named a job that still exists (ready,
	// inflight, or dead hash). After ack the id may be reused.
	ErrAlreadyEnqueued = errors.New("cf_valkey_jobs: job id already exists")
	// ErrInvalidJobID means WithID used a reserved name (ready, inflight,
	// dead, cron) or contained ":", which would collide with index keys.
	ErrInvalidJobID = errors.New("cf_valkey_jobs: invalid job id")
)

Sentinel errors for dead-letter operator calls (ListDead / Replay / PurgeDead).

Functions

This section is empty.

Types

type CFValkeyJobs 

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

CFValkeyJobs is the caerus-framework-valkey-jobs component: a lightweight delayed task queue over a cf_valkey.CFValkey peer. It is a stateless consumer of the peer (never a client snapshot) and builds every command through the peer's live Client() and prefix-aware Key(), so reconnects and key prefixes stay consistent.

Delivery is at-least-once: a job runs, its handler either acknowledges it, requeues it with a retry delay, or (attempts exhausted) dead-letters it. A crashed or hung worker is recovered via the visibility timeout (the job is re-queued after its deadline). Handlers must be idempotent.

func New 

func New(opts ...Option) *CFValkeyJobs

New creates a jobs component. The valkey peer is resolved at Init, not here.

func (*CFValkeyJobs) Client 

func (c *CFValkeyJobs) Client() valkey.Client

Client returns the peer's live valkey client (nil before Init or after Shutdown). Useful for direct commands against the same key space.

func (*CFValkeyJobs) Enqueue 

func (c *CFValkeyJobs) Enqueue(ctx context.Context, jobType string, payload []byte, opts ...EnqueueOption) (string, error)

Enqueue stores a job with a JSON-ish opaque payload. The job type names the handler that must be registered on a worker. Returns the job id.

func (*CFValkeyJobs) GetDependencies 

func (c *CFValkeyJobs) GetDependencies() []string

GetDependencies implements cf.Dependencies. The component depends on the valkey component it consumes (the actual peer name when WithValkeyName is set, the default ComponentName otherwise), logs through the framework logs component, and depends on configuration when WithConfigSource is set. Peer names are fixed at construction, so the graph is stable before Init.

func (*CFValkeyJobs) GetInitOrderStage 

func (c *CFValkeyJobs) GetInitOrderStage() cf.Stage

GetInitOrderStage implements cf.CaerusComponent.

func (*CFValkeyJobs) Health 

func (c *CFValkeyJobs) Health(ctx context.Context) error

Health implements cf.HealthProvider. It reports healthy while the peer's valkey client is initialized; real connectivity is owned by the valkey component's own Health (aggregated by observability's /readyz).

func (*CFValkeyJobs) Init 

func (c *CFValkeyJobs) Init(ctx context.Context, fw *cf.CaerusFramework) error

Init implements cf.CaerusComponent. It resolves the valkey peer component (by name or the default "valkey"), failing fast when it is missing or not yet initialized. No connection is opened here; the peer owns its client.

func (*CFValkeyJobs) ListDead 

func (c *CFValkeyJobs) ListDead(ctx context.Context, offset, limit int64) ([]Job, error)

ListDead returns a page of dead-lettered jobs, oldest dead-letter first (ZSET score = dead-at). offset is 0-based. limit is capped at 100; a non-positive limit means that cap.

func (*CFValkeyJobs) Metrics 

func (c *CFValkeyJobs) Metrics() []cf_observability.Metric

Metrics implements cf_observability.MetricsProvider. It reports operation counters (per job type) while the peer's client is initialized; before Init or after Shutdown it returns nil, so the observability component skips it (lazy pickup). Counters are cumulative for the process lifetime.

func (*CFValkeyJobs) Name 

func (c *CFValkeyJobs) Name() string

Name implements cf.CaerusComponent. Returns the custom name set via WithName, or the default ComponentName ("valkey-jobs") if no custom name was set.

func (*CFValkeyJobs) OnConfigReload 

func (c *CFValkeyJobs) OnConfigReload(source string, cfg any)

OnConfigReload implements cf.ConfigReloader. It re-reads the worker tunables from the bound configuration source; the poll loop picks them up on the next tick, and a concurrency change swaps the semaphore.

func (*CFValkeyJobs) PurgeDead 

func (c *CFValkeyJobs) PurgeDead(ctx context.Context, id string) error

PurgeDead deletes one dead-lettered job (ZSET member + payload hash). Returns ErrJobNotDead if the id is not in the dead set (does not delete a ready or inflight hash).

func (*CFValkeyJobs) PurgeDeadAll 

func (c *CFValkeyJobs) PurgeDeadAll(ctx context.Context) (int64, error)

PurgeDeadAll deletes every dead-lettered job. The method name is the confirmation; an empty DLQ succeeds with a zero count.

func (*CFValkeyJobs) RegisterConfigSources 

func (c *CFValkeyJobs) RegisterConfigSources(conf any) error

RegisterConfigSources implements cf.ConfigSourceRegistrar. The framework calls it during argv absorption; it registers this component's configuration source (name, path, env prefix, format, Owner) with the configuration component. No-op when no source is bound.

func (*CFValkeyJobs) Replay 

func (c *CFValkeyJobs) Replay(ctx context.Context, id string) error

Replay moves a dead-lettered job back to ready with the same id, attempts and retry bookkeeping cleared, due now. The next claim is attempt 1 of max_attempts again. Returns ErrJobNotDead or ErrJobMissing when the row cannot be restored.

func (*CFValkeyJobs) Run 

func (c *CFValkeyJobs) Run(ctx context.Context) error

Run implements cf.Runnable. It polls until ctx is canceled, then drains claimed jobs back to ready. Each poll checks worker_enabled so a reload can stop or resume claiming without a process restart. No handlers still means the loop does not claim.

func (*CFValkeyJobs) Shutdown 

func (c *CFValkeyJobs) Shutdown(ctx context.Context) error

Shutdown implements cf.CaerusComponent. It unsubscribes the logs subscription and drops the valkey peer. Further use returns an error.

type EnqueueOption 

type EnqueueOption func(*enqueueOptions)

EnqueueOption configures a single Enqueue call.

func WithDelay 

func WithDelay(d time.Duration) EnqueueOption

WithDelay schedules the job d from now (overrides WithRunAt).

func WithID 

func WithID(id string) EnqueueOption

WithID sets a stable job id for this Enqueue. A second Enqueue with the same id while the hash still exists returns ErrAlreadyEnqueued (at-most-one copy in the queue). This is not exactly-once *execution*: a handler may still run more than once if visibility expires. Empty id is ignored. Ids must be a single path segment: no ":", and not ready / inflight / dead / cron (those are the index keys). Invalid ids return ErrInvalidJobID before any Valkey write.

func WithMaxAttempts 

func WithMaxAttempts(n int64) EnqueueOption

WithMaxAttempts sets the max number of runs before dead-lettering (default 3).

func WithRetention 

func WithRetention(d time.Duration) EnqueueOption

WithRetention sets how long the job payload is kept after enqueue (and after dead-lettering) for inspection (default 7d).

func WithRunAt 

func WithRunAt(t time.Time) EnqueueOption

WithRunAt schedules the job to become claimable at t. Zero or a past time makes it immediately claimable.

func WithVisibility 

func WithVisibility(d time.Duration) EnqueueOption

WithVisibility sets how long a claimed job stays owned by a worker before it is considered hung and retried (default 1m). Set it well above the handler's expected runtime.

type Job 

type Job struct {
	ID          string
	Type        string
	Payload     []byte
	Attempts    int64 // 1-based attempt number of this run (stored count on ListDead)
	MaxAttempts int64
	CreatedAt   time.Time
	// DeadAt is set by ListDead (ZSET score). Zero on a claimed handler job.
	DeadAt time.Time
}

Job is a claimed job handed to a JobHandler, or a dead-letter row from ListDead.

type JobHandler 

type JobHandler func(ctx context.Context, job Job) error

JobHandler processes one claimed job. It must honor ctx cancellation and be idempotent: delivery is at-least-once, and a hung handler is retried after its visibility timeout expires. A nil error acknowledges the job; any error requeues it per the retry policy (dead-lettered once attempts are exhausted).

type JobsConfig 

type JobsConfig struct {
	// WorkerEnabled turns the worker loop on or off. Nil (key omitted) keeps
	// the construct default (WithWorkerEnabled, else on when handlers exist).
	// Explicit false is how a producer-only replica stops claiming.
	WorkerEnabled *bool `json:"worker_enabled,omitempty" yaml:"worker_enabled,omitempty" env:"WORKER_ENABLED"`
	// PollIntervalMs is the worker poll cadence in ms (default 500).
	PollIntervalMs int64 `json:"poll_interval_ms,omitempty" yaml:"poll_interval_ms,omitempty" env:"POLL_INTERVAL_MS"`
	// BatchSize is the max jobs claimed per poll (default 16).
	BatchSize int64 `json:"batch_size,omitempty" yaml:"batch_size,omitempty" env:"BATCH_SIZE"`
	// Concurrency is the max number of handlers running at once (default 8).
	Concurrency int64 `json:"concurrency,omitempty" yaml:"concurrency,omitempty" env:"CONCURRENCY"`
	// RetryFixedDelayMs is the delay between attempts while in the fixed phase
	// (default 5000).
	RetryFixedDelayMs int64 `json:"retry_fixed_delay_ms,omitempty" yaml:"retry_fixed_delay_ms,omitempty" env:"RETRY_FIXED_DELAY_MS"`
	// RetryFixedPhaseMs is how long a failing job retries on the fixed delay
	// before switching to jittered exponential backoff (default 30000).
	RetryFixedPhaseMs int64 `json:"retry_fixed_phase_ms,omitempty" yaml:"retry_fixed_phase_ms,omitempty" env:"RETRY_FIXED_PHASE_MS"`
	// RetryMaxDelayMs caps the backoff (default 300000).
	RetryMaxDelayMs int64 `json:"retry_max_delay_ms,omitempty" yaml:"retry_max_delay_ms,omitempty" env:"RETRY_MAX_DELAY_MS"`
	// RetryJitter is the backoff jitter as a fraction around the base delay
	// (default 0.5 → delay in [0.5x, 1.5x]). Nil keeps the construct default.
	// Explicit zero disables jitter.
	RetryJitter *float64 `json:"retry_jitter,omitempty" yaml:"retry_jitter,omitempty" env:"RETRY_JITTER"`
}

JobsConfig is the file/env-drivable behavior configuration. Load it through the configuration component (caerus-framework-configuration) and pass it via WithConfigSource; both JSON and YAML tags are provided.

type Option 

type Option func(*options)

Option configures the component at construction time.

func WithBatchSize 

func WithBatchSize(n int64) Option

WithBatchSize sets the max jobs claimed per poll (default 16).

func WithConcurrency 

func WithConcurrency(n int64) Option

WithConcurrency sets the max number of handlers running at once (default 8).

func WithConfig 

func WithConfig(cfg JobsConfig) Option

WithConfig sets a static configuration snapshot. Non-zero fields of cfg override the values set by the convenience options. Prefer WithConfigSource when using caerus-framework-configuration with hot-reload.

func WithConfigSource 

func WithConfigSource(name, path string, opts ...SourceOption) Option

WithConfigSource binds this component to a named configuration source and registers that source with the configuration component (via the framework's ConfigSourceRegistrar pass during argv absorption). The module owns the Source: the config type, the default EnvPrefix and its Owner (Name(), so named instances reload correctly). main only points the instance at where the config lives. 

cf_valkey_jobs.New(cf_valkey_jobs.WithConfigSource("jobs", "config/jobs.json"))

A path of "" registers an env-only (fileless) source when the EnvPrefix is non-empty. The path CLI override stays --<source-name> (ParseFlags). Declares a dependency on "configuration".

func WithJobHandler 

func WithJobHandler(jobType string, fn JobHandler) Option

WithJobHandler registers a handler for a job type. Jobs of that type claimed by the worker run fn; handlers must be idempotent (at-least-once delivery). Registering at least one handler enables the worker loop; enqueue works regardless.

func WithLogger 

func WithLogger(logger *slog.Logger) Option

WithLogger overrides the logger used for component diagnostics. By default the component logs through the framework logs component (declared in GetDependencies); WithLogger is an explicit override for tests and embedded use and wins over the framework logger. slog.Default() remains the fallback only when neither is available.

func WithName 

func WithName(name string) Option

WithName sets a custom component name, allowing multiple jobs instances in the same process. The default name is "valkey-jobs" (ComponentName). Retrieve named instances with GetByName[*CFValkeyJobs](fw, "jobs").

func WithPollInterval 

func WithPollInterval(d time.Duration) Option

WithPollInterval sets the worker poll cadence (default 500ms).

func WithRepeat 

func WithRepeat(jobType string, every time.Duration, payload []byte) Option

WithRepeat enqueues jobType on a fixed interval while Run is alive. This is not a crontab (no calendar, no TZ, no catch-up of missed ticks). Several replicas share a valkey NX lock so only one fire happens per interval. every shorter than 1s is raised to 1s.

func WithRetryPolicy 

func WithRetryPolicy(fixedDelay, fixedPhase, maxDelay time.Duration) Option

WithRetryPolicy sets the mixed retry policy: fixedDelay applies while a job has been failing for less than fixedPhase; after that, jittered exponential backoff (counting attempts since the phase switch) grows toward maxDelay.

func WithShutdownDrainTimeout 

func WithShutdownDrainTimeout(d time.Duration) Option

WithShutdownDrainTimeout bounds how long Run waits for in-flight handlers to finish after ctx cancellation before draining claimed jobs back to ready (default 10s).

func WithValkeyName 

func WithValkeyName(name string) Option

WithValkeyName binds the component to a valkey component with the given name (WithName on the valkey side). The default is the valkey ComponentName ("valkey").

func WithWorkerEnabled 

func WithWorkerEnabled(on bool) Option

WithWorkerEnabled forces the worker loop on/off regardless of handlers (default: enabled when any handler is registered).

type SourceOption 

type SourceOption func(*sourceOptions)

SourceOption configures the self-registered configuration source created by WithConfigSource.

func WithSourceEnvPrefix 

func WithSourceEnvPrefix(prefix string) SourceOption

WithSourceEnvPrefix sets the environment overlay prefix for the source (default: the uppercase source name with "-" replaced by "_", plus "_"). An empty prefix disables env overlay.

func WithSourceFormat 

func WithSourceFormat(f cf_configuration.Format) SourceOption

WithSourceFormat forces the file format instead of inferring it from the path extension (".yaml"/".yml" → YAML; anything else JSON).

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.