cf_valkey_jobs

package module

v0.0.7 Latest Latest Go to latest Published: Aug 16, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

README ¶

caerus-framework-valkey-jobs

Caerus Framework Valkey Jobs Component. A lightweight delayed task queue built on valkey sorted sets and Lua: enqueue a job with an optional run-at time, a worker claims due jobs and runs registered handlers, failed jobs retry with a mixed fixed/jittered-exponential policy, and jobs that exhaust their attempts (or have no handler) go to a dead-letter set for inspection.

Delivery is at-least-once: a handler may run more than once (crash between claim and ack, or a visibility-timeout recovery), so handlers must be idempotent. There is no cross-job ordering or exactly-once guarantee.

Wiring

Two wiring shapes are supported. Prefer the app-owned shape (demoapp golden path): main declares only the chassis (valkey-jobs alongside postgres / valkey) and the app class; product machinery that produces/consumes jobs lives under the app and resolves valkey-jobs as a peer at Init. Use the simple main-level shape for one-off binaries.

App-owned consumer (golden — demoapp pattern)

main declares valkey-jobs as chassis and runs the app class; it never touches valkey-jobs itself:

fw := cf.New(&cf.FrameworkOptions{
	Logs: &cf.LogsSettings{Format: "json", Level: "info", ConfigSource: "logs"},
	Observability: &cf.ObservabilitySettings{Bind: ":9090", ConfigSource: "observability"},
	Components: []cf.CaerusComponent{
		cf_valkey.New(cf_valkey.WithConfigSource("valkey", "config/valkey.json")),
		cf_valkey_jobs.New(cf_valkey_jobs.WithConfigSource("jobs", "config/jobs.json")),
		app.New(app.Options{}),
	},
})
if err := fw.RunWithSignals(context.Background()); err != nil {
	log.Fatal(err)
}

The app resolves the valkey-jobs component pointer once at Init (never a client snapshot), declares it in GetDependencies, and calls Enqueue / registers handlers per use:

type App struct {
	jobs *cf_valkey_jobs.CFValkeyJobs
}

func (a *App) GetDependencies() []string {
	return []string{cf_valkey_jobs.ComponentName} // + logs, chassis peers
}

func (a *App) Init(ctx context.Context, fw *cf.CaerusFramework) error {
	j, ok := cf.Get[*cf_valkey_jobs.CFValkeyJobs](fw)
	if !ok {
		return errors.New("app: jobs component missing")
	}
	a.jobs = j
	return nil
}

Simple `main`-level wiring

For a one-off binary, register the components directly and use cf.MustGet to reach the component:

fw := cf.New()

logs := cf_logs.New(cf_logs.WithWriter(os.Stdout))
vk := cf_valkey.New(cf_valkey.WithConfigSource("valkey", "config/valkey.json"))
jobs := cf_valkey_jobs.New(cf_valkey_jobs.WithConfigSource("jobs", "config/jobs.json"))
fw.AddComponent(logs)
fw.AddComponent(vk)
fw.AddComponent(jobs) // GetDependencies() -> [valkey logs configuration]

In both shapes the component is cf.ConfigSourceRegistrar-self-sufficient: WithConfigSource registers the Source[JobsConfig] with the configuration component during argv absorption, so main never touches os.Getenv/ParseFlags. The --jobs path flag and per-field flags come from the source declaration.

Usage

Register handlers before the framework starts. The worker is only enabled when at least one handler is registered and worker_enabled is true (default).

jobs := cf_valkey_jobs.New(
	cf_valkey_jobs.WithConfigSource("jobs", "config/jobs.json"),
	cf_valkey_jobs.WithJobHandler("email.send", func(ctx context.Context, job cf_valkey_jobs.Job) error {
		// job.Payload is the raw bytes; return an error to retry.
		return sendEmail(ctx, job.Payload)
	}),
)

// elsewhere, via the app's resolved peer
id, err := app.jobs.Enqueue(ctx, "email.send", []byte(`{"to":"a@b.c"}`),
	cf_valkey_jobs.WithDelay(5*time.Minute),
	cf_valkey_jobs.WithMaxAttempts(5),
)

Enqueue returns the job id. Scheduling options:

Option	Description
`WithRunAt(t)`	claimable from `t` (zero/past = immediately)
`WithDelay(d)`	run `d` from now (overrides `WithRunAt`)
`WithMaxAttempts(n)`	max runs before dead-letter (default `3`)
`WithVisibility(d)`	time a claimed job stays hidden before recovery (default `1m`)
`WithRetention(d)`	TTL on the job payload (default `7d`)

Retry policy

On a handler error the job is requeued with a mixed policy:

While now − first-failure ≤ retry_fixed_phase (default 30s), retry every retry_fixed_delay (default 5s).
Past the fixed phase, back off exponentially counting attempts since the phase switch: retry_fixed_delay · 2^(n−1) for n = attempts − jitter_base, jittered ±retry_jitter (default 50%), capped at retry_max_delay (default 5m).

When a job exhausts max_attempts (or no handler is registered for its type) it is dead-lettered into the jobs:dead zset for inspection; the payload hash survives until retention expires. Failed runs increment attempts in the payload hash, so Job.Attempts seen by a handler is 1-based.

Options

Option	Description
`WithConfig(JobsConfig)`	static config snapshot; non-zero fields override option-set defaults
`WithConfigSource(name, path, …)`	bind a configuration source for Init + `OnConfigReload`; the module registers the `Source[JobsConfig]` itself (declares `configuration` dep)
`WithJobHandler(type, fn)`	register a handler for a job type
`WithWorkerEnabled(on)`	force the worker on/off (default on once handlers exist)
`WithPollInterval(d)`	worker poll cadence (default `500ms`)
`WithBatchSize(n)`	max jobs claimed per pass (default `16`)
`WithConcurrency(n)`	max concurrent handler runs (default `8`)
`WithRetryPolicy(fixed, phase, max)`	retry tunables (defaults 5s / 30s / 5m)
`WithShutdownDrainTimeout(d)`	max time to wait for running handlers on shutdown (default `10s`)
`WithValkeyName(name)`	name of the valkey peer to use (default `"valkey"`)
`WithName(name)`	custom component name for multiple instances (default `"valkey-jobs"`)
`WithLogger(*slog.Logger)`	explicit logger override; defaults to the framework `logs` component's logger (re-delivered on `logs` `Reconfigure`), falling back to `slog.Default()`

Configuration

Load JobsConfig through the configuration component. The default EnvPrefix is JOBS_ (from the source name). Worker tunables reload live via OnConfigReload; the retry math and visibility are read fresh per decision.

{
  "worker_enabled": true,
  "poll_interval_ms": 500,
  "batch_size": 16,
  "concurrency": 8,
  "retry_fixed_delay_ms": 5000,
  "retry_fixed_phase_ms": 30000,
  "retry_max_delay_ms": 300000,
  "retry_jitter": 0.5
}

Health reports healthy once the valkey peer is initialized (client present); before Init it reports unhealthy. Metrics emits the following while initialized, nil before Init/after Shutdown:

Metric	Type	Labels
`valkey_jobs_info`	gauge 1	`component`
`valkey_jobs_config_reloads_total`	counter	`component`
`valkey_jobs_enqueued_total`	counter	`component`, `type`
`valkey_jobs_run_total`	counter	`component`, `type`
`valkey_jobs_failed_total`	counter	`component`, `type`
`valkey_jobs_requeued_total`	counter	`component`, `type`
`valkey_jobs_dead_total`	counter	`component`, `type`
`valkey_jobs_duration_seconds_sum`	counter	`component`, `type`
`valkey_jobs_duration_seconds_count`	counter	`component`, `type`

Queued work is visible on the valkey keys directly: jobs:ready (score = due ms), jobs:inflight (score = visibility deadline), jobs:dead (score = dead-lettered-at), and a jobs:<id> payload hash per job. Dead-lettered jobs can be re-enqueued manually by re-adding the id to jobs:ready with a due score.

Key layout

Key	Type	Meaning
`<prefix>jobs:ready`	ZSET	due score in ms; poll targets score ≤ now
`<prefix>jobs:inflight`	ZSET	visibility deadline in ms; reaper recovers expired
`<prefix>jobs:dead`	ZSET	dead-lettered at ms; inspection only
`<prefix>jobs:<id>`	HASH	`type`, `payload`, `max_attempts`, `created_ms`, `attempts`, `retry_start_ms`, `jitter_base`, `visibility_ms`; TTL = retention

Tests

Unit tests cover the Init contract, dependencies (named peers, config source), retry math (fixed phase, jitter phase, cap, jitter range), option layering, and claim decoding — no external service.

Integration tests (integration_test.go) run against a real valkey when VALKEY_ADDR is set: happy-path run, scheduling with WithDelay/WithRunAt, retry-then-dead-letter, no-handler dead-letter, visibility recovery of a hung job, the concurrency bound, shutdown-drain requeue, and named-peer resolution.

docker run -d --rm -p 6379:6379 --name v valkey/valkey:8
VALKEY_ADDR=127.0.0.1:6379 go test -race ./...

License

Apache License 2.0 — see LICENSE.

Documentation ¶

Index ¶

Constants
type CFValkeyJobs
- func New(opts ...Option) *CFValkeyJobs
type EnqueueOption
type Job
type JobHandler
type JobsConfig
type Option
type SourceOption
- func WithSourceEnvPrefix(prefix string) SourceOption
- func WithSourceFormat(f cf_configuration.Format) SourceOption

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 ¶

This section is empty.

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) 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) 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) Run ¶

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

Run implements cf.Runnable. It runs the worker poll loop until ctx is canceled, then drains claimed jobs back to the ready queue and returns promptly. No-op when no handlers are registered or worker_enabled is false.

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 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
	MaxAttempts int64
	CreatedAt   time.Time
}

Job is a claimed job handed to a JobHandler.

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 forces the worker loop off when false even if handlers are
	// registered (e.g. a producer-only deployment). Default true.
	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]). 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 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

jobs.go

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL