vpq

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: 16 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// ComponentName is the framework component name for the valkey priority
	// queue component.
	ComponentName = "vpq"

	// ComponentStage is the stage the queue initializes 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 ErrClosed = errors.New("cf_vpq: queue is not initialized or is shut down")

ErrClosed is returned by queue operations after Shutdown or before Init.

Functions

This section is empty.

Types

type BGetObject 

type BGetObject struct {
	ObjectID    string
	ObjectScore float64
	ObjectValue string
}

BGetObject is an item popped from the queue. ObjectScore is the item weight (the number of times it was added since it last left the queue).

type Handler 

type Handler func(context.Context, *BGetObject) error

Handler processes one item in the auto-consumer loop. Returning an error requeues the item (weight +1) and drops its deadlock tracking. Honour ctx for shutdown and deadlines.

type InFlightItem 

type InFlightItem struct {
	ObjectID string
	PoppedAt time.Time
}

InFlightItem is an item that was popped (removed from the queue) but not yet acked or requeued.

type Option 

type Option func(*options)

Option configures the queue at construction time.

func WithBlockDuration 

func WithBlockDuration(d time.Duration) Option

WithBlockDuration sets how long a blocking pop waits for an item (default 1s).

func WithCacheTimeout 

func WithCacheTimeout(d time.Duration) Option

WithCacheTimeout sets how long an item may remain queued before it is discarded (default 0 = no residence limit). Expiry removes the zqueue member and payload together; the payload key itself is never Redis-EXPIRE'd, so partial "ghost" expiry cannot occur. In-flight items are unaffected.

func WithConfig 

func WithConfig(cfg PQConfig) Option

WithConfig sets a static queue 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, default EnvPrefix and its Owner (Name(), so named instances reload correctly). main only points the instance at where the config lives. 

cf_vpq.New(cf_vpq.WithConfigSource("vpq", "config/vpq.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". Queue identity (name/prefix) is applied at Init; reload updates tunables only (durations, recover, health thresholds).

func WithHandler 

func WithHandler(h Handler) Option

WithHandler sets the auto-consumer callback used by Run. When set, deadlock recovery runs every 30s by default (see WithRecoverInterval), and Health gets default MaxDepth / MaxInFlight thresholds (see WithMaxDepth / WithMaxInFlight). Handlers must be safe for concurrent use when WithWorkers(n) has n > 1.

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 WithMaxDepth 

func WithMaxDepth(n int64) Option

WithMaxDepth fails Health when the queued item count exceeds n. With a handler, the default is 10000 when this option is omitted; pass 0 to disable the check explicitly.

func WithMaxInFlight 

func WithMaxInFlight(n int64) Option

WithMaxInFlight fails Health when popped-but-unacked items exceed n. With a handler, the default is max(64, workers*16) when this option is omitted; pass 0 to disable the check explicitly.

func WithName 

func WithName(name string) Option

WithName sets a custom component name, allowing multiple VPQ instances in the same process. The default name is "vpq" (ComponentName). Use this when you need multiple queues (e.g., email and billing) in one binary. Retrieve named instances with GetByName[*PriorityQueue](fw, "email").

func WithPollInterval 

func WithPollInterval(d time.Duration) Option

WithPollInterval sets the fallback consumer poll interval (default 1s).

func WithPublishWatermarkDelay 

func WithPublishWatermarkDelay(d time.Duration) Option

WithPublishWatermarkDelay sets the minimum interval between pub/sub notifications on Add (default 0 = no pub/sub; consumers then rely on the poll interval).

func WithQueueName 

func WithQueueName(name string) Option

WithQueueName sets the queue name. It is the pub/sub channel and the key namespace segment; it is required.

func WithRecoverInterval 

func WithRecoverInterval(d time.Duration) Option

WithRecoverInterval sets how often Run calls RecoverDeadlocked. With a handler, the default is 30s. Pass 0 to disable. Without a handler, recovery stays off unless this sets a positive interval (dedicated recoverer).

func WithRecoverMaxAge 

func WithRecoverMaxAge(d time.Duration) Option

WithRecoverMaxAge sets the minimum age of in-flight items before RecoverDeadlocked requeues them (default 5m). Pass 0 to recover all currently in-flight items on each tick.

func WithValkeyName 

func WithValkeyName(name string) Option

WithValkeyName binds this queue to a valkey component Name() other than the default "valkey". GetDependencies reports that name.

func WithWorkers 

func WithWorkers(n int) Option

WithWorkers sets how many concurrent auto-consumer goroutines Run/Consume use (default 1). Values below 1 are treated as 1. Claim is atomic; each worker runs the handler independently — handlers must be concurrency-safe when n > 1.

type PQConfig 

type PQConfig struct {
	QueueName             string `json:"queue_name" yaml:"queue_name" env:"QUEUE_NAME"`
	BlockDuration         int    `json:"block_duration_sec,omitempty" yaml:"block_duration_sec,omitempty" env:"BLOCK_DURATION_SEC"`
	PublishWatermarkDelay int    `json:"publish_watermark_delay_sec,omitempty" yaml:"publish_watermark_delay_sec,omitempty" env:"PUBLISH_WATERMARK_DELAY_SEC"`
	CacheTimeout          int    `json:"cache_timeout_sec,omitempty" yaml:"cache_timeout_sec,omitempty" env:"CACHE_TIMEOUT_SEC"`
	PollInterval          int    `json:"poll_interval_sec,omitempty" yaml:"poll_interval_sec,omitempty" env:"POLL_INTERVAL_SEC"`
	RecoverInterval       int    `json:"recover_interval_sec,omitempty" yaml:"recover_interval_sec,omitempty" env:"RECOVER_INTERVAL_SEC"`
	RecoverMaxAge         int    `json:"recover_max_age_sec,omitempty" yaml:"recover_max_age_sec,omitempty" env:"RECOVER_MAX_AGE_SEC"`
	MaxDepth              int    `json:"max_depth,omitempty" yaml:"max_depth,omitempty" env:"MAX_DEPTH"`
	MaxInFlight           int    `json:"max_in_flight,omitempty" yaml:"max_in_flight,omitempty" env:"MAX_IN_FLIGHT"`
	Workers               int    `json:"workers,omitempty" yaml:"workers,omitempty" env:"WORKERS"`
}

PQConfig is the file/env-drivable queue configuration. Load it through the configuration component via WithConfigSource (preferred) or WithConfig. Durations are in seconds; zero means "unset" (keep option defaults).

type PriorityQueue 

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

PriorityQueue is the weighted priority-queue machine in caerus-framework-valkey-queues. Add increments the item weight; consumers pop the highest-weighted item. It holds a valkey peer pointer.

func New 

func New(opts ...Option) *PriorityQueue

New creates a priority queue component. It requires a valkey component and a queue name at Init.

func (*PriorityQueue) Ack 

func (q *PriorityQueue) Ack(ctx context.Context, id string) error

Ack completes a popped item: it removes the deadlock tracking, deletes the payload and clears any lingering queue/expiry member. Idempotent; safe to call for already-acked ids.

func (*PriorityQueue) Add 

func (q *PriorityQueue) Add(ctx context.Context, id, value string) (bool, error)

Add enqueues an item and returns whether its payload was newly stored. A false result means the id is already queued: the existing payload is kept and the item weight is incremented (this is the "add more weight" semantic; it does not overwrite). When a watermark delay is configured, a pub/sub notification is throttled to that interval; notification failures are logged but do not fail the add.

func (*PriorityQueue) BlockingBGet 

func (q *PriorityQueue) BlockingBGet(ctx context.Context) (*BGetObject, error)

BlockingBGet pops and returns the highest-weighted item, waiting up to the block duration. It returns (nil, nil) when the wait expires with no item. Pop and deadlock tracking are a single atomic Lua claim (no crash window). Waiting uses a wake list (BRPOP), not BZPOPMAX, so the wait cannot orphan an item.

func (*PriorityQueue) Consume 

func (q *PriorityQueue) Consume(ctx context.Context, handler Handler) error

Consume runs the auto-consumer until ctx is canceled: a pub/sub wakeup triggers an immediate pass, otherwise the queue is polled at the poll interval. Uses WithWorkers concurrent loops sharing one subscription. A failed handler is logged and the item requeued. Handlers must be safe for concurrent use when workers > 1.

func (*PriorityQueue) Count 

func (q *PriorityQueue) Count(ctx context.Context) (int64, error)

Count returns the number of queued items (distinct ids).

func (*PriorityQueue) Deadlocked 

func (q *PriorityQueue) Deadlocked(ctx context.Context) ([]InFlightItem, error)

Deadlocked returns the items currently in flight (popped but not acked or requeued), with the time each was popped. A consumer that crashed between a successful pop and its Ack/Requeue leaves its item listed here until RecoverDeadlocked returns it to the queue.

func (*PriorityQueue) GetDependencies 

func (q *PriorityQueue) GetDependencies() []string

GetDependencies implements cf.Dependencies. Depends on configuration when WithConfigSource is set.

func (*PriorityQueue) GetInitOrderStage 

func (q *PriorityQueue) GetInitOrderStage() cf.Stage

GetInitOrderStage implements cf.CaerusComponent.

func (*PriorityQueue) Health 

func (q *PriorityQueue) Health(ctx context.Context) error

Health implements cf.HealthProvider. It pings the backing valkey server and, when thresholds are non-zero, fails if queue depth or in-flight count exceed them (WithMaxDepth / WithMaxInFlight; defaults apply when WithHandler is set). A nil client is unhealthy.

func (*PriorityQueue) InFlightCount 

func (q *PriorityQueue) InFlightCount(ctx context.Context) (int64, error)

InFlightCount returns the number of popped but unacked items (deadlock set).

func (*PriorityQueue) Init 

func (q *PriorityQueue) Init(ctx context.Context, fw *cf.CaerusFramework) error

Init implements cf.CaerusComponent. It resolves the valkey client and validates the queue name, failing fast before the framework starts runners.

func (*PriorityQueue) IntCount 

func (q *PriorityQueue) IntCount(ctx context.Context) int64

IntCount returns the number of queued items, or 0 on error.

func (*PriorityQueue) Metrics 

func (q *PriorityQueue) Metrics() []cf_observability.Metric

Metrics implements cf_observability.MetricsProvider. While initialized it reports info, depth, in-flight, and cumulative recoveries; before Init or after Shutdown it returns nil (observability lazy pickup).

func (*PriorityQueue) Name 

func (q *PriorityQueue) Name() string

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

func (*PriorityQueue) OnConfigReload 

func (q *PriorityQueue) OnConfigReload(source string, cfg any)

OnConfigReload implements cf.ConfigReloader. It re-reads queue tunables from the bound configuration source. Queue name and key prefix are not changed. Credential rotation for the shared client is handled by the valkey component.

func (*PriorityQueue) PurgeExpired 

func (q *PriorityQueue) PurgeExpired(ctx context.Context) (int64, error)

PurgeExpired removes queued items whose CacheTimeout residence has elapsed, deleting the zqueue member and payload together. It is a no-op when CacheTimeout is 0. Safe to call concurrently; also run from recover ticks and before each claim.

func (*PriorityQueue) RecoverDeadlocked 

func (q *PriorityQueue) RecoverDeadlocked(ctx context.Context, maxAge time.Duration) (int64, error)

RecoverDeadlocked requeues every in-flight item that has been unacked for longer than maxAge, restoring it to the queue with weight +1. It returns the number of items recovered. Run invokes this on WithRecoverInterval (default 30s when a handler is set); it may also be called manually.

func (*PriorityQueue) RegisterConfigSources 

func (q *PriorityQueue) 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 and Owner) with the configuration component. No-op when no source is bound.

func (*PriorityQueue) Requeue 

func (q *PriorityQueue) Requeue(ctx context.Context, id string) error

Requeue returns a popped item to the queue with weight +1 and drops its deadlock tracking. Call it after a manual consumer failed to process the item; the payload is preserved. CacheTimeout residence is refreshed.

func (*PriorityQueue) Run 

func (q *PriorityQueue) Run(ctx context.Context) error

Run implements cf.Runnable. It runs the auto-consumer (when WithHandler is set) and/or the deadlock-recovery ticker until ctx is canceled.

func (*PriorityQueue) Shutdown 

func (q *PriorityQueue) Shutdown(ctx context.Context) error

Shutdown implements cf.CaerusComponent. It stops serving; the shared valkey client is owned and closed by the valkey component.

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.