queue

package
v0.96.0 Latest Latest
Warning

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

 Go to latest Published: Nov 7, 2026 License: MIT Imports: 34 Imported by: 0

Documentation

Overview

Package queue is part of the GoFastr framework. See https://github.com/DonaldMurillo/gofastr for documentation.

Package queue provides a pluggable job queue with in-memory and Redis backends, a goroutine pool for concurrent processing, and scheduled job support.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrQueueClosed = errors.New("queue is closed")
	ErrNoJob       = errors.New("no job available")

	// ErrDuplicateJob is returned by Enqueue (and Replay) when a live job
	// of the same Type already holds the job's UniqueKey. Nothing was
	// enqueued; callers that only need the work to be queued can treat it
	// as success.
	ErrDuplicateJob = errors.New("queue: a job with this unique key is already queued")

	// ErrUniqueUnsupported is returned by a backend that cannot hold a
	// UniqueKey (RedisQueue). The job is not enqueued.
	ErrUniqueUnsupported = errors.New("queue: this backend does not support unique keys")
)

Sentinel errors.

View Source 
var ErrRedisEmpty = errors.New("redis: list is empty")

ErrRedisEmpty is the sentinel a RedisClient implementation returns (or wraps) when a read finds nothing: RPop on an empty list, HGet on a missing hash or field. It is how RedisQueue distinguishes a genuinely missing value from a real backend failure: "nothing there" is a normal signal (an empty poll → ErrNoJob; an Ack/Nack with nothing claimed → idempotent no-op), while any other error means the backend is unreachable and MUST be surfaced so an outage is not masked as idle.

Adapter authors translating a real driver MUST map its own nil-sentinel (e.g. go-redis's redis.Nil, redigo's redis.ErrNil) onto this via errors.Is.

Functions

func MetricsCollector added in v0.63.0 

func MetricsCollector(q Browsable, lane string) func(io.Writer)

MetricsCollector returns a Prometheus text-exposition collector that samples a Browsable queue's Stats at scrape time, emitting queue_depth (pending jobs, a gauge) and queue_dead_letter_total (dead jobs, a counter) labelled with the queue lane. Register the returned func on the app's *middleware.Metrics via RegisterCollector so the queue surfaces on the single /metrics endpoint. lane is the label value (the queue's name/lane).

The collector tolerates a Stats error by emitting nothing for that scrape (a transient DB error must never break /metrics).

Types

type Browsable 

type Browsable interface {
	// ListJobs returns up to limit jobs in the given status, skipping
	// the newest offset of them; pass an empty status to return all jobs
	// regardless of state. Jobs are ordered newest-first by created_at.
	// A negative offset reads as zero.
	ListJobs(ctx context.Context, status string, limit, offset int) ([]Job, error)
	// Stats returns counts grouped by status. Cheap by design, admin
	// dashboards may poll it.
	Stats(ctx context.Context) (JobStats, error)
}

Browsable is the optional read-only inspection interface, implemented by DBQueue, MemoryQueue and RedisQueue so admin tooling can list and aggregate jobs without guessing at the underlying schema. MemoryQueue and RedisQueue list dead-letter jobs only; admin code should type-assert the capability.

type CompareAndDeleter added in v0.78.0 

type CompareAndDeleter interface {
	HDelIfEqual(ctx context.Context, key, field, expect string) (bool, error)
}

CompareAndDeleter is the optional capability a RedisClient implements when it can delete a hash field atomically, and only while that field still holds an expected value. RedisQueue uses it to retire a processing entry: Ack/Nack read the entry, compare its ClaimToken in process, and then delete — and nothing binds the delete to the entry whose token was compared. One round trip of skew (a GC pause, a Redis failover) is enough for the lease to expire, Reclaim to run, and another worker to re-claim, at which point the unconditional delete removes the NEW claimant's entry: the job is then on no list, invisible to Reclaim, and silently lost.

Implement it with a Lua script (or WATCH/MULTI) — go-redis: 

if redis.call('HGET', KEYS[1], ARGV[1]) == ARGV[2] then
  return redis.call('HDEL', KEYS[1], ARGV[1]) end
return 0

It returns whether the field was deleted; false means the value had changed, which the queue treats as a fenced no-op.

A client that does not implement it still works: RedisQueue falls back to re-reading immediately before the delete, which narrows the window but cannot close it.

type CompareAndMover added in v0.96.0 

type CompareAndMover interface {
	MoveIfEqual(ctx context.Context, key, field, expect, list, value string) (bool, error)
}

CompareAndMover is the optional capability a RedisClient implements when it can move a claimed job in one atomic step: push value onto list and delete the hash field, only while that field still holds an expected value. RedisQueue uses it for the requeues Nack and Reclaim decide from a snapshot of the processing entry (a retry push, a dead-letter push, a lease-expiry re-delivery). Checking the claim and then pushing in separate round trips lets the claim change in between: the lease expires, another worker re-claims, and the stale push lands a second copy on the main list. A third worker then claims that copy, consumes the same attempt as the current claimant, and overwrites its processing entry. CompareAndDeleter cannot close this: the push has already happened by the time the conditional delete refuses.

Implement it with a Lua script — go-redis: 

if redis.call('HGET', KEYS[1], ARGV[1]) == ARGV[2] then
  redis.call('LPUSH', KEYS[2], ARGV[3])
  redis.call('HDEL', KEYS[1], ARGV[1])
  return 1 end
return 0

key and field name the processing entry, list the destination. On Redis Cluster a script may only touch keys in one slot, so give the queue a hash-tagged name ("{jobs}") so its lists and hash share one. It returns whether the move happened; false means the entry had changed, which the queue treats as a fenced no-op.

A client that does not implement it still works: RedisQueue pushes and then releases the claim with the conditional delete, which leaves the window above open.

type DBQueue 

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

DBQueue is a SQL-backed queue. Jobs persist in a single table; Dequeue claims a row atomically so multiple consumers can race safely.

Postgres uses FOR UPDATE SKIP LOCKED, the canonical pattern for queue fan-out without distributed locks. SQLite uses a SERIALIZABLE-friendly SELECT-then-UPDATE inside a tx; the table-level lock SQLite takes on BEGIN IMMEDIATE serialises writers naturally.

func NewDBQueue 

func NewDBQueue(db *sql.DB, opts ...DBQueueOption) (*DBQueue, error)

NewDBQueue constructs a DBQueue and ensures its backing table exists. Probes the dialect once via SELECT version(); falls back to SQLite. Panics if the table name contains unsafe characters.

func (*DBQueue) Ack 

func (q *DBQueue) Ack(ctx context.Context, job Job) error

Ack permanently removes the job, work is done, no replay needed. The DELETE is qualified with status='claimed' so it retires only a live claim: a row the dead-letter sweep has already adjudicated ('failed', lease expired on the final attempt) survives a late Ack from the crashed worker. The work may well have completed, but the sweep already logged the dead-letter at ERROR and Stats counted a failure, deleting the record afterwards would strand that signal with no way to reconcile it. The operator reconciles explicitly via Replay, which makes the row claimable again so a live claim's Ack deletes it through this same path. This mirrors RedisQueue, where a stale claim's Ack is a fenced no-op.

The DELETE is additionally fenced on claim_token, the identifier Dequeue mints per claim: a worker that overran its lease and had the row re-claimed under it matches no row, so its late Ack cannot retire the re-claimant's live work. Rows claimed before the column existed carry ”, which a caller presenting no token still matches.

Under WithDoneRetention the row is kept as status 'done' instead, under the same fence, and pruneDone deletes it once the retention passes.

func (*DBQueue) Close 

func (q *DBQueue) Close() error

Close stops worker goroutines started by Start. Idempotent.

func (*DBQueue) Dequeue 

func (q *DBQueue) Dequeue(ctx context.Context, types ...string) (Job, error)

Dequeue claims the highest-priority eligible job in a single atomic step. Returns ErrNoJob when nothing is ready (no pending row whose scheduled_at has passed). Claims from ANY lane, dedicated lane workers use the internal dequeue with a lane filter instead.

func (*DBQueue) Enqueue 

func (q *DBQueue) Enqueue(ctx context.Context, job Job) error

Enqueue inserts a job. Fills in ID/CreatedAt/MaxAttempts/ScheduledAt defaults when zero-valued so callers can pass {Type, Payload} only.

func (*DBQueue) ListJobs 

func (q *DBQueue) ListJobs(ctx context.Context, status string, limit, offset int) ([]Job, error)

ListJobs implements Browsable. Returns up to limit jobs in the supplied status, newest-first, after skipping offset of them. An empty or "all" status returns every job regardless of state. limit <= 0 defaults to 100; a negative offset reads as zero.

func (*DBQueue) Nack 

func (q *DBQueue) Nack(ctx context.Context, job Job) error

Nack returns a claimed job to the queue (status=pending) when it still has attempts left; on the final permitted attempt (attempts >= max_attempts) it dead-letters the row to status='failed' instead, where it stays visible to Stats, ListJobs("failed"), and Replay until replayed. When backoff is enabled (see WithBackoff), a requeued job's scheduled_at is pushed into the future so a flapping handler can't burn through every attempt in a tight loop. Every arm is fenced on claim_token. The lease-expiry clause in eligibleWhere deliberately re-claims a row whose worker is still running, so a stale claimant is a designed state on this backend, and an unfenced UPDATE by bare job ID let that stale worker flip the RE-CLAIMANT's live row to 'pending' (a third worker then runs the handler concurrently) or to 'failed' (dead-lettering a job someone is executing). Matching on the token Dequeue minted makes a stale Nack touch no row, mirroring RedisQueue's ownsClaim check. Rows predating the column carry ” and stay completable by a caller presenting no token. Every arm also requires status='claimed', as Ack and release do: the token survives the settle, so under WithDoneRetention a late Nack after Ack matched the retained 'done' row and re-queued completed work, and after the dead-letter sweep or a Replay it rewrote a row its claim no longer held.

func (*DBQueue) RegisterHandler 

func (q *DBQueue) RegisterHandler(jobType string, h Handler)

RegisterHandler binds a job type to a handler. Safe to call concurrently with a running worker loop.

func (*DBQueue) Replay 

func (q *DBQueue) Replay(ctx context.Context, jobID string) error

Replay implements Replayable: it resets a terminally-failed job to pending so a worker picks it up again, attempts cleared, scheduled immediately. The `AND status='failed'` clause makes it idempotent and safe: replaying an unknown, pending, running, or claimed job matches no row and is a no-op, so it can never double-run an in-flight job or resurrect a non-terminal one. A failed job whose UniqueKey a newer live job now holds stays failed, and Replay returns ErrDuplicateJob.

func (*DBQueue) SetGate added in v0.17.0 

func (q *DBQueue) SetGate(gate func(jobType string) bool)

SetGate installs a gate checked in the worker loop after a handler is resolved. When gate returns false the job is released back to pending (without consuming a retry attempt) and rescheduled slightly into the future to avoid a hot loop. Framework code uses it to defer jobs owned by a disabled module. Pass nil to clear.

func (*DBQueue) SetLeaseTimeout 

func (q *DBQueue) SetLeaseTimeout(d time.Duration)

SetLeaseTimeout adjusts the in-flight lease duration after construction. See WithLeaseTimeout for semantics; d <= 0 leaves the lease unchanged, and so does a lease no longer than the handler budget (WithDBHandlerTimeout), which is logged at WARN: NewDBQueue refuses that pair, and this setter has no error to return. Safe to call concurrently with a running worker loop.

func (*DBQueue) Start 

func (q *DBQueue) Start(ctx context.Context)

Start launches the shared worker pool (q.workers goroutines that claim any lane) plus one goroutine per dedicated lane worker added via WithLaneWorkers. Each loops dequeue → handle → Ack/Nack until Close. A worker goroutine that dies for any reason (including a panic escaping the handler-recover guard) is respawned so the pool can never be permanently drained by a poison message.

func (*DBQueue) Stats 

func (q *DBQueue) Stats(ctx context.Context) (JobStats, error)

Stats implements Browsable. Aggregates per-status counts over the whole table. Cheap: a single GROUP BY scan.

type DBQueueOption 

type DBQueueOption func(*DBQueue)

DBQueueOption configures DBQueue construction.

func WithBackoff 

func WithBackoff(base, max time.Duration) DBQueueOption

WithBackoff enables exponential retry backoff. On a Nack with retries remaining, scheduled_at is advanced by base*2^(attempts-1), so the first retry waits ~base, the second ~2*base, and so on, capped at max. A non-positive base disables backoff (jobs retry immediately, the default). A non-positive max means uncapped. The delay computation is core/backoff.Exponential, shared with the outbox relay.

func WithDBHandlerTimeout added in v0.11.0 

func WithDBHandlerTimeout(d time.Duration) DBQueueOption

WithDBHandlerTimeout caps a single handler invocation's wall-clock budget. The job's context is cancelled at the deadline and the worker stops waiting there: a black-holed dependency (an SMTP host that never responds, a hung HTTP call, a handler that ignores its context) can't wedge a worker forever, critical with the default single worker, where one stuck job stalls the entire queue — the job is nacked at the deadline and retried/dead-lettered, so a late side effect is the duplicate the at-least-once contract already allows. Zero (default) means no timeout; set it whenever handlers touch the network. The budget must be shorter than the claim lease (WithLeaseTimeout, five minutes by default): NewDBQueue refuses one that is not, since a handler still inside its budget would outlive its claim and have its job re-claimed and run again while it runs. (Named distinctly from the MemoryQueue's WithHandlerTimeout because both live in this package.)

func WithDBLaneWorkers added in v0.19.0 

func WithDBLaneWorkers(lane string, n int) DBQueueOption

WithDBLaneWorkers adds n dedicated worker goroutines that ONLY claim jobs whose Lane equals the given lane, on top of the shared workers from WithWorkers. This is the mechanism that prevents bulk backfills from starving urgent jobs: even when every shared worker is busy running a long-running bulk handler, the dedicated lane workers keep draining the reserved lane. Multiple calls for different lanes each add their own workers; multiple calls for the same lane sum. Priority ordering still selects among pending jobs within a worker's claim set. Panics if n <= 0 or lane is "" (use WithWorkers for the shared/default pool). Named with the DB prefix to match WithDBHandlerTimeout / WithDBLogger (both backends share this package, so the MemoryQueue variant is WithLaneWorkers).

func WithDBLogger added in v0.18.0 

func WithDBLogger(l *slog.Logger) DBQueueOption

WithDBLogger sets the logger used for handler-failure (WARN) and dead-letter (ERROR) records emitted from the worker loop. Defaults to slog.Default(); passing nil restores the default. Without it a failing handler dead-letters silently, you lose the job and the reason.

func WithDoneRetention added in v0.87.0 

func WithDoneRetention(d time.Duration) DBQueueOption

WithDoneRetention keeps an acked job as status "done" for d instead of deleting it, so an admin can list what ran; the claim loop deletes done jobs older than d, at most once a minute. Zero (the default) deletes on Ack, as before. A kept job still holds its ID, so a later Enqueue of the same ID fails until the retention has passed.

func WithLeaseTimeout 

func WithLeaseTimeout(d time.Duration) DBQueueOption

WithLeaseTimeout sets how long a claimed-but-unacked job may stay in-flight before it is considered abandoned (the worker crashed/was killed) and becomes eligible for re-dequeue. Defaults to 5 minutes; d <= 0 keeps the default. A zero lease is not "no lease": it made every claim abandoned the moment it was written, so the next poll ran a live job again and the dead-letter sweep failed a final attempt that was still running. With a handler budget set (WithDBHandlerTimeout) the lease must be longer than it; NewDBQueue refuses the pair otherwise.

func WithTable 

func WithTable(name string) DBQueueOption

WithTable overrides the default "queue_jobs" table name.

func WithWorkers 

func WithWorkers(n int) DBQueueOption

WithWorkers sets the number of background worker goroutines started by Start(). Defaults to 1 when not set. Zero starts no shared workers (only the WithDBLaneWorkers lanes run); a negative count keeps the default, since Start sized a channel with it and panicked.

type DurableScheduleBuilder added in v0.33.0 

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

DurableScheduleBuilder configures one persisted recurring schedule.

func (*DurableScheduleBuilder) Job added in v0.33.0 

func (b *DurableScheduleBuilder) Job(jobType string, payload any) *DurableScheduleBuilder

Job sets the queue job type and JSON payload for a durable schedule. Payload encoding errors are returned by Register or RegisterAt so the builder remains fluent.

func (*DurableScheduleBuilder) Lane added in v0.37.0 

func (b *DurableScheduleBuilder) Lane(name string) *DurableScheduleBuilder

Lane routes every Job fired by this schedule to the given capacity- reservation lane (see DBQueue.WithDBLaneWorkers). Empty (the default) lands the job on the shared/default lane. The value persists alongside the schedule definition; re-registering the same schedule ID updates it.

func (*DurableScheduleBuilder) MaxAttempts added in v0.37.0 

func (b *DurableScheduleBuilder) MaxAttempts(n int) *DurableScheduleBuilder

MaxAttempts bounds the retry ceiling carried into every Job fired by this schedule. Zero (the default) lets Enqueue resolve it to 3, matching the pre-options behaviour. The value persists alongside the schedule definition; re-registering the same schedule ID updates it.

func (*DurableScheduleBuilder) Priority added in v0.37.0 

func (b *DurableScheduleBuilder) Priority(p int) *DurableScheduleBuilder

Priority sets the priority carried into every Job fired by this schedule. Higher integers are dequeued first. Defaults to 0. The value persists alongside the schedule definition; re-registering the same schedule ID updates it.

func (*DurableScheduleBuilder) Register added in v0.33.0 

func (b *DurableScheduleBuilder) Register() error

Register persists the schedule definition without resetting an existing watermark. The anchor is UTC, so cron field evaluation keeps the historical "0 2 * * *" = 02:00 UTC semantics regardless of the server's local zone; register with RegisterAt and a time in a named IANA zone to evaluate a cron spec in that zone's wall-clock time.

func (*DurableScheduleBuilder) RegisterAt added in v0.33.0 

func (b *DurableScheduleBuilder) RegisterAt(base time.Time) error

RegisterAt is Register with a deterministic first-run anchor.

func (*DurableScheduleBuilder) UserID added in v0.96.0 

func (b *DurableScheduleBuilder) UserID(id string) *DurableScheduleBuilder

UserID names the person the schedule's payload is about. It persists with the schedule, and every Job the schedule fires carries it as Job.UserID. App.EraseUserData deletes the user's schedules and the jobs they fired. Leave it empty for a schedule that names nobody (a sweep, a cache warm). Re-registering the same schedule ID updates it.

type DurableScheduler added in v0.33.0 

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

DurableScheduler persists schedule watermarks and occurrences in the same SQL database as a DBQueue. A heartbeat-expiry lease avoids redundant evaluation, while unique occurrences and transactionally enqueued jobs are the deduplication authority.

func NewDurableScheduler added in v0.33.0 

func NewDurableScheduler(q *DBQueue, cfg DurableSchedulerConfig) (*DurableScheduler, error)

NewDurableScheduler creates a replica-safe scheduler backed by q's SQL database. It creates the schedule, occurrence, and lease tables when absent.

func (*DurableScheduler) Cron added in v0.33.0 

func (s *DurableScheduler) Cron(id, spec string) *DurableScheduleBuilder

Cron defines a durable cron schedule. id is the stable identity used to preserve its watermark across restarts and replica registration.

func (*DurableScheduler) Every added in v0.33.0 

func (s *DurableScheduler) Every(id string, interval time.Duration) *DurableScheduleBuilder

Every defines a durable fixed-interval schedule. id is the stable identity used to preserve its watermark across restarts and replica registration. Register refuses an interval below one millisecond.

func (*DurableScheduler) RunOnce added in v0.33.0 

func (s *DurableScheduler) RunOnce(ctx context.Context, now time.Time) error

RunOnce evaluates every due schedule at now. A replica that does not own the current fenced lease returns nil without side effects.

func (*DurableScheduler) Start added in v0.33.0 

func (s *DurableScheduler) Start(ctx context.Context) error

Start runs durable evaluation until ctx is cancelled. Registration wakes the loop immediately. The lease heartbeat is the maximum sleep; the earliest persisted next-run timestamp wakes it sooner.

An evaluation error no longer ends the loop: one poisoned schedule row or a transient DB blip returning from runOnce/nextWakeDelay is logged and the loop retries at the next heartbeat, so every healthy schedule on the replica keeps firing (the same skip-and-continue rule runOnce applies per row). Returning here permanently stopped all durable scheduling until process restart, silently once the boot log scrolled past.

type DurableSchedulerConfig added in v0.33.0 

type DurableSchedulerConfig struct {
	OwnerID               string
	LeaseDuration         time.Duration
	OccurrenceRetention   time.Duration
	MaxCatchUpOccurrences int
}

DurableSchedulerConfig identifies one scheduler replica and controls how quickly another replica may reclaim its leadership lease after a heartbeat stops. OwnerID defaults to a random process-local ID. LeaseDuration defaults to 30 seconds. OccurrenceRetention defaults to 30 days; a negative value disables occurrence pruning. MaxCatchUpOccurrences defaults to 1000 and bounds the history rows materialized after downtime.

type Handler 

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

Handler processes a job. Return a non-nil error to trigger a retry.

type Job 

type Job struct {
	ID           string          `json:"id"`
	OccurrenceID string          `json:"occurrence_id,omitempty"`
	Type         string          `json:"type"`
	Payload      json.RawMessage `json:"payload"`
	Priority     int             `json:"priority"`
	Lane         string          `json:"lane"`
	Attempts     int             `json:"attempts"`
	MaxAttempts  int             `json:"max_attempts"`
	CreatedAt    time.Time       `json:"created_at"`
	ScheduledAt  time.Time       `json:"scheduled_at"`

	// ClaimToken is minted by Dequeue per claim and identifies THAT claim,
	// not only the job. Backends that lease work to possibly-stale workers
	// (RedisQueue) verify it on Ack/Nack: a completion presented for a claim
	// that is no longer current, e.g. a worker whose visibility timeout
	// expired and whose job was re-claimed by another worker, is a fenced
	// no-op instead of mutating the current claimant's lease. Empty for jobs
	// that have never been claimed (and on backends without lease fencing).
	ClaimToken string `json:"claim_token,omitempty"`

	// UserID optionally names the person a job's payload is about. It is
	// what makes the job reachable by App.EraseUserData: queue_jobs lives
	// outside the entity registry, so the erase plane can only find a row
	// through a declared column, and a job carrying an address, an order,
	// or a document in its payload otherwise survives the user's erasure
	// and is re-disclosed by every later App.ExportData dump. Set it
	// whenever the payload is personal data; leave it empty for
	// infrastructure work (cache warms, sweeps, index rebuilds).
	//
	// DBQueue erases the job's row through its queue_jobs eraser;
	// RedisQueue and MemoryQueue remove the job through a store eraser,
	// registered until Close. A schedule's UserID option sets it on every
	// job the schedule fires.
	UserID string `json:"user_id,omitempty"`

	// UniqueKey, when set, makes the job unique among the live jobs of its
	// Type: while one job with this Type and key is pending, claimed, or
	// waiting on a retry, Enqueue refuses another with ErrDuplicateJob.
	// The key is free again once that job succeeds or is dead-lettered.
	// Use it for work where a second copy is wasted or harmful, such as a
	// per-user digest or a reindex of one record. RedisQueue refuses it
	// with ErrUniqueUnsupported.
	UniqueKey string `json:"unique_key,omitempty"`

	// Status, UpdatedAt and LastError are what a browsing backend
	// (DBQueue's ListJobs) reports about a stored job: its state
	// ("pending", "claimed", "failed", or "done" under
	// WithDoneRetention), when that state last changed, and the error
	// its last failed attempt returned (control bytes scrubbed, at most
	// 500 runes). Enqueue ignores them.
	Status    string    `json:"status,omitempty"`
	UpdatedAt time.Time `json:"updated_at,omitzero"`
	LastError string    `json:"last_error,omitempty"`
}

Job represents a unit of work enqueued for asynchronous processing.

type JobStats 

type JobStats map[string]int

JobStats is a snapshot of job counts grouped by status. The keys are status names (DBQueue: "pending", "claimed", "failed", and "done" under WithDoneRetention).

type MemoryQueue 

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

MemoryQueue is an in-memory queue backed by a goroutine pool. Pending jobs live in a priority heap (Priority DESC, enqueue-order ASC tiebreak) so higher-priority jobs are always taken first, priority is honoured, not ignored.

func NewMemoryQueue 

func NewMemoryQueue(workers int, opts ...MemoryQueueOption) *MemoryQueue

NewMemoryQueue creates a new in-memory queue with the given number of shared workers. Optional functional options (e.g. WithHandlerTimeout, WithLaneWorkers) may be passed. Shared workers take jobs from any lane by priority; dedicated lane workers (added via WithLaneWorkers) take only their own lane.

func (*MemoryQueue) Ack 

func (q *MemoryQueue) Ack(_ context.Context, job Job) error

Ack confirms a manually-dequeued job is done, discarding any tracked in-flight copy. For jobs processed by the automatic worker pool it is a no-op (those are auto-acknowledged after successful handler execution).

func (*MemoryQueue) Close 

func (q *MemoryQueue) Close() error

Close drains pending jobs and waits for all workers to finish. It signals the cond so every waiting worker wakes, processes any remaining claimable job, then exits when the store is empty. Shared workers drain every lane; lane workers drain only their own. No goroutine leaks, no panics.

Jobs whose ScheduledAt is still ahead are not run early and not waited for: an in-memory queue cannot keep them past Close, so they are dropped and their count is logged.

Close also removes the queue's store eraser, which NewMemoryQueue registered and which otherwise keeps the queue reachable.

func (*MemoryQueue) Dequeue 

func (q *MemoryQueue) Dequeue(ctx context.Context, types ...string) (Job, error)

Dequeue retrieves the highest-priority job from the pending store, optionally filtered by type. Useful for manual consumption without the automatic worker pool. It is non-blocking: returns ErrNoJob when nothing matches. Jobs of every lane are eligible (dedicated lane filtering is a worker-pool concept, not a manual-consumption one); a job whose ScheduledAt is still ahead is not.

func (*MemoryQueue) Enqueue 

func (q *MemoryQueue) Enqueue(ctx context.Context, job Job) error

Enqueue adds a job to the priority-ordered pending store. If the job has no ID, one is generated. The store is unbounded, so Enqueue never blocks on capacity (unlike the old buffered channel); it returns ErrQueueClosed only when the queue has been closed, and ErrDuplicateJob when a live job holds the job's UniqueKey.

func (*MemoryQueue) ListJobs 

func (q *MemoryQueue) ListJobs(_ context.Context, status string, limit, offset int) ([]Job, error)

ListJobs implements Browsable for the in-memory backend. The only state it can enumerate is the retained dead-letter set, so it returns those jobs for status "failed" (or an empty/"all" status), newest-first, and nothing for any other status, pending/claimed jobs live transiently on the pending heap. It skips the newest offset of them. limit <= 0 defaults to 100; a negative offset reads as zero.

func (*MemoryQueue) Nack 

func (q *MemoryQueue) Nack(ctx context.Context, job Job) error

Nack marks a manually-dequeued job as failed: it increments the attempt counter and re-enqueues the job when retries remain, otherwise drops it. The job must have been handed out by Dequeue (the in-flight set is consulted by ID). Jobs processed by the automatic worker pool retry internally and are never in the in-flight set; calling Nack for one is a harmless no-op.

The re-enqueue is lossless, the same write-to-next-home-before-drop rule Replay, RedisQueue.Nack, and processJob's own retry path follow: when Enqueue fails (typically because the CALLER's context is dead — a request disconnect fires r.Context(), which battery/admin passes through here), the job has already left the in-flight set, so it is retained in the dead-letter set (ListJobs("failed")/Replay can recover it) and the failure is logged before the error is returned. Without this, a cancelled request context destroyed the job outright: not pending, not dead, not in-flight, and a retried Nack is the documented no-op against a vanished job.

func (*MemoryQueue) RegisterHandler 

func (q *MemoryQueue) RegisterHandler(jobType string, handler Handler)

RegisterHandler registers a handler function for a given job type.

func (*MemoryQueue) Replay 

func (q *MemoryQueue) Replay(ctx context.Context, jobID string) error

Replay implements Replayable: it moves a retained terminally-failed job back onto the pending set with Attempts reset to 0 so Dequeue returns it again. Idempotent and safe: replaying an unknown or non-failed id matches no retained job and is a no-op (nil), never a double-enqueue.

func (*MemoryQueue) SetGate added in v0.17.0 

func (q *MemoryQueue) SetGate(gate func(jobType string) bool)

SetGate installs a gate checked in processJob after handlerFor. When gate returns false the job is re-enqueued with a short delay so it runs when the module re-enables. Framework code uses it to defer jobs owned by a disabled module. Pass nil to clear.

func (*MemoryQueue) Start 

func (q *MemoryQueue) Start()

Start launches the shared worker goroutines plus one goroutine per dedicated lane worker (WithLaneWorkers). Must be called before enqueuing jobs if you want automatic processing.

func (*MemoryQueue) Stats 

func (q *MemoryQueue) Stats(_ context.Context) (JobStats, error)

Stats implements Browsable. The in-memory backend can only count the retained dead-letter jobs (pending/claimed jobs are transient on the heap), so it reports those under "failed".

type MemoryQueueOption added in v0.3.1 

type MemoryQueueOption func(*MemoryQueue)

MemoryQueueOption is a functional option for NewMemoryQueue.

func WithHandlerTimeout added in v0.3.1 

func WithHandlerTimeout(d time.Duration) MemoryQueueOption

WithHandlerTimeout sets the per-job execution timeout for the automatic worker pool. The job's context is cancelled at the deadline and the worker stops waiting there: a handler that ignores its context cannot wedge its worker forever (one stuck job would otherwise stall the whole queue); the job is retried as a failure and a late side effect counts as the at-least-once duplicate. Defaults to 30 s; d <= 0 keeps the default. (A zero budget failed every job at once; DBQueue's WithDBHandlerTimeout reads zero as "no timeout" instead.)

func WithLaneWorkers added in v0.19.0 

func WithLaneWorkers(lane string, n int) MemoryQueueOption

WithLaneWorkers adds n dedicated worker goroutines that ONLY take jobs whose Lane equals the given lane, on top of the shared workers passed to NewMemoryQueue. This mirrors DBQueue.WithLaneWorkers and prevents a bulk backfill from starving urgent jobs even under worker saturation. Multiple calls for different lanes each add their own workers; multiple calls for the same lane sum. Panics if n <= 0 or lane is "" (use the shared worker count for the default lane).

func WithLogger added in v0.18.0 

func WithLogger(l *slog.Logger) MemoryQueueOption

WithLogger sets the logger used for handler-failure (WARN) and dead-letter (ERROR) records emitted from the worker pool. Defaults to slog.Default(); passing nil restores the default. Unprefixed to match the MemoryQueue's WithHandlerTimeout (the DBQueue uses WithDBLogger).

type PopRecorder added in v0.96.0 

type PopRecorder interface {
	RPopHSet(ctx context.Context, list, hash, field string) (string, error)
}

PopRecorder is the optional capability a RedisClient implements when it can pop the tail of a list and record the popped value in a hash in one atomic step. Dequeue takes a job off the main list and writes it to the processing hash in separate round trips, and a worker that dies between them (a deploy, an OOM kill) leaves the job on neither: Reclaim cannot see it, and it is lost. With PopRecorder the pop itself leaves a record of the job in the queue's pop hash ("<queue>:popped") under a field holding the pop time. Dequeue deletes the record once the job has its next home (the processing hash, the dead-letter list, or back on the main list), and Reclaim puts a record older than the visibility timeout back on the main list.

Implement it with a Lua script — go-redis: 

local v = redis.call('RPOP', KEYS[1])
if not v then return false end
redis.call('HSET', KEYS[2], ARGV[1], v)
return v

list is the list to pop, hash and field where the record goes. An empty list MUST be reported as ErrRedisEmpty, as for RPop (go-redis returns redis.Nil for the script's false). On Redis Cluster give the queue a hash-tagged name, as for CompareAndMover.

A client that does not implement it still works: Dequeue pops with RPop, and a worker that dies between the pop and the processing write loses the job.

type Queue 

type Queue interface {
	// Enqueue adds a job to the queue.
	Enqueue(ctx context.Context, job Job) error
	// Dequeue retrieves and removes the next available job, optionally filtered by type.
	Dequeue(ctx context.Context, types ...string) (Job, error)
	// Ack confirms successful processing of the claimed job. Backends with
	// lease fencing (RedisQueue) verify Job.ClaimToken and treat a stale
	// claim's Ack as a no-op.
	Ack(ctx context.Context, job Job) error
	// Nack marks a claimed job as failed and triggers retry logic. Backends
	// with lease fencing (RedisQueue) verify Job.ClaimToken and treat a
	// stale claim's Nack as a no-op.
	Nack(ctx context.Context, job Job) error
	// Close gracefully shuts down the queue, draining in-progress work.
	Close() error
}

Queue is the interface that every queue backend must implement.

type RedisClient 

type RedisClient interface {
	LPush(ctx context.Context, key string, values ...any) error
	RPop(ctx context.Context, key string) (string, error)
	// HGet returns the value stored at field in the hash at key. Like RPop,
	// a missing hash or field MUST be reported as ErrRedisEmpty (map the
	// driver's nil-sentinel, e.g. redis.Nil), the queue relies on it to
	// make Ack/Nack of a non-claimed job an idempotent no-op instead of a
	// hard error: a missing claim is normal (already completed, or the
	// lease expired and another worker took it) and must not be mistaken
	// for a backend outage.
	HGet(ctx context.Context, key, field string) (string, error)
	HSet(ctx context.Context, key string, values ...any) error
	// HGetAll returns every field→value pair in the hash. Used by Reclaim to
	// scan the processing hash for expired in-flight jobs.
	HGetAll(ctx context.Context, key string) (map[string]string, error)
	HDel(ctx context.Context, key string, fields ...string) error
	// Del removes one or more keys entirely. RedisQueue's own lease
	// protocol never calls it, per-job bookkeeping uses HDel, but it
	// stays in the adapter contract for callers that hold a RedisClient
	// and need whole-key cleanup.
	Del(ctx context.Context, keys ...string) error
	// LRange returns the elements of the list at key in the inclusive range
	// [start, stop]; negative indices count from the tail (-1 is the last
	// element). Used by Replay to read the dead-letter list.
	LRange(ctx context.Context, key string, start, stop int64) ([]string, error)
	// LRem removes up to count occurrences of value from the list at key and
	// returns the number removed. Used by Replay to pull one entry off the
	// dead-letter list.
	LRem(ctx context.Context, key string, count int64, value any) (int64, error)
}

RedisClient defines the minimal Redis operations needed by RedisQueue. This is an interface so callers can inject any Redis client (go-redis, redigo, etc.) without this package importing a specific driver.

type RedisQueue 

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

RedisQueue implements the Queue interface backed by Redis lists and hashes. It supports a visibility timeout for in-flight jobs and a dead letter queue for jobs that exceed MaxAttempts.

func NewRedisQueue 

func NewRedisQueue(client RedisClient, queueName string) *RedisQueue

NewRedisQueue creates a new Redis-backed queue.

func (*RedisQueue) Ack 

func (q *RedisQueue) Ack(ctx context.Context, job Job) error

Ack removes the job's processing entry after successful handling. The claim is fenced by Job.ClaimToken: an Ack from a worker whose visibility timeout already expired (the job was re-claimed elsewhere) is a no-op, deleting the newer claimant's entry would make the job unreclaimable if that claimant then crashed. Acking an ID nothing is claimed under (already acked) is an idempotent no-op.

func (*RedisQueue) Close 

func (q *RedisQueue) Close() error

Close removes the queue from App.EraseUserData's reach. The caller manages the Redis connection.

func (*RedisQueue) Dequeue 

func (q *RedisQueue) Dequeue(ctx context.Context, types ...string) (Job, error)

Dequeue pops a job from the Redis list and moves it to the processing queue. If types are specified, only jobs matching one of those types are returned; non-matching jobs are pushed back onto the list.

Crash-safety contract for the pop→processing transition: a job is popped off the main list before it is recorded in the processing hash. If that recording fails, the popped payload is pushed back onto the main list so the job is re-delivered later instead of being permanently lost in the gap. A worker that dies in the gap runs no such push; with PopRecorder the pop's record survives it and Reclaim re-delivers the job, and without it the job is lost.

Attempts are bumped at claim, matching DBQueue: a worker that crashes before Ack/Nack has still consumed a delivery, so a poison message cannot redeliver forever. A job whose attempts exceed MaxAttempts at claim is moved to the dead-letter queue instead of being handed out again.

Before popping, Dequeue moves every delayed job that has come due onto the main list (see promoteDue), so a job enqueued with a future ScheduledAt is invisible until then and never sits in front of a due one.

func (*RedisQueue) Enqueue 

func (q *RedisQueue) Enqueue(ctx context.Context, job Job) (err error)

Enqueue pushes a job onto the Redis list, applying defaults for ID, CreatedAt, and MaxAttempts when not set. A job whose ScheduledAt is ahead of the queue clock goes onto the delayed list instead, where Dequeue leaves it until it is due.

func (*RedisQueue) ListJobs added in v0.3.1 

func (q *RedisQueue) ListJobs(ctx context.Context, status string, limit, offset int) ([]Job, error)

ListJobs implements Browsable for the Redis backend. The only durable job state accessible without a scan of the full main/processing lists is the dead-letter queue, so this returns dead jobs for status "failed" (or an empty/"all" status) and nothing for any other status value. Jobs are returned newest-first (head of the Redis list) up to limit entries, after skipping offset of them. limit <= 0 defaults to 100; a negative offset reads as zero.

func (*RedisQueue) Nack 

func (q *RedisQueue) Nack(ctx context.Context, claimed Job) error

Nack handles a failed job. If retries remain, it re-enqueues the job; otherwise it moves it to the dead letter queue. Like Ack it is fenced by Job.ClaimToken: a stale claimant's Nack must neither re-enqueue the newer claimant's in-flight job (concurrent double-run) nor delete its entry (unreclaimable loss). Nacking an ID nothing is claimed under is an idempotent no-op.

func (*RedisQueue) Reclaim 

func (q *RedisQueue) Reclaim(ctx context.Context) (reclaimed int, err error)

Reclaim scans the processing set for in-flight jobs whose visibility timeout has passed (the worker that claimed them crashed before Ack/Nack), re-enqueues them onto the main list, and removes the stale processing entry. It first puts back the jobs a worker died holding inside Dequeue (see PopRecorder). Returns the number of jobs re-delivered. Call it periodically (e.g. from a background ticker) to make in-flight Redis work crash-safe.

func (*RedisQueue) Replay 

func (q *RedisQueue) Replay(ctx context.Context, jobID string) error

Replay implements Replayable: it pulls a terminally-failed job off the dead-letter list and re-enqueues it onto the main queue with its attempts counter reset, so it gets a full set of retries again. It is idempotent, replaying an unknown job ID is a no-op (returns nil), matching DBQueue.Replay.

The entry is LPush'd back onto the main queue first and only removed from the dead list on success, so a failure between the two ops leaves the job on the dead list (recoverable) rather than dropping it. A crash in that window can leave one copy on each list; the next Replay/Dequeue tolerates the duplicate.

func (*RedisQueue) SetVisibilityTimeout 

func (q *RedisQueue) SetVisibilityTimeout(d time.Duration)

SetVisibilityTimeout configures how long a job can be in-flight before it is considered abandoned and eligible for re-delivery. Defaults to 30 seconds; d <= 0 leaves the timeout unchanged, since a zero window made Reclaim re-deliver a job the moment its worker claimed it. Safe for concurrent use with Dequeue.

func (*RedisQueue) StaleClaimCount added in v0.66.0 

func (q *RedisQueue) StaleClaimCount() int64

StaleClaimCount returns how many Ack/Nack calls were rejected as coming from a stale claim (fenced out by Job.ClaimToken mismatch). Monitoring it distinguishes "handlers routinely exceed the visibility timeout" from silent double-delivery confusion.

func (*RedisQueue) Start added in v0.3.2 

func (q *RedisQueue) Start(ctx context.Context, interval time.Duration)

Start launches a background goroutine that calls Reclaim on every tick to re-enqueue in-flight jobs whose visibility timeout has expired (e.g. because the worker that claimed them crashed before Ack/Nack). It mirrors DBQueue's built-in lease-expiry reclaim so crashed-worker jobs are not silently stranded.

The goroutine exits when ctx is cancelled. interval controls how often Reclaim is called; 0 keeps the 30 second default. A negative interval panics — Start returns no error, and a sign/unit error must not silently become the slowest reclaim cadence. Typical use: 

q.Start(ctx, 30*time.Second)

func (*RedisQueue) Stats added in v0.3.1 

func (q *RedisQueue) Stats(ctx context.Context) (JobStats, error)

Stats implements Browsable for the Redis backend. It reports the count of dead-lettered jobs under the "failed" key; pending/in-flight jobs are not enumerable without a full scan and are omitted. Cheap: a single LRange(0, -1) length read.

type Replayable 

type Replayable interface {
	// Replay resets a terminally-failed job back to pending so it is picked up
	// again (attempts counter cleared, scheduled immediately). It MUST only
	// touch terminal ('failed') rows and be idempotent: replaying an unknown,
	// pending, or running job is a no-op, not an error or a double-run.
	Replay(ctx context.Context, jobID string) error
}

Replayable is the optional capability for re-queuing a dead-lettered job, implemented by DBQueue, MemoryQueue and RedisQueue. Each retains terminally failed jobs for inspection and replay. Admin tooling type-asserts for the capability before offering a replay action on a supplied Queue.

type ScheduleBuilder 

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

ScheduleBuilder provides a fluent API for building scheduled jobs.

func (*ScheduleBuilder) Job 

func (b *ScheduleBuilder) Job(jobType string, payload any) *ScheduleBuilder

Job sets the job type and payload for the scheduled job. A payload json.Marshal refuses is returned by Register or RegisterAt, so the builder remains fluent.

func (*ScheduleBuilder) Lane added in v0.37.0 

func (b *ScheduleBuilder) Lane(name string) *ScheduleBuilder

Lane routes every Job fired by this schedule to the given capacity- reservation lane (see DBQueue.WithDBLaneWorkers / MemoryQueue.WithLaneWorkers). Empty (the default) lands the job on the shared/default lane.

func (*ScheduleBuilder) MaxAttempts added in v0.37.0 

func (b *ScheduleBuilder) MaxAttempts(n int) *ScheduleBuilder

MaxAttempts bounds the retry ceiling carried into every Job fired by this schedule. Zero (the default) lets Enqueue resolve it to 3, matching the pre-options behaviour.

func (*ScheduleBuilder) Priority added in v0.37.0 

func (b *ScheduleBuilder) Priority(p int) *ScheduleBuilder

Priority sets the priority carried into every Job fired by this schedule. Higher integers are dequeued first. Defaults to 0.

func (*ScheduleBuilder) Register 

func (b *ScheduleBuilder) Register() error

Register adds the scheduled job to the scheduler, computing the first NextRun relative to the current wall clock. It returns an error when the payload given to Job cannot be encoded as JSON or a Cron schedule's spec is invalid, and registers nothing then.

func (*ScheduleBuilder) RegisterAt added in v0.3.3 

func (b *ScheduleBuilder) RegisterAt(base time.Time) error

RegisterAt is like Register but anchors the first NextRun to base instead of the wall clock. It exists so cron schedules can be registered deterministically (tests, replayed fixtures) without depending on time.Now().

func (*ScheduleBuilder) UserID added in v0.96.0 

func (b *ScheduleBuilder) UserID(id string) *ScheduleBuilder

UserID names the person the schedule's payload is about. Every Job it fires carries it as Job.UserID, which is what App.EraseUserData matches on in the queue the job lands in. The schedule itself lives in process memory, out of an erasure's reach; use a DurableScheduler for a schedule that must go with the user.

type ScheduledJob 

type ScheduledJob struct {
	Type     string          `json:"type"`
	Payload  json.RawMessage `json:"payload"`
	Interval time.Duration   `json:"interval"`
	NextRun  time.Time       `json:"next_run"`

	// Lane, Priority, and MaxAttempts are carried into every Job fired by
	// this schedule. They default to "", 0, and 0 (which enqueueWith
	// resolves to 3), matching the pre-options behaviour for callers that
	// never set them.
	Lane        string `json:"lane,omitempty"`
	Priority    int    `json:"priority"`
	MaxAttempts int    `json:"max_attempts"`

	// UserID is carried into every Job fired by this schedule (see
	// Job.UserID). Empty for a schedule that names nobody.
	UserID string `json:"user_id,omitempty"`
	// contains filtered or unexported fields
}

ScheduledJob defines a recurring job configuration.

A schedule fires on either a fixed Interval (set via Scheduler.Every) or a cron expression (set via Scheduler.Cron). Interval and cron are mutually exclusive: when cron is non-nil the Interval field is unused and NextRun is advanced by the cron expression instead.

type Scheduler 

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

Scheduler is the non-durable, single-process scheduler. Its watermarks live only in memory. Use DurableScheduler with DBQueue when multiple replicas or restart continuity must be safe.

func NewInMemoryScheduler added in v0.33.0 

func NewInMemoryScheduler(queues ...Queue) *Scheduler

NewInMemoryScheduler explicitly creates the non-durable, single-process scheduler. It is equivalent to NewScheduler.

func NewScheduler 

func NewScheduler(queues ...Queue) *Scheduler

NewScheduler creates the non-durable, single-process Scheduler that dispatches to the given queues. Enqueue errors are logged via slog.Default(). NewInMemoryScheduler is the explicit spelling for new code.

func NewSchedulerWithLogger added in v0.3.1 

func NewSchedulerWithLogger(q Queue, logger *slog.Logger) *Scheduler

NewSchedulerWithLogger creates a new Scheduler with an explicit logger. Pass a non-nil *slog.Logger to control where enqueue-error messages are routed; passing nil falls back to slog.Default().

func (*Scheduler) Close added in v0.96.0 

func (s *Scheduler) Close() error

Close removes the scheduler from App.EraseUserData's reach. It does not stop Start; cancel Start's context for that.

func (*Scheduler) Cron added in v0.3.3 

func (s *Scheduler) Cron(spec string) *ScheduleBuilder

Cron returns a ScheduleBuilder that fires when the cron expression's next time arrives. The spec accepts the standard 5-field syntax plus the @shortcuts (e.g. "0 2 * * *" for every day at 02:00, or "@daily"); it is parsed by framework/cron.Parse, so the queue does not carry a second cron parser. Spec errors surface from Register / RegisterAt, not here, so the fluent chain stays clean.

func (*Scheduler) Every 

func (s *Scheduler) Every(interval time.Duration) *ScheduleBuilder

Every returns a ScheduleBuilder that fires on a fixed interval. Register refuses a negative interval or a positive one below one millisecond, the DurableScheduler floor.

func (*Scheduler) Start 

func (s *Scheduler) Start(ctx context.Context)

Start begins the scheduling loop. It blocks until ctx is cancelled.

The loop re-reads the schedule set (under lock) on every tick, so jobs registered AFTER Start still fire, the natural wiring is "start subsystems, then register jobs", and an empty-at-Start scheduler that exited would silently drop everything registered later. The tick cadence adapts to the finest live interval each pass, so adding a sub-minute schedule after Start takes effect on the next tick.

type ScriptRunner added in v0.96.0 

type ScriptRunner interface {
	Eval(ctx context.Context, script string, keys []string, args ...any) (any, error)
}

ScriptRunner is the capability RedisQueue needs to erase a user's jobs (App.EraseUserData reaches the queue through its store eraser): run a Lua script with EVAL and return its reply. The queue owns the script; the adapter only passes it through. go-redis: 

func (c adapter) Eval(ctx context.Context, script string, keys []string, args ...any) (any, error) {
	return c.rdb.Eval(ctx, script, keys, args...).Result()
}

The erasure is one script because Redis runs a script with no other command in between. Read key by key, a worker's Dequeue or Nack can move a job into each key just after the sweep read it, so no number of passes proves the job gone; and a hash field read and then deleted in a second round trip can belong to another user's claim by then.

The script names all five of the queue's keys, so on Redis Cluster give the queue a hash-tagged name ("{jobs}"), as for CompareAndMover. An integer reply MUST come back as an integer type (go-redis returns int64).

With it, every move of a job from one of the queue's keys to another also runs as a script (moveScript) that writes only while the source still holds the entry: Dequeue's claim from its pop record into the processing hash, its dead-lettering and requeues, Nack and Reclaim without CompareAndMover, Replay, and the promotion of due delayed jobs. A worker that read a job before an erasure removed it would otherwise write it back; with the script it drops the job instead.

A client that does not implement it still runs the queue, but erasing a user's data from it fails with an error instead of sweeping key by key.

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.