orderedjob

Sep 14, 2026

Changes in this version

type Engine

+ func (e *Engine) BulkPurgeDLQ(ctx context.Context, filter JobFilter) (int64, error)

+ func (e *Engine) BulkReplayDLQ(ctx context.Context, filter JobFilter) (int64, error)

+ func (e *Engine) BulkSkipDLQ(ctx context.Context, filter JobFilter) (int64, error)

+ type EventListener interface

+ OnChainBlocked func(ctx context.Context, chainID string, job Job)

+ OnDLQThresholdExceeded func(ctx context.Context, dlqCount int64)

+ OnJobFailed func(ctx context.Context, job Job, err error)

+ type FunctionalEventListener struct

+ OnChainBlockedFunc func(ctx context.Context, chainID string, job Job)

+ OnDLQThresholdExceededFunc func(ctx context.Context, dlqCount int64)

+ OnJobFailedFunc func(ctx context.Context, job Job, err error)

+ func (f *FunctionalEventListener) OnChainBlocked(ctx context.Context, chainID string, job Job)

+ func (f *FunctionalEventListener) OnDLQThresholdExceeded(ctx context.Context, dlqCount int64)

+ func (f *FunctionalEventListener) OnJobFailed(ctx context.Context, job Job, err error)

+ type NoopEventListener struct

+ func (NoopEventListener) OnChainBlocked(context.Context, string, Job)

+ func (NoopEventListener) OnDLQThresholdExceeded(context.Context, int64)

+ func (NoopEventListener) OnJobFailed(context.Context, Job, error)

type Option

+ func WithDLQThreshold(n int64) Option

+ func WithEventListener(l EventListener) Option

+ func WithWebhook(cfg WebhookConfig) Option

type Repository

+ BulkPurgeDLQ func(ctx context.Context, filter JobFilter) (int64, error)

+ BulkReplayDLQ func(ctx context.Context, filter JobFilter) (int64, error)

+ BulkSkipDLQ func(ctx context.Context, filter JobFilter) (int64, error)

+ CountDLQ func(ctx context.Context) (int64, error)

+ type WebhookConfig struct

+ HTTPClient *http.Client

+ Headers map[string]string

+ Secret string

+ SlackFormat bool

+ URL string

+ type WebhookDispatcher struct

+ func NewWebhookDispatcher(cfg WebhookConfig) *WebhookDispatcher

+ func (w *WebhookDispatcher) OnChainBlocked(ctx context.Context, chainID string, job Job)

+ func (w *WebhookDispatcher) OnDLQThresholdExceeded(ctx context.Context, dlqCount int64)

+ func (w *WebhookDispatcher) OnJobFailed(ctx context.Context, job Job, err error)

+ type WebhookPayload struct

+ ChainID string

+ DLQCount int64

+ Error string

+ Event string

+ JobID string

+ JobType string

+ Timestamp time.Time

Sep 14, 2026

Changes in this version

+ func ExtractOTelTraceContext(ctx context.Context, job Job) context.Context

+ func InjectOTelTraceContext(ctx context.Context, req *EnqueueRequest)

+ type OTelMetrics struct

+ func NewOTelMetrics(mp metric.MeterProvider) (*OTelMetrics, error)

+ func (m *OTelMetrics) IncBlockedChain()

+ func (m *OTelMetrics) IncClaimConflicts()

+ func (m *OTelMetrics) IncClaimed()

+ func (m *OTelMetrics) IncCompleted(jobType string)

+ func (m *OTelMetrics) IncFailed(jobType string)

+ func (m *OTelMetrics) IncPanics(jobType string)

+ func (m *OTelMetrics) IncRetry(jobType string)

+ func (m *OTelMetrics) IncStaleRecovered(n int)

+ func (m *OTelMetrics) ObserveExecDuration(jobType string, d time.Duration)

+ func (m *OTelMetrics) ObserveQueueDelay(jobType string, d time.Duration)

+ func (m *OTelMetrics) SetActiveLeases(n int)

type Option

+ func WithTracer(t trace.Tracer) Option

+ func WithTracerProvider(tp trace.TracerProvider) Option

Sep 14, 2026

Changes in this version

+ var ErrJobNotEligible = errors.New("orderedjob: job not eligible for rescheduling")

+ var ErrValidationFailed = errors.New("orderedjob: payload validation failed")

+ func RegisterTypedWithValidator[T any](e *Engine, jobType string, ...)

type Engine

+ func (e *Engine) EnqueueDelayed(ctx context.Context, req EnqueueRequest, delay time.Duration) (Job, error)

+ func (e *Engine) RescheduleJob(ctx context.Context, id uuid.UUID, newAvailableAt time.Time) (Job, error)

+ func (e *Engine) Schedule(ctx context.Context, req EnqueueRequest, runAt time.Time) (Job, error)

type JobFilter

+ ScheduledOnly bool

+ TenantID string

+ TraceID string

type Metrics

+ IncPanics func(jobType string)

type NoopMetrics

+ func (NoopMetrics) IncPanics(string)

type Option

+ func WithPanicHandler(h PanicHandler) Option

+ func WithRetryOnPanic(retry bool) Option

+ type PanicError struct

+ Stack []byte

+ Value any

+ func (p *PanicError) Error() string

+ type PanicHandler func(ctx context.Context, job Job, panicVal any, stack []byte) error

+ type PayloadValidatorFunc func(ctx context.Context, payload T) error

type Repository

+ ReplayJobWithPayload func(ctx context.Context, id uuid.UUID, newPayload json.RawMessage) error

+ RescheduleJob func(ctx context.Context, id uuid.UUID, availableAt time.Time) (Job, error)

type Stats

+ Scheduled int64

+ type Validator interface

+ Validate func() error

+ type ValidatorCtx interface

+ ValidateCtx func(ctx context.Context) error

Sep 14, 2026

Changes in this version

+ const OrderingModeDeadLetterAndContinue

+ const OrderingModeSkipOnFailure

+ const OrderingModeStrict

type EnqueueRequest

+ OrderingMode string

type Job

+ OrderingMode string

type Option

+ func WithOrderingStrategy(s OrderingStrategyResolver) Option

+ type OrderingStrategyResolver func(req EnqueueRequest) string

Sep 14, 2026

Changes in this version

type Engine

+ func (e *Engine) JobTypes() []string

type Registry

+ func (r *Registry) JobTypes() []string

Sep 14, 2026

Changes in this version

+ type ChainFilter struct

+ Limit int

+ Offset int

+ OrderBy string

+ OrderDir string

+ Search string

type JobFilter

+ OrderBy string

+ OrderDir string

Sep 14, 2026

Changes in this version

+ type ChainSummary struct

+ ChainID string

+ FailedJobs int64

+ LatestStatus string

+ MaxSequence int64

+ PendingJobs int64

+ TotalJobs int64

+ type JobFilter struct

+ ChainID string

+ JobType string

+ Limit int

+ Offset int

+ Search string

+ Status string

type Repository

+ GetStats func(ctx context.Context) (Stats, error)

+ ListChains func(ctx context.Context) ([]ChainSummary, error)

+ ListJobs func(ctx context.Context, filter JobFilter) ([]Job, int64, error)

+ type Stats struct

+ ActiveChains int64

+ Blocked int64

+ CancelRequested int64

+ Cancelled int64

+ Completed int64

+ DeadLettered int64

+ Failed int64

+ Pending int64

+ Processing int64

+ Retrying int64

+ Total int64

Sep 13, 2026

Changes in this version

+ var ErrAlreadyExists = errors.New("orderedjob: job already exists")

+ var ErrChainBlocked = errors.New("orderedjob: chain blocked")

+ var ErrDuplicateChainSeq = errors.New("orderedjob: duplicate chain_id, sequence")

+ var ErrHandlerNotFound = errors.New("orderedjob: handler not found")

+ var ErrInvalidTransition = errors.New("orderedjob: invalid state transition")

+ var ErrLeaseConflict = errors.New("orderedjob: lease conflict / fencing violation")

+ var ErrNotClaimable = errors.New("orderedjob: job not claimable")

+ var ErrNotFound = errors.New("orderedjob: job not found")

+ var ErrSequenceGap = errors.New("orderedjob: sequence gap")

+ func AggregateChainStats(jobs []Job) map[string]int64

+ func ExtractTraceID(ctx context.Context) string

+ func IsNonRetryable(err error) bool

+ func IsRetryable(err error) bool

+ func LogJobAttrs(job Job) []any

+ func NonRetryable(err error) error

+ func ParallelClaimDemo()

+ func RegisterTyped[T any](e *Engine, jobType string, ...)

+ func Retryable(err error) error

+ func WithTraceID(ctx context.Context, traceID string) context.Context

+ type Engine struct

+ func New(repo Repository, opts ...Option) *Engine

+ func (e *Engine) Enqueue(ctx context.Context, req EnqueueRequest) (Job, error)

+ func (e *Engine) EnqueueBatch(ctx context.Context, reqs []EnqueueRequest) ([]Job, error)

+ func (e *Engine) Get(ctx context.Context, id uuid.UUID) (Job, error)

+ func (e *Engine) NotifyChain(chainID string)

+ func (e *Engine) Register(jobType string, h Handler)

+ func (e *Engine) RegisterFunc(jobType string, fn func(ctx context.Context, job Job) error)

+ func (e *Engine) ReplayJob(ctx context.Context, id uuid.UUID) error

+ func (e *Engine) Shutdown(ctx context.Context) error

+ func (e *Engine) SkipJob(ctx context.Context, id uuid.UUID) error

+ func (e *Engine) Start(ctx context.Context) error

+ type EnqueueRequest struct

+ AvailableAt *time.Time

+ ChainID string

+ DeadlineAt *time.Time

+ IdempotencyKey string

+ MaxAttempts int

+ Payload any

+ Sequence int64

+ TenantID string

+ TraceID string

+ Type string

+ type Handler interface

+ Handle func(ctx context.Context, job Job) error

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

+ func (f HandlerFunc) Handle(ctx context.Context, job Job) error

+ type Job struct

+ Attempt int

+ AvailableAt time.Time

+ ChainID string

+ CompletedAt *time.Time

+ CreatedAt time.Time

+ DeadlineAt *time.Time

+ FailedAt *time.Time

+ ID uuid.UUID

+ IdempotencyKey string

+ LastError string

+ LeaseGeneration int

+ LeaseUntil *time.Time

+ MaxAttempts int

+ Payload json.RawMessage

+ Sequence int64

+ StartedAt *time.Time

+ Status string

+ TenantID string

+ TraceID string

+ Type string

+ UpdatedAt time.Time

+ WorkerID string

+ func BatchEnqueueParallel(ctx context.Context, eng *Engine, reqs []EnqueueRequest) ([]Job, error)

+ func (j Job) IsTerminal() bool

+ type Logger interface

+ Debug func(ctx context.Context, msg string, args ...any)

+ Error func(ctx context.Context, msg string, args ...any)

+ Info func(ctx context.Context, msg string, args ...any)

+ type Metrics interface

+ IncBlockedChain func()

+ IncClaimConflicts func()

+ IncClaimed func()

+ IncCompleted func(jobType string)

+ IncFailed func(jobType string)

+ IncRetry func(jobType string)

+ IncStaleRecovered func(n int)

+ ObserveExecDuration func(jobType string, d time.Duration)

+ ObserveQueueDelay func(jobType string, d time.Duration)

+ SetActiveLeases func(n int)

+ type NonRetryableError struct

+ Err error

+ func (e NonRetryableError) Error() string

+ func (e NonRetryableError) Unwrap() error

+ type NoopLogger struct

+ func (NoopLogger) Debug(context.Context, string, ...any)

+ func (NoopLogger) Error(context.Context, string, ...any)

+ func (NoopLogger) Info(context.Context, string, ...any)

+ type NoopMetrics struct

+ func (NoopMetrics) IncBlockedChain()

+ func (NoopMetrics) IncClaimConflicts()

+ func (NoopMetrics) IncClaimed()

+ func (NoopMetrics) IncCompleted(string)

+ func (NoopMetrics) IncFailed(string)

+ func (NoopMetrics) IncRetry(string)

+ func (NoopMetrics) IncStaleRecovered(int)

+ func (NoopMetrics) ObserveExecDuration(string, time.Duration)

+ func (NoopMetrics) ObserveQueueDelay(string, time.Duration)

+ func (NoopMetrics) SetActiveLeases(int)

+ type Option func(*Engine)

+ func WithConcurrency(n int) Option

+ func WithLease(d time.Duration) Option

+ func WithLogger(l Logger) Option

+ func WithMetrics(m Metrics) Option

+ func WithNotify(enabled bool) Option

+ func WithOrderingMode(mode string) Option

+ func WithPollInterval(d time.Duration) Option

+ func WithRecoveryInterval(d time.Duration) Option

+ func WithRetryPolicy(p retry.Policy) Option

+ func WithSlogLogger(l *slog.Logger) Option

+ func WithWorkerID(id string) Option

+ type Registry struct

+ func NewRegistry() *Registry

+ func (r *Registry) Get(jobType string) (Handler, bool)

+ func (r *Registry) MustGet(jobType string) (Handler, error)

+ func (r *Registry) Register(jobType string, h Handler)

+ type Repository interface

+ Claim func(ctx context.Context, workerID string, lease time.Duration) (Job, error)

+ Complete func(ctx context.Context, id uuid.UUID, workerID string, leaseGen int) error

+ ConfirmCancel func(ctx context.Context, id uuid.UUID, workerID string, leaseGen int) error

+ Enqueue func(ctx context.Context, req EnqueueRequest) (Job, error)

+ EnqueueBatch func(ctx context.Context, reqs []EnqueueRequest) ([]Job, error)

+ Fail func(ctx context.Context, id uuid.UUID, workerID string, leaseGen int, ...) error

+ Get func(ctx context.Context, id uuid.UUID) (Job, error)

+ GetByChainSeq func(ctx context.Context, chainID string, seq int64) (Job, error)

+ Heartbeat func(ctx context.Context, id uuid.UUID, workerID string, leaseGen int, ...) error

+ ListPendingChains func(ctx context.Context) ([]string, error)

+ Listen func(ctx context.Context, callback func(chainID string)) error

+ NotifyChannel func() string

+ PromoteNext func(ctx context.Context, chainID string, completedSeq int64) error

+ RecoverStale func(ctx context.Context, limit int, retryPolicy func(attempt int) time.Duration) (int, error)

+ ReplayJob func(ctx context.Context, id uuid.UUID) error

+ RequestCancel func(ctx context.Context, id uuid.UUID) error

+ Retry func(ctx context.Context, id uuid.UUID, workerID string, leaseGen int, ...) error

+ SkipJob func(ctx context.Context, id uuid.UUID) error

+ type RetryableError struct

+ Err error

+ func (e RetryableError) Error() string

+ func (e RetryableError) Unwrap() error

+ type SlogLogger struct

+ L *slog.Logger

+ func NewSlogLogger(l *slog.Logger) *SlogLogger

+ func (s *SlogLogger) Debug(ctx context.Context, msg string, args ...any)

+ func (s *SlogLogger) Error(ctx context.Context, msg string, args ...any)

+ func (s *SlogLogger) Info(ctx context.Context, msg string, args ...any)