Documentation
¶
Overview ¶
Package jobqueue provides worker pool implementations for asynchronous run execution. Jobs are submitted by the async run endpoint and executed by a fixed pool of goroutines. On completion an optional webhook callback is fired.
Two implementations are available:
- MemQueue — in-process bounded channel (default; jobs lost on restart)
- RedisQueue — Redis Streams backed (durable; jobs survive restarts)
Use New() for the in-memory variant and NewRedis() for the Redis variant. Both satisfy the Queue interface consumed by the rest of the codebase.
Webhook security: when a WEBHOOK_SECRET is configured every callback POST carries an X-VaultRun-Signature header of the form "sha256=<hex>", computed as HMAC-SHA256 of the raw request body. Receivers should verify this before processing the payload.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Job ¶
type Job struct {
// Req is the run request to execute.
Req runner.RunRequest
// Run is the pre-created pending run record (created by runner.PrepareAsync).
Run *models.Run
// CallbackURL, when non-empty, receives an HTTP POST with the run result.
CallbackURL string
}
Job is one unit of async work.
type Queue ¶
type Queue interface {
// Submit enqueues a job. Returns false (job dropped) when at capacity or
// when the queue is shutting down.
Submit(j Job) bool
// Len returns the approximate number of pending jobs.
Len() int
// Drain signals the queue to stop accepting new jobs and blocks until all
// in-flight workers have finished or ctx expires. Safe to call multiple times.
Drain(ctx context.Context)
}
Queue is the interface satisfied by both MemQueue and RedisQueue.
func New ¶
New creates an in-memory Queue with workers concurrent goroutines and a buffer of bufSize pending jobs. webhookSecret is used to HMAC-sign callback payloads (pass "" to disable signing).
func NewRedis ¶
func NewRedis( rnr *runner.Runner, db *sqlx.DB, redisAddr, redisPassword string, redisDB, workers int, webhookSecret string, ) (Queue, error)
NewRedis creates a Redis Streams-backed Queue.
- redisAddr — host:port of the Redis server (e.g. "localhost:6379")
- redisPassword — password (empty = no auth)
- redisDB — logical database index (0–15)
- workers — number of concurrent consumer goroutines
- webhookSecret — HMAC key for callback signing (empty = no signing)
Returns an error if the Redis connection cannot be established.
type RedisQueue ¶
type RedisQueue struct {
// contains filtered or unexported fields
}
RedisQueue is a durable, Redis Streams-backed Queue. Jobs survive process restarts because they are stored in Redis until explicitly acknowledged.
Each worker goroutine uses a named consumer ("worker-N") within the consumer group. Unacknowledged jobs from crashed workers can be reclaimed via the PEL reaper goroutine that runs XAUTOCLAIM periodically.
func (*RedisQueue) Drain ¶
func (q *RedisQueue) Drain(ctx context.Context)
Drain signals workers to stop after finishing in-flight jobs, then waits until all goroutines exit or ctx expires.
func (*RedisQueue) Len ¶
func (q *RedisQueue) Len() int
Len returns the total number of messages in the stream (not only pending). Used by the health endpoint for monitoring.
func (*RedisQueue) Submit ¶
func (q *RedisQueue) Submit(j Job) bool
Submit serialises the job to JSON and appends it to the Redis Stream. Returns false if the XADD fails (e.g. Redis unreachable).
Source Files