workhorse

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 3, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

README

github.com/stablemates/workhorse/go

The Go client, worker runtime, and dashboard handler for the Workhorse durable job queue for PostgreSQL.

Public beta: Workhorse is usable for evaluation and early production adoption. A 0.x minor release may change behaviour, so read the changelog before you upgrade. It will not ask you to recreate your database: migrations are ordered, and inside a major line a migration only adds, so a running deployment upgrades in place.

An AI agent should read the Workhorse documentation index first.

Install

go get github.com/stablemates/workhorse/go

Install the schema once, as a deployment step. The application never installs or migrates it.

npx --package @stablemates/workhorse@0.1.0 workhorse schema install

The machine that runs that deployment step needs Node.js 22 or newer. The application itself needs no Node.js.

Pin that version to the github.com/stablemates/workhorse/go version the application depends on. The two are released together from one commit, so the numbers match. A schema tool older than the application leaves a schema the application refuses to start against.

Runtime processes verify compatibility instead of changing the schema. Call AssertSchemaCompatible at startup.

Requires Go 1.25 or newer and PostgreSQL 15 through 18. pgx v5.9.2 is the minimum and the tested version.

Run one job

package main

import (
	"context"
	"fmt"
	"os"

	"github.com/jackc/pgx/v5/pgxpool"
	workhorse "github.com/stablemates/workhorse/go"
)

func main() {
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		panic(err)
	}
	defer pool.Close()

	queue := workhorse.NewQueue(workhorse.NewPGXExecutor(pool), "default")
	jobID, err := queue.Enqueue(ctx, "email.welcome", map[string]any{"to": "ada@example.com"})
	if err != nil {
		panic(err)
	}

	worker, err := workhorse.NewWorker(pool, workhorse.WorkerOptions{PollingOnly: true})
	if err != nil {
		panic(err)
	}
	worker.Handle("email.welcome", func(
		_ context.Context,
		payload any,
		_ *workhorse.HandlerContext,
	) (any, error) {
		return map[string]any{"deliveredTo": payload.(map[string]any)["to"]}, nil
	})
	if _, err := worker.RunOnce(ctx); err != nil {
		panic(err)
	}

	fmt.Println(jobID)
}

Handlers receive at-least-once delivery. Use stable provider idempotency keys around external effects; named checkpoints prevent completed application stages from running after a later restart.

Package boundary

This module provides transactional enqueue and worker APIs over caller-owned pgx or database/sql resources. Workers borrow a caller-owned pgx pool for claims and lifecycle calls. The module never installs or migrates the shared PostgreSQL schema.

Next

License

Apache-2.0. See LICENSE and NOTICE in the module.

Documentation

Overview

Package workhorse provides the public beta Go client and worker for the Workhorse durable job queue for PostgreSQL. During 0.x, a minor release may change behaviour, but the schema upgrades in place: migrations are ordered, and inside a major line a migration only adds.

Index

Constants

View Source 
const (
	MaxAdminPageSize     = 1000
	MaxAdminRequestBytes = 512
)
View Source 
const (
	// ProtocolVersion is the SQL protocol version implemented by this module.
	ProtocolVersion = 1

	// MaxEnqueueBatchSize is PostgreSQL's atomic enqueue batch limit.
	MaxEnqueueBatchSize = 1000
)
View Source 
const MaxChildJobs = 100

MaxChildJobs is PostgreSQL's linked-child limit for one parent.

View Source 
const MaxJobDependencies = 100

MaxJobDependencies is PostgreSQL's prerequisite fan-in limit for one job.

View Source 
const Version = "0.1.0"

Version is the published Workhorse module version. A parity test keeps it in step with the TypeScript and Python manifests, because the Go module has no manifest of its own to read.

Variables

View Source 
var ErrCancellationRequested = errors.New(cancellationRequestedMessage)

ErrCancellationRequested matches cooperative cancellation requested by an operator.

View Source 
var ErrDeadlineExceeded = errors.New(deadlineExceededMessage)

ErrDeadlineExceeded matches cancellation at a job's immutable deadline.

View Source 
var ErrDependencyCycle = errors.New(dependencyCycleMessage)

ErrDependencyCycle matches a dependency graph rejected because it would become cyclic.

View Source 
var ErrDependencyLimitExceeded = errors.New(dependencyLimitExceededMessage)

ErrDependencyLimitExceeded matches a dependency graph that exceeds a PostgreSQL limit.

View Source 
var ErrEnqueueBatchTooLarge = errors.New(enqueueBatchTooLargeMessage)

ErrEnqueueBatchTooLarge reports a batch that exceeds MaxEnqueueBatchSize.

View Source 
var ErrEnqueueIdempotencyConflict = errors.New(idempotencyConflictMessage)

ErrEnqueueIdempotencyConflict matches materially different requests under one retained key.

View Source 
var ErrExecutionTimeout = errors.New(executionTimeoutMessage)

ErrExecutionTimeout matches cancellation after an attempt consumes its execution budget.

View Source 
var ErrInvalidEnqueueOptions = errors.New(invalidEnqueueOptionsMessage)

ErrInvalidEnqueueOptions reports an option combination rejected before PostgreSQL is queried.

View Source 
var ErrInvalidEnqueueResult = errors.New(invalidEnqueueResultMessage)

ErrInvalidEnqueueResult reports a result set that violates the SQL protocol contract.

View Source 
var ErrInvalidPolicyOptions = errors.New(invalidPolicyOptionsMessage)

ErrInvalidPolicyOptions reports an unsupported synchronization option combination.

View Source 
var ErrInvalidPolicyResult = errors.New(invalidPolicyResultMessage)

ErrInvalidPolicyResult reports a policy row that violates the public result contract.

View Source 
var ErrInvalidScheduleDefinition = errors.New(invalidScheduleDefinitionMessage)

ErrInvalidScheduleDefinition reports a recurring definition rejected before PostgreSQL is queried.

View Source 
var ErrLeaseLost = errors.New(leaseLostMessage)

ErrLeaseLost matches a handler cancellation caused by PostgreSQL rejecting its fence.

View Source 
var ErrStaleLease = errors.New(staleLeaseMessage)

ErrStaleLease matches a lifecycle settlement rejected under an expired or superseded fence.

Functions

func AssertCompatible deprecated 

func AssertCompatible(ctx context.Context, executor Executor) error

AssertCompatible reads the installed schema and reports whether this client may mutate it.

Deprecated: renamed to AssertSchemaCompatible, which is the name the other two SDKs share.

func AssertSchemaCompatible 

func AssertSchemaCompatible(ctx context.Context, executor Executor) error

AssertSchemaCompatible reads the installed schema on every call. One statement returns both the schema version and the client protocols the schema declares it serves, so the check stays one round trip.

func CheckCompatibility 

func CheckCompatibility(installedSchemaVersion *int, clientProtocolVersion int, servedProtocolVersions []int) error

CheckCompatibility compares an installed schema and client protocol version.

The client declares a floor and no ceiling. Inside a major line a migration only adds, so a schema newer than the one this build was compiled against still carries every function it calls. The ceiling comes from servedProtocolVersions, which the installed schema declares in workhorse.protocol_version: a major release drops the client protocols it stops serving.

Types

type Admin 

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

Admin exposes operator reads and controls through a caller-owned executor.

func NewAdmin 

func NewAdmin(executor Executor) *Admin

NewAdmin constructs an operator client without taking ownership of the executor.

func (*Admin) GetCheckpoint 

func (admin *Admin) GetCheckpoint(ctx context.Context, jobID, name string) (*JobCheckpoint, error)

func (*Admin) GetJob 

func (admin *Admin) GetJob(ctx context.Context, id string) (*JobSnapshot, error)

func (*Admin) GetJobTimeline 

func (admin *Admin) GetJobTimeline(ctx context.Context, jobID string, query JobTimelineQuery) (JobTimelinePage, error)

func (*Admin) GetProgress 

func (admin *Admin) GetProgress(ctx context.Context, jobID string) (*JobProgress, error)

func (*Admin) GetWait 

func (admin *Admin) GetWait(ctx context.Context, jobID, name string) (*JobWait, error)

func (*Admin) Health 

func (admin *Admin) Health(ctx context.Context) (QueueHealth, error)

Health reads PostgreSQL's database-authoritative queue health snapshot.

func (*Admin) ListCheckpoints 

func (admin *Admin) ListCheckpoints(ctx context.Context, jobID string) ([]JobCheckpoint, error)

func (*Admin) ListDeadLetters 

func (admin *Admin) ListDeadLetters(ctx context.Context, query DeadLetterQuery) (DeadLetterPage, error)

func (*Admin) ListHumanWaits 

func (admin *Admin) ListHumanWaits(ctx context.Context, query ExternalWaitQuery) (ExternalWaitPage, error)

func (*Admin) ListJobs 

func (admin *Admin) ListJobs(ctx context.Context, query JobListQuery) (JobListPage, error)

func (*Admin) ListSignalWaits 

func (admin *Admin) ListSignalWaits(ctx context.Context, query ExternalWaitQuery) (ExternalWaitPage, error)

func (*Admin) ListWaits 

func (admin *Admin) ListWaits(ctx context.Context, jobID string) ([]JobWait, error)

func (*Admin) ListWorkers 

func (admin *Admin) ListWorkers(ctx context.Context) ([]WorkerRegistryEntry, error)

func (*Admin) PauseQueue 

func (admin *Admin) PauseQueue(ctx context.Context, queue string, audit AdminAudit) error

func (*Admin) PurgeQueue 

func (admin *Admin) PurgeQueue(ctx context.Context, queue string, audit AdminAudit) (int, error)

func (*Admin) Redrive 

func (admin *Admin) Redrive(ctx context.Context, sourceJobID string, audit AdminAudit) (RedriveResult, error)

func (*Admin) RedriveMany 

func (admin *Admin) RedriveMany(ctx context.Context, filter DeadLetterFilter, audit AdminAudit, options BulkRedriveOptions) (BulkRedrivePage, error)

func (*Admin) ResumeQueue 

func (admin *Admin) ResumeQueue(ctx context.Context, queue string, audit AdminAudit) error

func (*Admin) SetWorkerPaused 

func (admin *Admin) SetWorkerPaused(ctx context.Context, workerID string, paused bool, audit AdminAudit) (*WorkerPauseResult, error)

type AdminAudit 

type AdminAudit struct {
	Actor     string
	Reason    string
	RequestID string
}

AdminAudit attributes an administrative mutation and makes retries idempotent. PostgreSQL records this identity; it does not authorize it.

type AdminCursor 

type AdminCursor struct {
	OccurredAt time.Time
	JobID      string
	Kind       string
	RecordID   string
	Signature  string
}

type BatchFailed 

type BatchFailed struct{ Error error }

BatchFailed submits Error through one member's persisted retry policy.

type BatchHandler 

type BatchHandler func([]BatchHandlerItem) []BatchHandlerOutcome

BatchHandler processes one ordered group and returns one positional outcome per member.

type BatchHandlerContext 

type BatchHandlerContext struct {
	Job     ClaimedJob
	Context context.Context
	// contains filtered or unexported fields
}

BatchHandlerContext exposes one member's fence and checkpoint operations without suspension APIs.

func (*BatchHandlerContext) Checkpoint 

func (handler *BatchHandlerContext) Checkpoint(
	name string,
	operation func() (any, error),
) (any, error)

Checkpoint returns a member's stored value, or runs and saves operation once.

func (*BatchHandlerContext) GetProgress 

func (handler *BatchHandlerContext) GetProgress() (*JobProgress, error)

GetProgress returns the member's latest progress projection.

func (*BatchHandlerContext) SetProgress 

func (handler *BatchHandlerContext) SetProgress(value any) (*JobProgress, error)

SetProgress replaces the member's latest progress under its fenced lease.

type BatchHandlerItem 

type BatchHandlerItem struct {
	Payload any
	Context *BatchHandlerContext
}

BatchHandlerItem is one claimed payload and its independent fenced context.

type BatchHandlerOptions 

type BatchHandlerOptions struct {
	MaxSize int
	Linger  time.Duration
}

BatchHandlerOptions bounds one process-local batch rendezvous.

type BatchHandlerOutcome 

type BatchHandlerOutcome interface {
	// contains filtered or unexported methods
}

BatchHandlerOutcome is one member's explicit success or failure.

type BatchSucceeded 

type BatchSucceeded struct{ Result any }

BatchSucceeded completes one member with Result.

type BulkRedriveOptions 

type BulkRedriveOptions struct {
	Limit  int
	DryRun bool
	Cursor *AdminCursor
}

type BulkRedrivePage 

type BulkRedrivePage struct {
	Results    []RedriveResult
	NextCursor *AdminCursor
}

type CachedCompatibilityCheck 

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

CachedCompatibilityCheck runs one compatibility query and reuses its result.

func NewCachedCompatibilityCheck 

func NewCachedCompatibilityCheck(executor Executor) *CachedCompatibilityCheck

NewCachedCompatibilityCheck builds the opt-in one-shot gate for hot worker loops.

func (*CachedCompatibilityCheck) Assert 

func (check *CachedCompatibilityCheck) Assert(ctx context.Context) error

Assert returns the result of the first compatibility query for every call.

type CancelResult 

type CancelResult struct {
	Status         CancelStatus
	JobID          string
	State          *JobState
	CurrentAttempt *int
	RequestedAt    *time.Time
	RequestedBy    *string
	Reason         *string
	FinishedAt     *time.Time
}

CancelResult contains safe lifecycle metadata and omits payload and worker ownership.

type CancelStatus 

type CancelStatus string

CancelStatus is PostgreSQL's disposition for a cancellation request. 

const (
	CancelCanceled        CancelStatus = canceledValue
	CancelRequested       CancelStatus = cancelRequestedValue
	CancelAlreadyTerminal CancelStatus = alreadyTerminalValue
	CancelNotFound        CancelStatus = notFoundValue
)

type CancellationRequest 

type CancellationRequest struct {
	RequestedBy *string
	Reason      *string
}

CancellationRequest supplies optional audit attribution. PostgreSQL does not treat it as authorization.

type CancellationRequestedError 

type CancellationRequestedError struct{ JobID string }

CancellationRequestedError identifies an operator cancellation delivered to a handler.

func (*CancellationRequestedError) Error 

func (err *CancellationRequestedError) Error() string

func (*CancellationRequestedError) Unwrap 

func (err *CancellationRequestedError) Unwrap() error

type CheckpointConflictError 

type CheckpointConflictError struct {
	JobID          string
	CheckpointName string
}

CheckpointConflictError identifies an immutable checkpoint name reused with another value.

func (*CheckpointConflictError) Error 

func (err *CheckpointConflictError) Error() string

type CheckpointLeaseLostError 

type CheckpointLeaseLostError struct {
	JobID          string
	CheckpointName string
}

CheckpointLeaseLostError identifies a checkpoint rejected under a stale fence.

func (*CheckpointLeaseLostError) Error 

func (err *CheckpointLeaseLostError) Error() string

func (*CheckpointLeaseLostError) Unwrap 

func (err *CheckpointLeaseLostError) Unwrap() error

type ChildCanceled 

type ChildCanceled struct {
	Error any `json:"error"`
}

func (ChildCanceled) MarshalJSON 

func (outcome ChildCanceled) MarshalJSON() ([]byte, error)

func (ChildCanceled) OutcomeStatus 

func (ChildCanceled) OutcomeStatus() string

type ChildConflictError 

type ChildConflictError struct {
	ParentJobID string
	ChildName   string
}

ChildConflictError identifies a retained child name or set replayed with another request.

func (*ChildConflictError) Error 

func (err *ChildConflictError) Error() string

type ChildFailed 

type ChildFailed struct {
	Error any `json:"error"`
}

func (ChildFailed) MarshalJSON 

func (outcome ChildFailed) MarshalJSON() ([]byte, error)

func (ChildFailed) OutcomeStatus 

func (ChildFailed) OutcomeStatus() string

type ChildJobRequest 

type ChildJobRequest struct {
	Name    string
	Type    string
	Payload any
	Options EnqueueOptions
}

ChildJobRequest describes one named child created by an active parent handler.

type ChildLeaseLostError 

type ChildLeaseLostError struct{ ParentJobID string }

ChildLeaseLostError identifies child creation rejected under a stale parent fence.

func (*ChildLeaseLostError) Error 

func (err *ChildLeaseLostError) Error() string

func (*ChildLeaseLostError) Unwrap 

func (err *ChildLeaseLostError) Unwrap() error

type ChildLimitExceededError 

type ChildLimitExceededError struct{ ParentJobID string }

ChildLimitExceededError identifies a parent that exceeds MaxChildJobs.

func (*ChildLimitExceededError) Error 

func (err *ChildLimitExceededError) Error() string

type ChildOutcome 

type ChildOutcome interface {
	OutcomeStatus() string
	// contains filtered or unexported methods
}

ChildOutcome is one tagged terminal outcome returned by RunChildren.

type ChildResult 

type ChildResult struct {
	Name    string       `json:"name"`
	Outcome ChildOutcome `json:"outcome"`
}

ChildResult retains one settled outcome beside its stable child name.

type ChildResultLimitExceededError 

type ChildResultLimitExceededError struct {
	ParentJobID      string
	ResultBytes      int
	ResultLimitBytes int
}

ChildResultLimitExceededError identifies joined results larger than the parent's contract.

func (*ChildResultLimitExceededError) Error 

func (err *ChildResultLimitExceededError) Error() string

type ChildSucceeded 

type ChildSucceeded struct {
	Result any `json:"result"`
}

func (ChildSucceeded) MarshalJSON 

func (outcome ChildSucceeded) MarshalJSON() ([]byte, error)

func (ChildSucceeded) OutcomeStatus 

func (ChildSucceeded) OutcomeStatus() string

type ChildSuccessResult 

type ChildSuccessResult struct {
	Name   string `json:"name"`
	Result any    `json:"result"`
}

ChildSuccessResult is one successful result from RunChildrenAll.

type ClaimedJob 

type ClaimedJob struct {
	ID                 string
	Queue              string
	Type               string
	Priority           int
	Payload            any
	ContractVersion    *string
	ResultMaxBytes     int
	RedactErrorDetails bool
	TraceContext       any
	Attempt            int
	MaxAttempts        int
	RetryPolicy        map[string]any
	Deadline           *time.Time
	ExecutionTimeout   time.Duration
	AttemptTimeout     *time.Time
	FenceToken         int64
	LeaseExpiresAt     time.Time
}

ClaimedJob is PostgreSQL's immutable snapshot of one fenced attempt.

type CompatibilityCode 

type CompatibilityCode string

CompatibilityCode identifies why a client must refuse a mutation. 

const (
	SchemaNotInstalled   CompatibilityCode = "schema-not-installed"
	SchemaTooOld         CompatibilityCode = "schema-too-old"
	SchemaTooNew         CompatibilityCode = "schema-too-new"
	ClientProtocolTooOld CompatibilityCode = "client-protocol-too-old"
	ClientProtocolTooNew CompatibilityCode = "client-protocol-too-new"
)

type CompatibilityError 

type CompatibilityError struct {
	Code CompatibilityCode
}

CompatibilityError refuses a mutation against an incompatible SQL protocol.

func (*CompatibilityError) Error 

func (err *CompatibilityError) Error() string

func (*CompatibilityError) Is 

func (err *CompatibilityError) Is(target error) bool

Is matches compatibility errors by refusal code.

type ConcurrencyPolicy 

type ConcurrencyPolicy struct {
	Namespace       string
	Queue           string
	MaxActive       int
	MaxActivePerKey *int
	UpdatedAt       time.Time
}

ConcurrencyPolicy is one persisted queue concurrency policy.

type ConcurrencyPolicyDefinition 

type ConcurrencyPolicyDefinition struct {
	Queue           string `json:"queue"`
	MaxActive       int    `json:"maxActive"`
	MaxActivePerKey *int   `json:"maxActivePerKey,omitempty"`
}

ConcurrencyPolicyDefinition is one queue's deployment-synchronized active-job budget.

type DeadLetter 

type DeadLetter struct {
	JobID              string
	Queue              string
	Type               string
	ConcurrencyKey     *string
	Priority           int
	Payload            any
	Tags               []string
	CurrentAttempt     int
	MaxAttempts        int
	RetryPolicy        map[string]any
	DeadlineAt         *time.Time
	ExecutionTimeoutMS *int64
	Error              any
	FinishedAt         time.Time
	RedriveCount       int64
}

type DeadLetterFilter 

type DeadLetterFilter struct {
	Queue          string
	Type           string
	Tags           []string
	ErrorName      string
	FinishedAfter  *time.Time
	FinishedBefore *time.Time
}

type DeadLetterPage 

type DeadLetterPage struct {
	Items      []DeadLetter
	NextCursor *AdminCursor
}

type DeadLetterQuery 

type DeadLetterQuery struct {
	DeadLetterFilter
	Limit  int
	Cursor *AdminCursor
}

type DeadlineExceededError 

type DeadlineExceededError struct{ JobID string }

DeadlineExceededError identifies a job whose immutable deadline cancelled its handler.

func (*DeadlineExceededError) Error 

func (err *DeadlineExceededError) Error() string

func (*DeadlineExceededError) Unwrap 

func (err *DeadlineExceededError) Unwrap() error

type Debounce 

type Debounce struct {
	Key      string           `json:"key"`
	Scope    string           `json:"scope"`
	WindowMS int              `json:"windowMs"`
	Schedule DebounceSchedule `json:"schedule"`
}

Debounce replaces a pending keyed job during a PostgreSQL-owned window.

type DebounceSchedule 

type DebounceSchedule string

DebounceSchedule controls whether a replacement resets or preserves the original window. 

const (
	DebounceReset    DebounceSchedule = debounceResetValue
	DebouncePreserve DebounceSchedule = debouncePreserveValue
)

type Dependencies 

type Dependencies struct {
	PrerequisiteJobIDs []string                 `json:"prerequisiteJobIds"`
	OnSuccess          DependencyTerminalPolicy `json:"onSuccess"`
	OnFailure          DependencyTerminalPolicy `json:"onFailure"`
	OnCancellation     DependencyTerminalPolicy `json:"onCancellation"`
}

Dependencies declares prerequisite jobs and the terminal outcomes accepted from each.

type DependencyCycleDetails 

type DependencyCycleDetails struct {
	DependentJobID    string   `json:"dependentJobId"`
	PrerequisiteJobID string   `json:"prerequisiteJobId"`
	CycleJobIDs       []string `json:"cycleJobIds"`
	Truncated         bool     `json:"truncated"`
}

DependencyCycleDetails is PostgreSQL's bounded description of a rejected cycle.

type DependencyCycleError 

type DependencyCycleError struct {
	Details DependencyCycleDetails
	// contains filtered or unexported fields
}

DependencyCycleError contains PostgreSQL's structured cycle details.

func (DependencyCycleError) Error 

func (err DependencyCycleError) Error() string

func (DependencyCycleError) Unwrap 

func (err DependencyCycleError) Unwrap() error

type DependencyLimit 

type DependencyLimit string

DependencyLimit identifies the bounded graph dimension PostgreSQL rejected. 

const (
	DependencyPrerequisites        DependencyLimit = dependencyLimitPrerequisites
	DependencyDependents           DependencyLimit = dependencyLimitDependents
	DependencyUnresolvedDependents DependencyLimit = dependencyLimitUnresolved
)

type DependencyLimitDetails 

type DependencyLimitDetails struct {
	JobID string          `json:"jobId"`
	Limit DependencyLimit `json:"limit"`
	Max   int             `json:"max"`
}

DependencyLimitDetails is PostgreSQL's dependency limit diagnosis.

type DependencyLimitExceededError 

type DependencyLimitExceededError struct {
	Details DependencyLimitDetails
	// contains filtered or unexported fields
}

DependencyLimitExceededError contains PostgreSQL's structured limit details.

func (DependencyLimitExceededError) Error 

func (err DependencyLimitExceededError) Error() string

func (DependencyLimitExceededError) Unwrap 

func (err DependencyLimitExceededError) Unwrap() error

type DependencyTerminalPolicy 

type DependencyTerminalPolicy string

DependencyTerminalPolicy controls what a dependent does after a prerequisite settles. 

const (
	DependencyRelease DependencyTerminalPolicy = dependencyReleaseValue
	DependencyCancel  DependencyTerminalPolicy = dependencyCancelValue
	DependencyFail    DependencyTerminalPolicy = dependencyFailValue
)

type EnqueueIdempotencyConflictDetails 

type EnqueueIdempotencyConflictDetails struct {
	Scope                 string   `json:"scope"`
	KeyPreview            string   `json:"keyPreview"`
	KeyDigest             string   `json:"keyDigest"`
	KeyLength             int      `json:"keyLength"`
	ExistingJobID         string   `json:"existingJobId"`
	Ordinal               int      `json:"ordinal"`
	ConflictingFields     []string `json:"conflictingFields"`
	StoredRequestDigest   string   `json:"storedRequestDigest"`
	RejectedRequestDigest string   `json:"rejectedRequestDigest"`
}

EnqueueIdempotencyConflictDetails is PostgreSQL's retained-key conflict diagnosis.

type EnqueueIdempotencyConflictError 

type EnqueueIdempotencyConflictError struct {
	Details EnqueueIdempotencyConflictDetails
	// contains filtered or unexported fields
}

EnqueueIdempotencyConflictError contains PostgreSQL's structured conflict details.

func (EnqueueIdempotencyConflictError) Error 

func (err EnqueueIdempotencyConflictError) Error() string

func (EnqueueIdempotencyConflictError) Unwrap 

func (err EnqueueIdempotencyConflictError) Unwrap() error

type EnqueueNonReplaceableReason 

type EnqueueNonReplaceableReason string

EnqueueNonReplaceableReason explains why PostgreSQL retained a debounced job. 

const (
	IncompatibleKeyMode  EnqueueNonReplaceableReason = reasonIncompatibleKeyModeValue
	NotPending           EnqueueNonReplaceableReason = reasonNotPendingValue
	WindowElapsedPending EnqueueNonReplaceableReason = reasonWindowElapsedValue
)

IncompatibleKeyMode, NotPending, and WindowElapsedPending name the three reasons PostgreSQL retains a debounced job. They carry the same values as the NonReplaceable-prefixed constants that replace them, so a comparison against either spelling behaves the same way.

Deprecated: every other constant group in this package carries its enum prefix. Use the NonReplaceable-prefixed spelling of each reason. 

const (
	NonReplaceableIncompatibleKeyMode EnqueueNonReplaceableReason = reasonIncompatibleKeyModeValue
	NonReplaceableNotPending          EnqueueNonReplaceableReason = reasonNotPendingValue
	NonReplaceableWindowElapsed       EnqueueNonReplaceableReason = reasonWindowElapsedValue
)

type EnqueueOptions 

type EnqueueOptions struct {
	Queue              string
	Priority           int
	ConcurrencyKey     string
	RunAt              *time.Time
	Deadline           *time.Time
	ExecutionTimeoutMS int
	MaxAttempts        int
	RetryPolicy        map[string]any
	Tags               []string
	Idempotency        *Idempotency
	Debounce           *Debounce
	Throttle           *Throttle
	Dependencies       *Dependencies
}

EnqueueOptions controls a job's initial dispatch and durable acceptance behavior. Zero values select PostgreSQL-compatible client defaults or omit optional values.

type EnqueueOutcome 

type EnqueueOutcome string

EnqueueOutcome is PostgreSQL's durable disposition for one request. 

const (
	EnqueueAccepted       EnqueueOutcome = enqueueAcceptedValue
	EnqueueReplayed       EnqueueOutcome = enqueueReplayedValue
	EnqueueReplaced       EnqueueOutcome = enqueueReplacedValue
	EnqueueNonReplaceable EnqueueOutcome = enqueueNonReplaceableValue
	EnqueueCoalesced      EnqueueOutcome = enqueueCoalescedValue
)

type EnqueueRequest 

type EnqueueRequest struct {
	Type    string
	Payload any
	Options EnqueueOptions
}

EnqueueRequest is one job submitted through an atomic enqueue batch.

type EnqueueResult 

type EnqueueResult struct {
	JobID   string
	Outcome EnqueueOutcome
	Reason  *EnqueueNonReplaceableReason
}

EnqueueResult contains a job's stable identity and durable enqueue disposition.

type ExecutionTimeoutError 

type ExecutionTimeoutError struct {
	JobID   string
	Attempt int
}

ExecutionTimeoutError identifies an attempt whose active execution budget was consumed.

func (*ExecutionTimeoutError) Error 

func (err *ExecutionTimeoutError) Error() string

func (*ExecutionTimeoutError) Unwrap 

func (err *ExecutionTimeoutError) Unwrap() error

type Executor 

type Executor interface {
	Query(ctx context.Context, statement string, arguments ...any) ([]Row, error)
}

Executor runs SQL with PostgreSQL positional parameters and returns all result rows.

func NewPGXExecutor 

func NewPGXExecutor(queryer PGXQueryer) Executor

NewPGXExecutor adapts a caller-owned pgx transaction, connection, or pool.

func NewSQLExecutor 

func NewSQLExecutor(queryer SQLQueryer) Executor

NewSQLExecutor adapts a caller-owned database/sql transaction, connection, or database.

type ExternalWait 

type ExternalWait struct {
	JobID      string
	Queue      string
	JobType    string
	Name       string
	Context    any
	Attempt    int
	CreatedAt  time.Time
	DeadlineAt *time.Time
}

type ExternalWaitDelivery 

type ExternalWaitDelivery struct {
	IdempotencyKey string
	RequestedBy    string
}

ExternalWaitDelivery supplies idempotency and trusted attribution for an external value.

type ExternalWaitOptions 

type ExternalWaitOptions struct {
	Timeout time.Duration
}

ExternalWaitOptions controls the PostgreSQL-owned deadline for a signal or human wait.

type ExternalWaitPage 

type ExternalWaitPage struct {
	Items      []ExternalWait
	NextCursor *AdminCursor
}

type ExternalWaitQuery 

type ExternalWaitQuery struct {
	Limit  int
	Cursor *AdminCursor
}

type Handler 

type Handler func(context.Context, any, *HandlerContext) (any, error)

Handler processes one claimed payload with fenced durable operations outside a transaction.

type HandlerContext 

type HandlerContext struct {
	Job ClaimedJob
	// contains filtered or unexported fields
}

HandlerContext exposes fenced durable operations for one claimed handler activation.

func (*HandlerContext) Checkpoint 

func (handler *HandlerContext) Checkpoint(
	name string,
	operation func() (any, error),
) (any, error)

Checkpoint returns the stored value for name, or runs and immutably saves operation once.

func (*HandlerContext) CreateChild deprecated 

func (handler *HandlerContext) CreateChild(
	name string,
	jobType string,
	payload any,
	options ...EnqueueOptions,
) (any, error)

CreateChild creates one named child or joins its retained result after parent replay.

Deprecated: renamed to RunChild, which is the name the other two SDKs share.

func (*HandlerContext) CreateChildren deprecated 

func (handler *HandlerContext) CreateChildren(children []ChildJobRequest) ([]ChildResult, error)

CreateChildren creates one bounded child set or joins its results after parent replay.

Deprecated: renamed to RunChildren, which is the name the other two SDKs share.

func (*HandlerContext) CreateChildrenAll deprecated 

func (handler *HandlerContext) CreateChildrenAll(children []ChildJobRequest) ([]ChildSuccessResult, error)

CreateChildrenAll preserves propagation semantics and returns only successful child results.

Deprecated: renamed to RunChildrenAll, which is the name the other two SDKs share.

func (*HandlerContext) GetProgress 

func (handler *HandlerContext) GetProgress() (*JobProgress, error)

GetProgress returns the latest progress observed by this handler activation.

func (*HandlerContext) RunChild 

func (handler *HandlerContext) RunChild(
	name string,
	jobType string,
	payload any,
	options ...EnqueueOptions,
) (any, error)

RunChild creates one named child or joins its retained result after parent replay.

func (*HandlerContext) RunChildren 

func (handler *HandlerContext) RunChildren(children []ChildJobRequest) ([]ChildResult, error)

RunChildren creates one bounded child set or joins its results after parent replay. Results retain request order, while Name provides stable keyed lookup to callers.

func (*HandlerContext) RunChildrenAll 

func (handler *HandlerContext) RunChildrenAll(children []ChildJobRequest) ([]ChildSuccessResult, error)

RunChildrenAll preserves propagation semantics and returns only successful child results.

func (*HandlerContext) SetProgress 

func (handler *HandlerContext) SetProgress(value any) (*JobProgress, error)

SetProgress replaces the latest progress under this handler's fenced lease.

func (*HandlerContext) Sleep 

func (handler *HandlerContext) Sleep(name string, duration time.Duration) error

Sleep suspends the job without consuming its logical attempt until duration elapses.

func (*HandlerContext) SleepUntil 

func (handler *HandlerContext) SleepUntil(name string, wakeAt time.Time) error

SleepUntil suspends the job without consuming its logical attempt until wakeAt.

func (*HandlerContext) WaitForHuman 

func (handler *HandlerContext) WaitForHuman(
	name string,
	waitContext any,
	options ...ExternalWaitOptions,
) (any, error)

WaitForHuman suspends the job until a named decision is completed, then returns its JSON result.

func (*HandlerContext) WaitForSignal 

func (handler *HandlerContext) WaitForSignal(name string, options ...ExternalWaitOptions) (any, error)

WaitForSignal suspends the job until a named signal is delivered, then returns its JSON payload.

type HumanWaitAlreadyWaitingError 

type HumanWaitAlreadyWaitingError struct {
	JobID    string
	WaitName string
}

HumanWaitAlreadyWaitingError identifies a human wait already owned by another activation.

func (*HumanWaitAlreadyWaitingError) Error 

func (err *HumanWaitAlreadyWaitingError) Error() string

type HumanWaitCompletionResult 

type HumanWaitCompletionResult struct {
	Status      HumanWaitCompletionStatus
	JobID       string
	Name        string
	Payload     any
	CompletedAt *time.Time
	CompletedBy string
}

HumanWaitCompletionResult contains the accepted or retained human decision.

type HumanWaitCompletionStatus 

type HumanWaitCompletionStatus string

HumanWaitCompletionStatus is PostgreSQL's disposition for a human decision. 

const (
	HumanWaitCompleted        HumanWaitCompletionStatus = externalCompletedValue
	HumanWaitDuplicate        HumanWaitCompletionStatus = externalDuplicateValue
	HumanWaitNotWaiting       HumanWaitCompletionStatus = externalNotWaitingValue
	HumanWaitAlreadyCompleted HumanWaitCompletionStatus = externalAlreadyCompletedValue
	HumanWaitStale            HumanWaitCompletionStatus = durableStaleValue
	HumanWaitNotFound         HumanWaitCompletionStatus = externalNotFoundValue
)

type HumanWaitConflictError 

type HumanWaitConflictError struct {
	JobID    string
	WaitName string
}

HumanWaitConflictError identifies a human wait name replayed with different context.

func (*HumanWaitConflictError) Error 

func (err *HumanWaitConflictError) Error() string

type HumanWaitIdempotencyConflictError 

type HumanWaitIdempotencyConflictError struct {
	JobID    string
	WaitName string
}

HumanWaitIdempotencyConflictError identifies a retained key reused with another decision.

func (*HumanWaitIdempotencyConflictError) Error 

func (err *HumanWaitIdempotencyConflictError) Error() string

type HumanWaitLeaseLostError 

type HumanWaitLeaseLostError struct {
	JobID    string
	WaitName string
}

HumanWaitLeaseLostError identifies a human wait rejected under a stale fence.

func (*HumanWaitLeaseLostError) Error 

func (err *HumanWaitLeaseLostError) Error() string

func (*HumanWaitLeaseLostError) Unwrap 

func (err *HumanWaitLeaseLostError) Unwrap() error

type HumanWaitLimitExceededError 

type HumanWaitLimitExceededError struct{ JobID string }

HumanWaitLimitExceededError identifies a job that owns the supported number of human waits.

func (*HumanWaitLimitExceededError) Error 

func (err *HumanWaitLimitExceededError) Error() string

type Idempotency 

type Idempotency struct {
	Key   string `json:"key"`
	Scope string `json:"scope"`
	TTLMS int    `json:"ttlMs"`
}

Idempotency retains one canonical request under a scoped key.

type JobCheckpoint 

type JobCheckpoint struct {
	JobID      string
	Name       string
	Value      any
	Attempt    int
	FenceToken int64
	WorkerID   string
	CreatedAt  time.Time
}

type JobContractUnavailableError 

type JobContractUnavailableError struct {
	JobType string
	Version string
}

func (*JobContractUnavailableError) Error 

func (err *JobContractUnavailableError) Error() string

type JobContractValidationError 

type JobContractValidationError struct {
	JobType string
	Version string
	Kind    string
}

func (*JobContractValidationError) Error 

func (err *JobContractValidationError) Error() string

type JobContractVersion 

type JobContractVersion struct {
	PayloadSchema        any
	ResultSchema         any
	MaxPayloadBytes      int
	MaxResultBytes       int
	SensitivePayloadKeys []string
	SensitiveResultKeys  []string
}

type JobListItem 

type JobListItem struct {
	ID                 string
	Queue              string
	Type               string
	ConcurrencyKey     *string
	Priority           int
	Tags               []string
	State              JobState
	PrerequisiteJobID  *string
	PrerequisiteJobIDs []string
	BlockedReason      *string
	ParentJobID        *string
	ChildJobIDs        []string
	CurrentAttempt     int
	MaxAttempts        int
	RetryPolicy        map[string]any
	DeadlineAt         *time.Time
	ExecutionTimeoutMS *int64
	RunAt              time.Time
	CancelRequestedAt  *time.Time
	CancelRequestedBy  *string
	CancelReason       *string
	CreatedAt          time.Time
	UpdatedAt          time.Time
	Payload            any
	PayloadStatus      string
	PayloadBytes       *int64
}

type JobListPage 

type JobListPage struct {
	Items      []JobListItem
	NextCursor *AdminCursor
}

type JobListQuery 

type JobListQuery struct {
	Queue             string
	Type              string
	States            []JobState
	CreatedAfter      *time.Time
	CreatedBefore     *time.Time
	IncludePayload    bool
	PayloadMaxBytes   int
	PayloadRedactKeys []string
	Limit             int
	Cursor            *AdminCursor
}

type JobProgress 

type JobProgress struct {
	JobID      string
	Value      any
	Revision   int64
	Attempt    int
	FenceToken int64
	WorkerID   string
	CreatedAt  time.Time
	UpdatedAt  time.Time
}

JobProgress is the latest mutable progress projection for a stable job identity.

type JobSnapshot 

type JobSnapshot struct {
	JobListItem
	ContractVersion *string
	FenceToken      int64
	Result          any
	Error           any
	Progress        *JobProgress
}

type JobState 

type JobState string

JobState is PostgreSQL's durable lifecycle state for a job. 

const (
	JobBlocked   JobState = jobBlockedValue
	JobScheduled JobState = jobScheduledValue
	JobReady     JobState = jobReadyValue
	JobActive    JobState = jobActiveValue
	JobSucceeded JobState = jobSucceededValue
	JobFailed    JobState = jobFailedValue
	JobCanceled  JobState = jobCanceledValue
)

type JobTimelineEntry 

type JobTimelineEntry struct {
	Kind       string
	RecordID   string
	Priority   int
	Attempt    *int
	EventType  *string
	Details    any
	FenceToken *int64
	WorkerID   *string
	Outcome    *string
	StartedAt  *time.Time
	ClaimedAt  *time.Time
	FinishedAt *time.Time
	Error      any
	OccurredAt time.Time
}

type JobTimelinePage 

type JobTimelinePage struct {
	Items      []JobTimelineEntry
	NextCursor *AdminCursor
}

type JobTimelineQuery 

type JobTimelineQuery struct {
	Limit  int
	Cursor *AdminCursor
}

type JobTypeContracts 

type JobTypeContracts struct {
	CurrentVersion string
	Versions       map[string]JobContractVersion
}

type JobWait 

type JobWait struct {
	JobID           string
	Name            string
	Mode            string
	DurationMS      *int64
	RequestedWakeAt *time.Time
	WakeAt          time.Time
	Attempt         int
	FenceToken      int64
	WorkerID        string
	CreatedAt       time.Time
}

type LeaseLostError 

type LeaseLostError struct{ JobID string }

LeaseLostError identifies the attempt whose fence PostgreSQL no longer accepts.

func (*LeaseLostError) Error 

func (err *LeaseLostError) Error() string

func (*LeaseLostError) Unwrap 

func (err *LeaseLostError) Unwrap() error

type PGXQueryer 

type PGXQueryer interface {
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
}

PGXQueryer is implemented by pgx.Tx, *pgx.Conn, and *pgxpool.Pool.

type ProgressLeaseLostError 

type ProgressLeaseLostError struct{ JobID string }

ProgressLeaseLostError identifies a progress write rejected under a stale fence.

func (*ProgressLeaseLostError) Error 

func (err *ProgressLeaseLostError) Error() string

func (*ProgressLeaseLostError) Unwrap 

func (err *ProgressLeaseLostError) Unwrap() error

type ProgressRateLimitError 

type ProgressRateLimitError struct {
	JobID      string
	RetryAfter time.Duration
}

ProgressRateLimitError reports when this ownership generation may next change progress.

func (*ProgressRateLimitError) Error 

func (err *ProgressRateLimitError) Error() string

type Queue 

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

Queue enqueues jobs through a caller-owned executor.

func NewQueue 

func NewQueue(executor Executor, defaultQueue string) *Queue

NewQueue constructs an enqueue client without taking ownership of the executor.

func (*Queue) Cancel 

func (queue *Queue) Cancel(
	ctx context.Context,
	jobID string,
	request CancellationRequest,
) (CancelResult, error)

Cancel requests cooperative cancellation with optional audit attribution.

func (*Queue) CompleteHumanWait 

func (queue *Queue) CompleteHumanWait(
	ctx context.Context,
	jobID string,
	name string,
	result any,
	delivery ExternalWaitDelivery,
) (HumanWaitCompletionResult, error)

CompleteHumanWait supplies one idempotent JSON result to a named human wait.

func (*Queue) Enqueue 

func (queue *Queue) Enqueue(
	ctx context.Context,
	jobType string,
	payload any,
	options ...EnqueueOptions,
) (string, error)

Enqueue submits one job and returns its stable identifier.

func (*Queue) EnqueueMany 

func (queue *Queue) EnqueueMany(ctx context.Context, requests []EnqueueRequest) ([]string, error)

EnqueueMany submits one atomic batch and returns the stable job identifiers in request order.

func (*Queue) EnqueueManyWithResults 

func (queue *Queue) EnqueueManyWithResults(
	ctx context.Context,
	requests []EnqueueRequest,
) ([]EnqueueResult, error)

EnqueueManyWithResults submits one atomic batch and returns canonical results in request order.

func (*Queue) EnqueueWithResult 

func (queue *Queue) EnqueueWithResult(
	ctx context.Context,
	jobType string,
	payload any,
	options ...EnqueueOptions,
) (EnqueueResult, error)

EnqueueWithResult submits one job and returns PostgreSQL's canonical result.

func (*Queue) Health 

func (queue *Queue) Health(ctx context.Context) (QueueHealth, error)

Health reads PostgreSQL's database-authoritative queue health snapshot.

func (*Queue) ListConcurrencyPolicies 

func (queue *Queue) ListConcurrencyPolicies(
	ctx context.Context,
	queueNames []string,
) ([]ConcurrencyPolicy, error)

ListConcurrencyPolicies returns persisted policies ordered by queue name. A nil or empty queue-name slice returns every policy.

func (*Queue) ListRateLimitPolicies 

func (queue *Queue) ListRateLimitPolicies(
	ctx context.Context,
	queueNames []string,
) ([]RateLimitPolicy, error)

ListRateLimitPolicies returns persisted policies ordered by queue name. A nil or empty queue-name slice returns every policy.

func (*Queue) SendSignal 

func (queue *Queue) SendSignal(
	ctx context.Context,
	jobID string,
	name string,
	payload any,
	delivery ExternalWaitDelivery,
) (SignalDeliveryResult, error)

SendSignal delivers one idempotent JSON payload to a named signal wait.

func (*Queue) SyncConcurrencyPolicies 

func (queue *Queue) SyncConcurrencyPolicies(
	ctx context.Context,
	namespace string,
	definitions []ConcurrencyPolicyDefinition,
	options ...SyncPolicyOptions,
) ([]ConcurrencyPolicy, error)

SyncConcurrencyPolicies atomically reconciles one namespace of concurrency policies. Omitted definitions are removed unless options explicitly set Prune to false.

func (*Queue) SyncContracts 

func (queue *Queue) SyncContracts(ctx context.Context, contracts map[string]JobTypeContracts) error

func (*Queue) SyncRateLimitPolicies 

func (queue *Queue) SyncRateLimitPolicies(
	ctx context.Context,
	namespace string,
	definitions []RateLimitPolicyDefinition,
	options ...SyncPolicyOptions,
) ([]RateLimitPolicy, error)

SyncRateLimitPolicies atomically reconciles one namespace of rate-limit policies. Omitted definitions are removed unless options explicitly set Prune to false.

func (*Queue) SyncSchedules 

func (queue *Queue) SyncSchedules(
	ctx context.Context,
	namespace string,
	definitions []ScheduleDefinition,
	options ...SyncSchedulesOptions,
) error

SyncSchedules atomically reconciles one namespace of recurring definitions. Omitted definitions are disabled unless options explicitly set Prune to false.

type QueueHealth 

type QueueHealth map[string]any

QueueHealth is PostgreSQL's versioned health document. Stable fields and reason codes are defined by queue_health_v1, so every language receives the same evaluation.

type RateLimit 

type RateLimit struct {
	Limit      int `json:"limit"`
	IntervalMS int `json:"intervalMs"`
	Burst      int `json:"burst"`
}

RateLimit is one continuously refilled PostgreSQL token bucket.

type RateLimitPolicy 

type RateLimitPolicy struct {
	Namespace string
	Queue     string
	Rate      RateLimit
	PerKey    *RateLimit
	UpdatedAt time.Time
}

RateLimitPolicy is one persisted and normalized queue rate-limit policy.

type RateLimitPolicyDefinition 

type RateLimitPolicyDefinition struct {
	Queue  string     `json:"queue"`
	Rate   RateLimit  `json:"rate"`
	PerKey *RateLimit `json:"perKey,omitempty"`
}

RateLimitPolicyDefinition is one queue's deployment-synchronized start-rate budget.

type RedriveResult 

type RedriveResult struct {
	Status      string
	SourceJobID string
	TargetJobID *string
	SourceState *JobState
	TargetState *JobState
	RequestedAt *time.Time
}

type Row 

type Row map[string]any

Row is one result row keyed by its PostgreSQL column name.

type SQLQueryer 

type SQLQueryer interface {
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}

SQLQueryer is implemented by *sql.Tx, *sql.Conn, and *sql.DB.

type ScheduleDefinition 

type ScheduleDefinition struct {
	Name     string
	Schedule string
	Timezone string
	Job      ScheduledJob
	Enabled  *bool
}

ScheduleDefinition is one desired recurring schedule. A nil Enabled value enables the definition by default.

type ScheduledJob 

type ScheduledJob struct {
	Type           string
	Payload        any
	Queue          string
	Priority       int
	ConcurrencyKey string
	MaxAttempts    int
	RetryPolicy    map[string]any
}

ScheduledJob describes the job created for each recurring occurrence.

type SignalDeliveryResult 

type SignalDeliveryResult struct {
	Status      SignalDeliveryStatus
	JobID       string
	Name        string
	Payload     any
	DeliveredAt *time.Time
	DeliveredBy string
}

SignalDeliveryResult contains the accepted or retained signal delivery.

type SignalDeliveryStatus 

type SignalDeliveryStatus string

SignalDeliveryStatus is PostgreSQL's disposition for a signal delivery. 

const (
	SignalDelivered        SignalDeliveryStatus = externalDeliveredValue
	SignalDuplicate        SignalDeliveryStatus = externalDuplicateValue
	SignalNotWaiting       SignalDeliveryStatus = externalNotWaitingValue
	SignalAlreadyDelivered SignalDeliveryStatus = externalAlreadyDeliveredValue
	SignalStale            SignalDeliveryStatus = durableStaleValue
	SignalNotFound         SignalDeliveryStatus = externalNotFoundValue
)

type SignalIdempotencyConflictError 

type SignalIdempotencyConflictError struct {
	JobID    string
	WaitName string
}

SignalIdempotencyConflictError identifies a retained key reused with another signal delivery.

func (*SignalIdempotencyConflictError) Error 

func (err *SignalIdempotencyConflictError) Error() string

type SignalWaitConflictError 

type SignalWaitConflictError struct {
	JobID    string
	WaitName string
}

SignalWaitConflictError identifies a signal name already waiting under another activation.

func (*SignalWaitConflictError) Error 

func (err *SignalWaitConflictError) Error() string

type SignalWaitLeaseLostError 

type SignalWaitLeaseLostError struct {
	JobID    string
	WaitName string
}

SignalWaitLeaseLostError identifies a signal wait rejected under a stale fence.

func (*SignalWaitLeaseLostError) Error 

func (err *SignalWaitLeaseLostError) Error() string

func (*SignalWaitLeaseLostError) Unwrap 

func (err *SignalWaitLeaseLostError) Unwrap() error

type SignalWaitLimitExceededError 

type SignalWaitLimitExceededError struct{ JobID string }

SignalWaitLimitExceededError identifies a job that owns the supported number of signal waits.

func (*SignalWaitLimitExceededError) Error 

func (err *SignalWaitLimitExceededError) Error() string

type StaleLeaseError 

type StaleLeaseError struct {
	JobID string
}

StaleLeaseError identifies the job whose fenced settlement PostgreSQL rejected.

func (*StaleLeaseError) Error 

func (err *StaleLeaseError) Error() string

func (*StaleLeaseError) Unwrap 

func (err *StaleLeaseError) Unwrap() error

type SyncPolicyOptions 

type SyncPolicyOptions struct {
	Prune bool
}

SyncPolicyOptions controls desired-state policy reconciliation.

type SyncSchedulesOptions 

type SyncSchedulesOptions struct {
	Prune bool
}

SyncSchedulesOptions controls desired-state reconciliation.

type Throttle 

type Throttle struct {
	Key      string `json:"key"`
	Scope    string `json:"scope"`
	WindowMS int    `json:"windowMs"`
}

Throttle accepts at most one equivalent keyed job during a PostgreSQL-owned window.

type WaitConflictError 

type WaitConflictError struct {
	JobID    string
	WaitName string
}

WaitConflictError identifies a durable wait name reused with another target.

func (*WaitConflictError) Error 

func (err *WaitConflictError) Error() string

type WaitLeaseLostError 

type WaitLeaseLostError struct {
	JobID    string
	WaitName string
}

WaitLeaseLostError identifies a durable wait rejected under a stale fence.

func (*WaitLeaseLostError) Error 

func (err *WaitLeaseLostError) Error() string

func (*WaitLeaseLostError) Unwrap 

func (err *WaitLeaseLostError) Unwrap() error

type WaitLimitExceededError 

type WaitLimitExceededError struct{ JobID string }

WaitLimitExceededError identifies a job that already owns the supported number of waits.

func (*WaitLimitExceededError) Error 

func (err *WaitLimitExceededError) Error() string

type Worker 

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

Worker claims and settles jobs through a caller-owned pool.

func NewWorker 

func NewWorker(pool *pgxpool.Pool, options WorkerOptions) (*Worker, error)

NewWorker constructs a bounded worker over a caller-owned pool.

func (*Worker) Handle 

func (worker *Worker) Handle(jobType string, handler Handler) *Worker

Handle registers the handler for a job type and returns the worker for chaining.

func (*Worker) HandleBatch 

func (worker *Worker) HandleBatch(
	jobType string,
	options BatchHandlerOptions,
	handler BatchHandler,
) *Worker

HandleBatch registers a process-local batch coordinator for one job type.

func (*Worker) Run 

func (worker *Worker) Run(ctx context.Context) error

Run listens and polls until the context is cancelled or an operational lifecycle error occurs, then drains.

func (*Worker) RunOnce 

func (worker *Worker) RunOnce(ctx context.Context) (bool, error)

RunOnce claims and processes at most one job.

type WorkerOptions 

type WorkerOptions struct {
	Queue                string
	Queues               []string
	WorkerID             string
	Concurrency          int
	LeaseDuration        time.Duration
	HeartbeatInterval    time.Duration
	PollInterval         time.Duration
	MaintenanceInterval  time.Duration
	RegistryInterval     time.Duration
	DisableRegistry      bool
	ScheduleNamespaces   []string
	ScheduleCatchupLimit int
	ShutdownGracePeriod  time.Duration
	PollingOnly          bool
	Logger               *slog.Logger
	OnRegistrationError  func(error)
}

WorkerOptions configures a bounded worker with notification-assisted polling.

type WorkerPauseResult 

type WorkerPauseResult struct {
	WorkerID    string
	Paused      bool
	RequestedAt time.Time
	RequestedBy string
	Reason      string
}

type WorkerRegistryEntry 

type WorkerRegistryEntry struct {
	WorkerID        string
	InstanceID      string
	Hostname        string
	PID             int
	QueueNames      []string
	Queue           string
	Concurrency     int
	ActiveSlots     int
	Draining        bool
	Paused          bool
	PausedBy        *string
	Reason          *string
	PausedAt        *time.Time
	StartedAt       time.Time
	LastHeartbeatAt time.Time
}

Source Files

View all Source files

Directories

Path Synopsis
Package dashboard exposes the versioned, language-neutral dashboard browser bundle.
 Package dashboard exposes the versioned, language-neutral dashboard browser bundle.
cmd/conformance command
Command conformance exposes the Go dashboard backend to the shared HTTP fixture runner.
 Command conformance exposes the Go dashboard backend to the shared HTTP fixture runner.
examples
dedicated-worker command
demo-worker command
orchestration command
quickstart command
transaction command

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.