orderedjob

package module
v1.11.1 Latest Latest
Warning

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

 Go to latest Published: Sep 14, 2026 License: MIT Imports: 23 Imported by: 0

README

⚡ OrderedJob

High-Throughput, Pluggable Multi-Database Ordered Job Engine for Go

GoDoc Release CI License PostgreSQL SQL Server

github.com/semmidev/orderedjob mengeksekusi job secara berurutan (FIFO per chain) dengan eksekusi paralel tinggi antar-chain.
Dilengkapi storage engine terdistribusi aman untuk locking, ordering, retry, recovery, dan observabilitas OpenTelemetry.

📌 Daftar Isi

📦 Instalasi

go get github.com/semmidev/orderedjob

📖 Panduan Penggunaan Fitur Lengkap (GUIDE.md)

[!IMPORTANT] Panduan Penggunaan Fitur dari Basic hingga Advanced: Untuk melihat contoh kode Go lengkap (go), penjelasan parameter, serta efek samping (side effects) dari seluruh fitur 13 skenario lengkap (seperti Auto-Sequence, OpenTelemetry Tracing, Panic Isolation, Webhooks, SSE Web UI, Struct Validation, dan Bulk DLQ), silakan baca GUIDE.md (Panduan Penggunaan Fitur Lengkap).

✨ Fitur Utama

  • 🔒 Per-Chain Ordering: Jaminan urutan FIFO ketat per chain_id tanpa race condition.
  • 🚀 100k Concurrent Chains: Eksekusi paralel masal antar-chain dengan zero cross-chain lock contention.
  • 🗄️ Multi-Database Support: Adapter terintegrasi penuh untuk PostgreSQL (FOR UPDATE SKIP LOCKED) dan Microsoft SQL Server (WITH (UPDLOCK, READPAST)).
  • ⚡ Sub-5ms Latency: Sinyal instan LISTEN/NOTIFY (atau polling interval jitter) untuk pengambil pekerjaan seketika.
  • 🎯 Type-Safe Handlers & Validation: Pendaftaran handler menggunakan Go Generics (RegisterTyped[T]) dengan deserialisasi JSON dan validator skema struct otomatis.
  • 🛡️ Panic Recovery Guard: Mengisolasi unhandled panic pada handler goroutine, mencatat stack trace lengkap, dan menjaga worker pool tetap stabil.
  • ⚙️ Policy-Driven Ordering: Mode urutan fleksibel (strict, skip-on-failure, dan dead-letter-and-continue).
  • 🛠️ Bulk DLQ & Webhooks: Replay, skip, dan purge masal untuk job bermasalah, dilengkapi pengiriman notifikasi HTTP Webhook otomatis.
  • 🔭 Native OpenTelemetry & Prometheus: Integrasi penelusuran terdistribusi (TraceID) dan pengumpul metrik queue latency & duration.
  • 🖥️ Live SSE Web UI Dashboard: Antarmuka pemantauan antrean real-time berbasis Server-Sent Events (/ui) lengkap dengan payload editor.

🗄️ Dukungan Database

orderedjob menyediakan arsitektur storage engine yang pluggable:

Database Engine Package Path Driver / Client Fitur Penguncian & Koordinasi Status
PostgreSQL repository/postgres pgxpool.Pool FOR UPDATE SKIP LOCKED, LISTEN/NOTIFY, Advisory Lock GA (Production Ready)
Microsoft SQL Server repository/sqlserver *sql.DB WITH (UPDLOCK, READPAST), sp_getapplock GA (Production Ready)
In-Memory repository/memory In-Memory Struct Mutex Locking (Tanpa Database) Testing / Local Dev

⚡ Quick Start

package main

import (
	"context"
	"log"

	"github.com/semmidev/orderedjob"
	"github.com/semmidev/orderedjob/repository/memory"
)

type PaymentPayload struct {
	AccountID string  `json:"account_id"`
	Amount    float64 `json:"amount"`
}

func main() {
	ctx := context.Background()
	repo := memory.New()
	eng := orderedjob.New(repo, orderedjob.WithConcurrency(5))

	// Pendaftaran Type-Safe Handler
	orderedjob.RegisterTyped(eng, "ProcessPayment", func(ctx context.Context, job orderedjob.Job, payload PaymentPayload) error {
		log.Printf("Memproses pembayaran %s senilai %.2f", payload.AccountID, payload.Amount)
		return nil
	})

	// Jalankan Engine
	if err := eng.Start(ctx); err != nil {
		log.Fatalf("Gagal menjalankan engine: %v", err)
	}
	defer eng.Shutdown(ctx)

	// Enqueue Job Berurutan
	_, _ = eng.Enqueue(ctx, orderedjob.EnqueueRequest{
		ChainID:  "account-12345",
		Sequence: 1,
		Type:     "ProcessPayment",
		Payload:  PaymentPayload{AccountID: "account-12345", Amount: 150.00},
	})
}

🏛️ Arsitektur Teknis

[!TIP] Dokumentasi Spesifikasi Arsitektur Lengkap: Untuk pemahaman mendalam mengenai siklus hidup eksekusi data, skema database DDL, diagram state machine, algoritma penguncian, penanganan edge cases, serta hasil pengujian skala besar 100.000 concurrent chains, silakan baca ARCH.md (Technical Architecture Specification) dan panduan skenario fitur di GUIDE.md.

Ringkasan Alur Data Core

graph TD
    Client["Client App / Service"] -->|"Enqueue(req)"| DB[("Storage Database\n(PostgreSQL / SQL Server)")]
    
    subgraph Engine["OrderedJob Engine"]
        Listen["Event Stream / Poller\n(< 5ms Latency)"]
        Scanner["SKIP LOCKED / READPAST Scanner"]
        WorkerPool["Worker Pool & Panic Guard"]
        Registry["Type-Safe Registry"]
    end

    DB -->|"Event / Poll"| Listen
    DB -->|"SKIP LOCKED / UPDLOCK"| Scanner
    Listen -->|"Wakeup Event"| WorkerPool
    Scanner -->|"Claim Job"| WorkerPool
    WorkerPool -->|"Execute"| Registry
    Registry -->|"Side Effects / Actions"| External["External APIs / DBs"]

💡 Contoh Usecase

  • 💳 Pipeline Transaksi Finansial: Menjamin siklus pembayaran (Validate Account ➔ Hold Balance ➔ Transfer ➔ Send Receipt) berjalan tepat berurutan per akun tanpa race condition.
  • 📦 Pemrosesan Pesanan E-Commerce: Memproses tahapan pesanan (Create Order ➔ Deduct Inventory ➔ Generate Invoice ➔ Ship Package) sesuai urutan per pesanan.
  • 🔄 Event-Driven State Machines: Mengolah stream kejadian berurutan (CDC / Event Sourcing) yang membutuhkan jaminan eksekusi FIFO per entity ID atau tenant ID.
  • 📖 Contoh Kode Aplikasi Lengkap: Kunjungi folder example/ dan baca GUIDE.md untuk melihat integrasi penuh dengan log/slog, metrik OpenTelemetry, HTTP UI, dan retry backoff.

🛠️ Penanganan DLQ & Webhooks

Jika sebuah job berstatus FAILED, engine menyediakan API operasional masal (Bulk Operations) untuk pemulihan:

// Replay masal job di DLQ
replayed, err := eng.BulkReplayDLQ(ctx, orderedjob.JobFilter{Status: orderedjob.StatusFailed})

// Skip masal job di DLQ
skipped, err := eng.BulkSkipDLQ(ctx, orderedjob.JobFilter{JobType: "LegacyTask"})

🧪 Testing & Benchmarks

Menjalankan suite unit test:

make test
# atau
go test -v ./...

Menjalankan suite benchmark performa skala besar (100.000 Concurrent Chains & Lock Contention):

make bench
# atau
go test -bench=. -benchmem ./...

📊 Hasil Benchmark Performa

Pengujian performa skala besar dilakukan menggunakan suite benchmark bawaan (engine_benchmark_test.go). Berikut adalah hasil benchmark riil yang dijalankan pada mesin lokal:

💻 Spesifikasi Lingkungan Pengujian (System Environment)

Component Hardware / Software Specification
Machine & Chip Apple Mac (Apple M1, 8 Cores)
RAM 8 GB Unified Memory
Operating System macOS 13.7.8 (darwin/arm64)
Go Version go1.27.0 darwin/arm64

⚡ Summary Hasil Benchmark (go test -bench=. -benchmem)

Benchmark Function Test Focus & Concurrency Duration / Op Memory / Op Allocations / Op Status / Deadlock Guard
BenchmarkLockContention_AutoSequenceAdvisoryLock High-Concurrency Auto-Sequence (Sequence = 0) across 20 goroutines 6.54 ms 647 KB 5,168 allocs PASS (0 Deadlocks)
BenchmarkEngine_ClaimThroughput Worker Claim & Processing Throughput (16 workers) 213 ms 127 MB 51,964 allocs PASS
BenchmarkEngine_Scale100kChains Parallel Inter-Chain Batch Processing (32 workers) 707 ms 450 MB 60,232 allocs PASS (Zero Cross-Chain Contention)

[!NOTE] Pengujian penguncian otomatis (Sequence = 0) mengonfirmasi bahwa penguncian Advisory Lock (PostgreSQL pg_advisory_xact_lock / MSSQL sp_getapplock) mampu menangani hingga ~152.000 klaim urutan per detik tanpa pernah mengalami deadlock atau race condition.

📄 Lisensi

MIT License

Documentation

Index

Constants

View Source 
const (
	OrderingModeStrict                = "strict"
	OrderingModeSkipOnFailure         = "skip-on-failure"
	OrderingModeDeadLetterAndContinue = "dead-letter-and-continue"
)

Ordering Mode Constants (Policy-Driven Execution Modes)

Variables

View Source 
var (
	ErrNotFound          = errors.New("orderedjob: job not found")
	ErrAlreadyExists     = errors.New("orderedjob: job already exists")
	ErrInvalidTransition = errors.New("orderedjob: invalid state transition")
	ErrLeaseConflict     = errors.New("orderedjob: lease conflict / fencing violation")
	ErrChainBlocked      = errors.New("orderedjob: chain blocked")
	ErrSequenceGap       = errors.New("orderedjob: sequence gap")
	ErrHandlerNotFound   = errors.New("orderedjob: handler not found")
	ErrNotClaimable      = errors.New("orderedjob: job not claimable")
	ErrDuplicateChainSeq = errors.New("orderedjob: duplicate chain_id, sequence")
	ErrJobNotEligible    = errors.New("orderedjob: job not eligible for rescheduling")
	ErrValidationFailed  = errors.New("orderedjob: payload validation failed")
)

Functions

func AggregateChainStats 

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

ProcessJobsParallel is an example of using rill for parallel post-processing (metrics aggregation).

func ExtractOTelTraceContext added in v1.8.0 

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

ExtractOTelTraceContext extracts TraceID from job into context.

func ExtractTraceID 

func ExtractTraceID(ctx context.Context) string

ExtractTraceID extracts trace ID from context.

func InjectOTelTraceContext added in v1.8.0 

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

InjectOTelTraceContext extracts W3C traceparent (or OTel SpanContext) from ctx and sets req.TraceID if req.TraceID is empty.

func IsNonRetryable 

func IsNonRetryable(err error) bool

func IsRetryable 

func IsRetryable(err error) bool

func LogJobAttrs 

func LogJobAttrs(job Job) []any

func NonRetryable 

func NonRetryable(err error) error

func ParallelClaimDemo 

func ParallelClaimDemo()

ParallelClaimDemo shows how rill can be used for fan-out execution if you ever need to process claimed jobs via pipeline.

func RegisterTyped 

func RegisterTyped[T any](e *Engine, jobType string, fn func(ctx context.Context, job Job, payload T) error)

RegisterTyped registers a type-safe handler function that automatically unmarshals JSON payloads. If payload type T implements Validator or ValidatorCtx interface, validation is executed automatically before calling fn.

func RegisterTypedWithValidator added in v1.7.0 

func RegisterTypedWithValidator[T any](
	e *Engine,
	jobType string,
	fn func(ctx context.Context, job Job, payload T) error,
	validator PayloadValidatorFunc[T],
)

RegisterTypedWithValidator registers a type-safe handler function with optional custom payload validation logic. Validation occurs after JSON unmarshaling and self-validation. If validation returns an error, execution fails immediately with a NonRetryable error wrapping ErrValidationFailed.

func Retryable 

func Retryable(err error) error

func WithTraceID 

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

WithTraceID attaches a trace ID to the context.

Types

type ChainFilter added in v1.4.0 

type ChainFilter struct {
	Search   string
	OrderBy  string
	OrderDir string
	Offset   int
	Limit    int
}

type ChainSummary added in v1.2.0 

type ChainSummary struct {
	ChainID      string `json:"chain_id"`
	TotalJobs    int64  `json:"total_jobs"`
	MaxSequence  int64  `json:"max_sequence"`
	PendingJobs  int64  `json:"pending_jobs"`
	FailedJobs   int64  `json:"failed_jobs"`
	LatestStatus string `json:"latest_status"`
}

type Engine 

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

Engine is the core orchestrator.

func New 

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

func (*Engine) BulkPurgeDLQ added in v1.9.0 

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

func (*Engine) BulkReplayDLQ added in v1.9.0 

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

func (*Engine) BulkSkipDLQ added in v1.9.0 

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

func (*Engine) Enqueue 

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

Enqueue delegates to repo.

func (*Engine) EnqueueBatch 

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

func (*Engine) EnqueueDelayed added in v1.7.0 

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

EnqueueDelayed enqueues a job scheduled for execution after delay.

func (*Engine) Get 

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

func (*Engine) JobTypes added in v1.5.0 

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

JobTypes returns all registered job type names in sorted order.

func (*Engine) NotifyChain 

func (e *Engine) NotifyChain(chainID string)

func (*Engine) Register 

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

Register handler.

func (*Engine) RegisterFunc 

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

func (*Engine) ReplayJob 

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

func (*Engine) RescheduleJob added in v1.7.0 

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

RescheduleJob modifies the scheduled available_at timestamp of a job.

func (*Engine) Schedule added in v1.7.0 

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

Schedule enqueues a job scheduled for execution at runAt.

func (*Engine) Shutdown 

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

func (*Engine) SkipJob 

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

func (*Engine) Start 

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

Start launches workers and recovery loop.

type EnqueueRequest 

type EnqueueRequest struct {
	ChainID        string     `json:"chain_id"`
	Sequence       int64      `json:"sequence"`
	Type           string     `json:"job_type"`
	Payload        any        `json:"payload"`
	IdempotencyKey string     `json:"idempotency_key,omitempty"`
	TenantID       string     `json:"tenant_id,omitempty"`
	TraceID        string     `json:"trace_id,omitempty"`
	OrderingMode   string     `json:"ordering_mode,omitempty"`
	MaxAttempts    int        `json:"max_attempts,omitempty"`
	DeadlineAt     *time.Time `json:"deadline_at,omitempty"`
	AvailableAt    *time.Time `json:"available_at,omitempty"`
}

EnqueueRequest is used by callers.

type EventListener added in v1.9.0 

type EventListener interface {
	OnJobFailed(ctx context.Context, job Job, err error)
	OnChainBlocked(ctx context.Context, chainID string, job Job)
	OnDLQThresholdExceeded(ctx context.Context, dlqCount int64)
}

EventListener defines callbacks for key queue lifecycle events.

type FunctionalEventListener added in v1.9.0 

type FunctionalEventListener struct {
	OnJobFailedFunc            func(ctx context.Context, job Job, err error)
	OnChainBlockedFunc         func(ctx context.Context, chainID string, job Job)
	OnDLQThresholdExceededFunc func(ctx context.Context, dlqCount int64)
}

FunctionalEventListener allows registering individual functions as EventListener.

func (*FunctionalEventListener) OnChainBlocked added in v1.9.0 

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

func (*FunctionalEventListener) OnDLQThresholdExceeded added in v1.9.0 

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

func (*FunctionalEventListener) OnJobFailed added in v1.9.0 

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

type Handler 

type Handler interface {
	Handle(ctx context.Context, job Job) error
}

type HandlerFunc 

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

func (HandlerFunc) Handle 

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

type Job 

type Job struct {
	ID             uuid.UUID       `json:"id"`
	ChainID        string          `json:"chain_id"`
	Sequence       int64           `json:"sequence"`
	Type           string          `json:"type"`
	Payload        json.RawMessage `json:"payload"`
	IdempotencyKey string          `json:"idempotency_key,omitempty"`
	TenantID       string          `json:"tenant_id,omitempty"`
	OrderingMode   string          `json:"ordering_mode,omitempty"`

	// runtime fields
	Status          string     `json:"status"`
	Attempt         int        `json:"attempt"`
	MaxAttempts     int        `json:"max_attempts"`
	AvailableAt     time.Time  `json:"available_at"`
	DeadlineAt      *time.Time `json:"deadline_at,omitempty"`
	WorkerID        string     `json:"worker_id,omitempty"`
	LeaseUntil      *time.Time `json:"lease_until,omitempty"`
	LeaseGeneration int        `json:"lease_generation"`
	LastError       string     `json:"last_error,omitempty"`
	CreatedAt       time.Time  `json:"created_at"`
	UpdatedAt       time.Time  `json:"updated_at"`
	StartedAt       *time.Time `json:"started_at,omitempty"`
	CompletedAt     *time.Time `json:"completed_at,omitempty"`
	FailedAt        *time.Time `json:"failed_at,omitempty"`

	TraceID string `json:"trace_id,omitempty"`
}

Job represents a unit of work bound to a chain.

func BatchEnqueueParallel 

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

BatchEnqueueParallel demonstrates rill usage for high-throughput enqueue validation / transformation. It validates payloads in parallel, then enqueues atomically via repository.

func (Job) IsTerminal 

func (j Job) IsTerminal() bool

type JobFilter added in v1.2.0 

type JobFilter struct {
	ChainID       string
	Status        string
	JobType       string
	TenantID      string
	TraceID       string
	Search        string
	OrderBy       string
	OrderDir      string
	Offset        int
	Limit         int
	ScheduledOnly bool
}

type Logger 

type Logger interface {
	Info(ctx context.Context, msg string, args ...any)
	Error(ctx context.Context, msg string, args ...any)
	Debug(ctx context.Context, msg string, args ...any)
}

Logger structured minimal per section 26.

type Metrics 

type Metrics interface {
	IncClaimed()
	IncCompleted(jobType string)
	IncFailed(jobType string)
	IncRetry(jobType string)
	ObserveExecDuration(jobType string, d time.Duration)
	ObserveQueueDelay(jobType string, d time.Duration)
	SetActiveLeases(n int)
	IncClaimConflicts()
	IncStaleRecovered(n int)
	IncBlockedChain()
	IncPanics(jobType string)
}

Metrics interface per proposal section 25.

type NonRetryableError 

type NonRetryableError struct{ Err error }

NonRetryableError marks a permanent failure.

func (NonRetryableError) Error 

func (e NonRetryableError) Error() string

func (NonRetryableError) Unwrap 

func (e NonRetryableError) Unwrap() error

type NoopEventListener added in v1.9.0 

type NoopEventListener struct{}

func (NoopEventListener) OnChainBlocked added in v1.9.0 

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

func (NoopEventListener) OnDLQThresholdExceeded added in v1.9.0 

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

func (NoopEventListener) OnJobFailed added in v1.9.0 

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

type NoopLogger 

type NoopLogger struct{}

func (NoopLogger) Debug 

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

func (NoopLogger) Error 

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

func (NoopLogger) Info 

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

type NoopMetrics 

type NoopMetrics struct{}

func (NoopMetrics) IncBlockedChain 

func (NoopMetrics) IncBlockedChain()

func (NoopMetrics) IncClaimConflicts 

func (NoopMetrics) IncClaimConflicts()

func (NoopMetrics) IncClaimed 

func (NoopMetrics) IncClaimed()

func (NoopMetrics) IncCompleted 

func (NoopMetrics) IncCompleted(string)

func (NoopMetrics) IncFailed 

func (NoopMetrics) IncFailed(string)

func (NoopMetrics) IncPanics added in v1.7.0 

func (NoopMetrics) IncPanics(string)

func (NoopMetrics) IncRetry 

func (NoopMetrics) IncRetry(string)

func (NoopMetrics) IncStaleRecovered 

func (NoopMetrics) IncStaleRecovered(int)

func (NoopMetrics) ObserveExecDuration 

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

func (NoopMetrics) ObserveQueueDelay 

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

func (NoopMetrics) SetActiveLeases 

func (NoopMetrics) SetActiveLeases(int)

type OTelMetrics added in v1.8.0 

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

OTelMetrics implements Metrics backed by OpenTelemetry Metrics API.

func NewOTelMetrics added in v1.8.0 

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

NewOTelMetrics creates a Metrics exporter backed by an OpenTelemetry Meter. If mp is nil, otel.GetMeterProvider() is used.

func (*OTelMetrics) IncBlockedChain added in v1.8.0 

func (m *OTelMetrics) IncBlockedChain()

func (*OTelMetrics) IncClaimConflicts added in v1.8.0 

func (m *OTelMetrics) IncClaimConflicts()

func (*OTelMetrics) IncClaimed added in v1.8.0 

func (m *OTelMetrics) IncClaimed()

func (*OTelMetrics) IncCompleted added in v1.8.0 

func (m *OTelMetrics) IncCompleted(jobType string)

func (*OTelMetrics) IncFailed added in v1.8.0 

func (m *OTelMetrics) IncFailed(jobType string)

func (*OTelMetrics) IncPanics added in v1.8.0 

func (m *OTelMetrics) IncPanics(jobType string)

func (*OTelMetrics) IncRetry added in v1.8.0 

func (m *OTelMetrics) IncRetry(jobType string)

func (*OTelMetrics) IncStaleRecovered added in v1.8.0 

func (m *OTelMetrics) IncStaleRecovered(n int)

func (*OTelMetrics) ObserveExecDuration added in v1.8.0 

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

func (*OTelMetrics) ObserveQueueDelay added in v1.8.0 

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

func (*OTelMetrics) SetActiveLeases added in v1.8.0 

func (m *OTelMetrics) SetActiveLeases(n int)

type Option 

type Option func(*Engine)

func WithConcurrency 

func WithConcurrency(n int) Option

func WithDLQThreshold added in v1.9.0 

func WithDLQThreshold(n int64) Option

func WithEventListener added in v1.9.0 

func WithEventListener(l EventListener) Option

func WithLease 

func WithLease(d time.Duration) Option

func WithLogger 

func WithLogger(l Logger) Option

func WithMetrics 

func WithMetrics(m Metrics) Option

func WithNotify 

func WithNotify(enabled bool) Option

func WithOrderingMode 

func WithOrderingMode(mode string) Option

func WithOrderingStrategy added in v1.6.0 

func WithOrderingStrategy(s OrderingStrategyResolver) Option

func WithPanicHandler added in v1.7.0 

func WithPanicHandler(h PanicHandler) Option

func WithPollInterval 

func WithPollInterval(d time.Duration) Option

func WithRecoveryInterval 

func WithRecoveryInterval(d time.Duration) Option

func WithRetryOnPanic added in v1.7.0 

func WithRetryOnPanic(retry bool) Option

func WithRetryPolicy 

func WithRetryPolicy(p retry.Policy) Option

func WithSlogLogger 

func WithSlogLogger(l *slog.Logger) Option

func WithTracer added in v1.8.0 

func WithTracer(t trace.Tracer) Option

func WithTracerProvider added in v1.8.0 

func WithTracerProvider(tp trace.TracerProvider) Option

func WithWebhook added in v1.9.0 

func WithWebhook(cfg WebhookConfig) Option

func WithWorkerID 

func WithWorkerID(id string) Option

type OrderingStrategyResolver added in v1.6.0 

type OrderingStrategyResolver func(req EnqueueRequest) string

OrderingStrategyResolver resolves ordering mode per job request dynamically.

type PanicError added in v1.7.0 

type PanicError struct {
	Value any
	Stack []byte
}

PanicError encapsulates a panic value and stack trace captured during job execution.

func (*PanicError) Error added in v1.7.0 

func (p *PanicError) Error() string

type PanicHandler added in v1.7.0 

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

PanicHandler handles panics caught during job execution.

type PayloadValidatorFunc added in v1.7.0 

type PayloadValidatorFunc[T any] func(ctx context.Context, payload T) error

PayloadValidatorFunc is a custom validation function type for a payload of type T.

type Registry 

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

Registry holds job_type -> handler

func NewRegistry 

func NewRegistry() *Registry

func (*Registry) Get 

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

func (*Registry) JobTypes added in v1.5.0 

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

JobTypes returns all registered job type names in sorted order.

func (*Registry) MustGet 

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

func (*Registry) Register 

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

type Repository 

type Repository interface {
	// Enqueue inserts a job. If req.Sequence == 0 it auto-assigns.
	Enqueue(ctx context.Context, req EnqueueRequest) (Job, error)
	// EnqueueBatch atomic if required.
	EnqueueBatch(ctx context.Context, reqs []EnqueueRequest) ([]Job, error)

	Get(ctx context.Context, id uuid.UUID) (Job, error)
	// GetByChainSeq fetches by chain+seq
	GetByChainSeq(ctx context.Context, chainID string, seq int64) (Job, error)

	// Claim finds and atomically marks a job as PROCESSING (C3, C4, C10).
	// Returns ErrNotFound if none eligible.
	Claim(ctx context.Context, workerID string, lease time.Duration) (Job, error)

	// Complete marks COMPLETED with fencing check (C10, C5).
	Complete(ctx context.Context, id uuid.UUID, workerID string, leaseGen int) error
	// Fail marks FAILED / DEAD_LETTERED
	Fail(ctx context.Context, id uuid.UUID, workerID string, leaseGen int, errMsg string, terminal bool) error
	// Retry marks RETRYING with next available_at (C6)
	Retry(ctx context.Context, id uuid.UUID, workerID string, leaseGen int, errMsg string, nextAvailable time.Time) error

	// Heartbeat extends lease
	Heartbeat(ctx context.Context, id uuid.UUID, workerID string, leaseGen int, lease time.Duration) error

	// Cancel
	RequestCancel(ctx context.Context, id uuid.UUID) error
	ConfirmCancel(ctx context.Context, id uuid.UUID, workerID string, leaseGen int) error

	// Recovery & DLQ Operations
	RecoverStale(ctx context.Context, limit int, retryPolicy func(attempt int) time.Duration) (int, error)
	ReplayJob(ctx context.Context, id uuid.UUID) error
	ReplayJobWithPayload(ctx context.Context, id uuid.UUID, newPayload json.RawMessage) error
	SkipJob(ctx context.Context, id uuid.UUID) error
	RescheduleJob(ctx context.Context, id uuid.UUID, availableAt time.Time) (Job, error)
	BulkReplayDLQ(ctx context.Context, filter JobFilter) (int64, error)
	BulkSkipDLQ(ctx context.Context, filter JobFilter) (int64, error)
	BulkPurgeDLQ(ctx context.Context, filter JobFilter) (int64, error)
	CountDLQ(ctx context.Context) (int64, error)

	// Make next eligible after completion (C5)
	PromoteNext(ctx context.Context, chainID string, completedSeq int64) error

	// List etc
	ListPendingChains(ctx context.Context) ([]string, error)

	// Inspection & Dashboard Methods
	GetStats(ctx context.Context) (Stats, error)
	ListJobs(ctx context.Context, filter JobFilter) ([]Job, int64, error)
	ListChains(ctx context.Context, filter ChainFilter) ([]ChainSummary, int64, error)

	// Real-time LISTEN/NOTIFY interface
	NotifyChannel() string
	Listen(ctx context.Context, callback func(chainID string)) error
}

Repository is the persistence port. All coordination decisions use DB time where consistency matters (C16).

type RetryableError 

type RetryableError struct{ Err error }

RetryableError marks a transient failure that should be retried.

func (RetryableError) Error 

func (e RetryableError) Error() string

func (RetryableError) Unwrap 

func (e RetryableError) Unwrap() error

type SlogLogger 

type SlogLogger struct{ L *slog.Logger }

func NewSlogLogger 

func NewSlogLogger(l *slog.Logger) *SlogLogger

func (*SlogLogger) Debug 

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

func (*SlogLogger) Error 

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

func (*SlogLogger) Info 

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

type Stats added in v1.2.0 

type Stats struct {
	Pending         int64 `json:"pending"`
	Blocked         int64 `json:"blocked"`
	Processing      int64 `json:"processing"`
	Completed       int64 `json:"completed"`
	Retrying        int64 `json:"retrying"`
	Failed          int64 `json:"failed"`
	DeadLettered    int64 `json:"dead_lettered"`
	CancelRequested int64 `json:"cancel_requested"`
	Cancelled       int64 `json:"cancelled"`
	Scheduled       int64 `json:"scheduled"`
	Total           int64 `json:"total"`
	ActiveChains    int64 `json:"active_chains"`
}

type Validator added in v1.7.0 

type Validator interface {
	Validate() error
}

Validator is an interface that payload structs can implement to perform self-validation.

type ValidatorCtx added in v1.7.0 

type ValidatorCtx interface {
	ValidateCtx(ctx context.Context) error
}

ValidatorCtx is an interface that payload structs can implement to perform self-validation with context.

type WebhookConfig added in v1.9.0 

type WebhookConfig struct {
	URL         string            `json:"url"`
	Secret      string            `json:"secret,omitempty"`
	Headers     map[string]string `json:"headers,omitempty"`
	SlackFormat bool              `json:"slack_format,omitempty"`
	HTTPClient  *http.Client      `json:"-"`
}

type WebhookDispatcher added in v1.9.0 

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

func NewWebhookDispatcher added in v1.9.0 

func NewWebhookDispatcher(cfg WebhookConfig) *WebhookDispatcher

func (*WebhookDispatcher) OnChainBlocked added in v1.9.0 

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

func (*WebhookDispatcher) OnDLQThresholdExceeded added in v1.9.0 

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

func (*WebhookDispatcher) OnJobFailed added in v1.9.0 

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

type WebhookPayload added in v1.9.0 

type WebhookPayload struct {
	Event     string    `json:"event"`
	Timestamp time.Time `json:"timestamp"`
	ChainID   string    `json:"chain_id,omitempty"`
	JobID     string    `json:"job_id,omitempty"`
	JobType   string    `json:"job_type,omitempty"`
	Error     string    `json:"error,omitempty"`
	DLQCount  int64     `json:"dlq_count,omitempty"`
}

Source Files

View all Source files

Directories

Path Synopsis
internal
clock
memory
postgres
sqlserver

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.