failed

package
v0.14.2 Latest Latest
Warning

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

 Go to latest Published: Aug 24, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Overview

Package failed is where a job goes when it gives up.

It is a FailedJobProvider interface and the implementations of it, so a dead letter list can live somewhere other than the queue that produced it.

The default is that it does not move

A github.com/arandu-io/hesape/queue.DatabaseQueue marks the job failed in the jobs table it was already in, and queue.Queue.Failed lists them and queue.Queue.Retry puts one back. Every driver implements those two, so an application that never wires a provider still has a dead letter list, one store and one answer to "is this job still queued".

What this package is for is the case that arrangement cannot serve: a queue whose store is not durable enough to keep failures, or one that is flushed by something other than the application. A RESP queue is both. The provider is then a deliberate second place, wired on purpose, and the cost -- two stores, two retentions -- is paid knowingly.

Every read takes a Grant

A failed job carries a customer's payload, so listing them is a read like any other: every method takes an auth.Grant and filters by auth.Tenant(g). A provider that answered across tenants would leak the arguments of every job every customer ever queued.

DatabaseUUIDFailedJobProvider is an alias of DatabaseFailedJobProvider, because the id is the uuid and the two would run the same query.

Index

Constants

View Source 
const DefaultTable = "failed_jobs"

DefaultTable is where failures are logged when no table name is given.

Variables

View Source 
var ErrNoTenant = errors.New("queue/failed: the Grant carries no tenant, and a failed job list cannot be scoped without one")

ErrNoTenant is returned when a Grant carries no tenant.

It mirrors jobs.ErrNoTenant and exists for the same reason: a query with no tenant to filter by is a query that reads every customer's failures.

View Source 
var ErrNotFound = errors.New("queue/failed: no such failed job")

ErrNotFound is what Find wraps when there is no such failed job.

It is an error rather than a zero value, because "no such job" and "a job with no payload" are different answers and a command that retries the second one silently does nothing.

Functions

This section is empty.

Types

type CountableFailedJobProvider 

type CountableFailedJobProvider interface {
	// Count is how many failed jobs there are. An empty connection or queue
	// means every one.
	Count(ctx context.Context, g auth.Grant, connection, queue string) (int, error)
}

CountableFailedJobProvider is a provider that can say how many.

It is a second interface rather than a sixth method on FailedJobProvider: counting is what a monitor does every minute, and a provider that would have to load every row to answer should not pretend it can.

type CreateFailedJobsTable added in v0.5.0 

type CreateFailedJobsTable struct {
	migrations.BaseMigration

	// Table is the table to create. Empty means DefaultTable.
	Table string
}

CreateFailedJobsTable creates the table a DatabaseFailedJobProvider logs to.

The table name is a field because the provider's is: an application that keeps its failures somewhere other than failed_jobs migrates the name it wired.

func (CreateFailedJobsTable) Down added in v0.5.0 

func (m CreateFailedJobsTable) Down(ctx context.Context, conn migrations.Connection) error

Down drops the failed jobs table, and the index with it.

func (CreateFailedJobsTable) GetName added in v0.5.0 

func (CreateFailedJobsTable) GetName() string

GetName returns the migration's name.

func (CreateFailedJobsTable) Up added in v0.5.0 

func (m CreateFailedJobsTable) Up(ctx context.Context, conn migrations.Connection) error

Up creates the failed jobs table and the index every read of it uses.

Portable types only: TEXT, INTEGER and TIMESTAMP mean the same thing on SQLite, Postgres and MySQL.

type DatabaseFailedJobProvider 

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

DatabaseFailedJobProvider keeps the failed jobs in a table.

It is what an application wires when it wants the dead letter list to survive the queue -- a job that failed on a Redis queue that was then flushed is still a job somebody has to answer for.

The table is its own, and that is the point of the type: a queue whose store is not a database has nowhere else to keep them.

func NewDatabaseFailedJobProvider 

func NewDatabaseFailedJobProvider(db *database.DB, table string) *DatabaseFailedJobProvider

NewDatabaseFailedJobProvider returns the provider over db.

An empty table means DefaultTable.

func (*DatabaseFailedJobProvider) All 

func (p *DatabaseFailedJobProvider) All(ctx context.Context, g auth.Grant) ([]FailedJob, error)

All is this tenant's failed jobs, newest first.

func (*DatabaseFailedJobProvider) Count 

func (p *DatabaseFailedJobProvider) Count(ctx context.Context, g auth.Grant, connectionName, queue string) (int, error)

Count is how many of this tenant's jobs have failed.

func (*DatabaseFailedJobProvider) Find 

func (p *DatabaseFailedJobProvider) Find(ctx context.Context, g auth.Grant, id string) (FailedJob, error)

Find is one of this tenant's failed jobs.

func (*DatabaseFailedJobProvider) Flush 

func (p *DatabaseFailedJobProvider) Flush(ctx context.Context, g auth.Grant, age time.Duration) error

Flush removes this tenant's failed jobs older than age, or all of them when age is zero.

func (*DatabaseFailedJobProvider) Forget 

func (p *DatabaseFailedJobProvider) Forget(ctx context.Context, g auth.Grant, id string) (bool, error)

Forget removes one of this tenant's failed jobs.

func (*DatabaseFailedJobProvider) GetTable 

func (p *DatabaseFailedJobProvider) GetTable() string

GetTable is the name of the table this provider reads.

It returns the name rather than a query over it: a query handed out is a caller writing its own SQL against a table it does not own, and the tenant filter is not optional.

func (*DatabaseFailedJobProvider) IDs 

func (p *DatabaseFailedJobProvider) IDs(ctx context.Context, g auth.Grant, queue string) ([]string, error)

IDs is the identifiers of this tenant's failed jobs, newest first.

func (*DatabaseFailedJobProvider) Log 

func (p *DatabaseFailedJobProvider) Log(ctx context.Context, g auth.Grant, job FailedJob) (string, error)

Log records a job that gave up.

func (*DatabaseFailedJobProvider) Migrations 

func (p *DatabaseFailedJobProvider) Migrations() []migrations.Migration

Migrations returns the failed jobs table.

The schema is on the provider rather than on the queue module because it belongs to whoever wired this provider: an application that keeps its failures in the jobs table declares nothing here.

func (*DatabaseFailedJobProvider) Prune 

func (p *DatabaseFailedJobProvider) Prune(ctx context.Context, g auth.Grant, before time.Time) (int, error)

Prune removes this tenant's failed jobs that failed before an instant, and returns how many went.

type DatabaseUUIDFailedJobProvider 

type DatabaseUUIDFailedJobProvider = DatabaseFailedJobProvider

DatabaseUUIDFailedJobProvider is DatabaseFailedJobProvider under the name for the provider keyed by the job's uuid.

It is an alias because there is nothing left to distinguish: the id is the uuid (see database.NewID), so a provider that looked up by uuid would run the same query.

type FailedJob 

type FailedJob struct {
	// ID identifies this failure. It is the job's own UUID, because the id is
	// minted by the application and a second one would only be a second thing
	// to quote at somebody.
	ID string
	// UUID is the job's identifier, which is the same string as ID. It is kept
	// because it is what a provider indexes and what `aru queue:retry` takes.
	UUID string
	// TenantID is who the work belonged to, and it is not optional: a failed
	// job list that crossed customers would be one customer reading another's
	// payloads.
	TenantID string
	// Connection is the queue connection it was on.
	Connection string
	// Queue is the queue it was on.
	Queue string
	// Name is what routes the job to a handler.
	Name string
	// Payload is the job's arguments, as they were stored.
	Payload []byte
	// Exception is why it gave up.
	Exception string
	// FailedAt is when.
	FailedAt time.Time
}

FailedJob is one job that gave up.

It is the record a dead letter list holds: what the job was, whose it was, why it gave up and when.

type FailedJobProvider 

type FailedJobProvider interface {
	// Log records a job that gave up, and returns the id it was recorded under.
	Log(ctx context.Context, g auth.Grant, job FailedJob) (string, error)

	// IDs is the identifiers of the failed jobs, newest first. An empty queue
	// means every queue.
	IDs(ctx context.Context, g auth.Grant, queue string) ([]string, error)

	// All is the failed jobs, newest first.
	All(ctx context.Context, g auth.Grant) ([]FailedJob, error)

	// Find is one failed job, or an error wrapping [ErrNotFound].
	Find(ctx context.Context, g auth.Grant, id string) (FailedJob, error)

	// Forget removes one failed job, and reports whether there was one.
	Forget(ctx context.Context, g auth.Grant, id string) (bool, error)

	// Flush removes the failed jobs older than age. A zero age removes all of
	// them, which is what `queue:flush` with no --hours does.
	Flush(ctx context.Context, g auth.Grant, age time.Duration) error
}

FailedJobProvider is where a job goes when it gives up.

Every method takes a context and an auth.Grant.

The Grant is not decoration. A failed job carries a customer's payload, and "list the failed jobs" is a read like any other -- so it takes a Grant and every implementation filters by auth.Tenant(g). A provider that answered across tenants would be the one query in the collection that leaks, and it would leak the arguments of every job every customer ever queued.

type FileFailedJobProvider 

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

FileFailedJobProvider keeps the failed jobs in a JSON file.

It exists for the deployment that has a queue and no database: a single process on one machine, draining a RESP queue, that still wants a dead letter list to survive a restart.

It holds the newest Limit failures and drops the rest, which is the right behaviour for a file: a dead letter list that grows without bound on a disk nobody is watching is an outage waiting to be filed as a disk-full alert.

It is not for more than one process. The file is rewritten whole under a mutex this process holds, and two of them would lose each other's writes -- which is exactly the case DatabaseFailedJobProvider is for.

func NewFileFailedJobProvider 

func NewFileFailedJobProvider(path string, limit int) *FileFailedJobProvider

NewFileFailedJobProvider returns the provider over path.

A limit of zero or less means a hundred.

func (*FileFailedJobProvider) All 

func (p *FileFailedJobProvider) All(_ context.Context, g auth.Grant) ([]FailedJob, error)

All is this tenant's failed jobs, newest first. It answers all().

func (*FileFailedJobProvider) Count 

func (p *FileFailedJobProvider) Count(ctx context.Context, g auth.Grant, connectionName, queue string) (int, error)

Count is how many of this tenant's jobs have failed. It answers count().

func (*FileFailedJobProvider) Find 

func (p *FileFailedJobProvider) Find(ctx context.Context, g auth.Grant, id string) (FailedJob, error)

Find is one of this tenant's failed jobs. It answers find().

func (*FileFailedJobProvider) Flush 

func (p *FileFailedJobProvider) Flush(ctx context.Context, g auth.Grant, age time.Duration) error

Flush removes this tenant's failed jobs older than age, or all of them when age is zero. It answers flush().

func (*FileFailedJobProvider) Forget 

func (p *FileFailedJobProvider) Forget(_ context.Context, g auth.Grant, id string) (bool, error)

Forget removes one of this tenant's failed jobs. It answers forget().

func (*FileFailedJobProvider) IDs 

func (p *FileFailedJobProvider) IDs(ctx context.Context, g auth.Grant, queue string) ([]string, error)

IDs is the identifiers of this tenant's failed jobs, newest first. It answers ids().

func (*FileFailedJobProvider) Log 

func (p *FileFailedJobProvider) Log(ctx context.Context, g auth.Grant, job FailedJob) (string, error)

Log records a job that gave up. It answers log().

func (*FileFailedJobProvider) Prune 

func (p *FileFailedJobProvider) Prune(_ context.Context, g auth.Grant, before time.Time) (int, error)

Prune removes this tenant's failed jobs that failed before an instant, and returns how many went. It answers prune().

type NullFailedJobProvider 

type NullFailedJobProvider struct{}

NullFailedJobProvider accepts every failure and keeps none of them.

It is what an application wires when the dead letter list lives somewhere else entirely -- an error tracker, a log pipeline -- and keeping a second copy in the database would be a table nobody reads.

It is a value and not a pointer, for the reason NullQueue is: a nil pointer that silently swallows every failure is a worse mistake than the one this type exists to make cheap.

func (NullFailedJobProvider) All 

func (NullFailedJobProvider) All(context.Context, auth.Grant) ([]FailedJob, error)

All is empty.

func (NullFailedJobProvider) Count 

func (NullFailedJobProvider) Count(context.Context, auth.Grant, string, string) (int, error)

Count is zero.

func (NullFailedJobProvider) Find 

func (NullFailedJobProvider) Find(_ context.Context, _ auth.Grant, id string) (FailedJob, error)

Find never finds anything.

func (NullFailedJobProvider) Flush 

func (NullFailedJobProvider) Flush(context.Context, auth.Grant, time.Duration) error

Flush has nothing to flush.

func (NullFailedJobProvider) Forget 

func (NullFailedJobProvider) Forget(context.Context, auth.Grant, string) (bool, error)

Forget has nothing to forget, and reports true: the caller asked for the job to be gone, and it is.

func (NullFailedJobProvider) IDs 

func (NullFailedJobProvider) IDs(context.Context, auth.Grant, string) ([]string, error)

IDs is empty.

func (NullFailedJobProvider) Log 

func (NullFailedJobProvider) Log(_ context.Context, g auth.Grant, _ FailedJob) (string, error)

Log discards the failure and returns no id.

type PrunableFailedJobProvider 

type PrunableFailedJobProvider interface {
	// Prune removes the entries that failed before this instant, and returns
	// how many went.
	Prune(ctx context.Context, g auth.Grant, before time.Time) (int, error)
}

PrunableFailedJobProvider is a provider that can drop old entries.

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.