orderedjob

package module
v1.6.0 Latest Latest
Warning

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

 Go to latest Published: Sep 14, 2026 License: MIT Imports: 14 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 antar-chain.
Storage engine terdistribusi digunakan untuk locking, ordering, retry, dan recovery secara aman.

📌 Daftar Isi

📦 Instalasi

go get github.com/semmidev/orderedjob

✨ Fitur Utama

  • 🔒 Per-Chain Ordering: Job $N$ hanya berjalan setelah Job $N-1$ berstatus COMPLETED.
  • 🚀 Paralelisme Antar Chain: Ribuan chain berjalan bersamaan secara independen tanpa lock contention.
  • 🗄️ Multi-Database Support: Dukungan penuh persistence engine untuk PostgreSQL dan Microsoft SQL Server.
  • 🔔 Instant Event Notification: Menggunakan PostgreSQL LISTEN/NOTIFY (atau polling interval ber-jitter) untuk latensi klaim < 5ms.
  • 🎯 Type-Safe Handlers: Pendaftaran handler menggunakan Go Generics (RegisterTyped[T]) dengan unmarshaling JSON otomatis.
  • 🛡️ Fencing Protection: Fencing token (lease_generation) mencegah race condition dari worker yang terhambat (lagging).
  • 🛠️ Manajemen DLQ: Menyediakan API ReplayJob dan SkipJob untuk memulihkan atau melewati job yang gagal.
  • 🌐 Distributed Tracing: Dukungan TraceContext propagation (TraceID) antar pemanggilan asinkron.
  • 🔌 In-Memory Adapter: Adapter memory bawaan untuk pengujian cepat tanpa database.

🗄️ Dukungan Database

orderedjob menyediakan arsitektur storage engine yang pluggable. Anda dapat memilih adapter database yang sesuai dengan infrastruktur stack aplikasi Anda:

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, OUTPUT Clause GA (Production Ready)
In-Memory repository/memory In-Memory Struct Mutex Locking (Tanpa Database) Testing / Local Dev

💡 Contoh Usecase

  • ✅ Multi-Stage Approval Workflow: Memproses alur persetujuan bertahap (Approve A ➔ Approve B ➔ ... ➔ Approve E) secara sekuensial per dokumen/permohonan dengan pengiriman API inter-service otomatis.
  • 💳 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.
  • 👤 User Onboarding Workflow: Eksekusi berurutan untuk registrasi pengguna (Create User ➔ Send Verification Email ➔ Provision Default Workspace ➔ Trigger Analytics Event).
  • 🔄 Event-Driven State Machines: Mengolah stream kejadian berurutan (CDC / Event Sourcing) yang membutuhkan jaminan eksekusi FIFO per entity ID atau tenant ID.
  • 🏢 SaaS Multi-Tenant Background Jobs: Menjalankan background processing berat dengan isolasi antar tenant tanpa memblokir pemrosesan tenant lain.

🏗️ Arsitektur

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\n(Partial Index)"]
        WorkerPool["Worker Pool\n(Goroutines)"]
        Registry["Handler Registry\n(Type-Safe Generics)"]
        Recovery["Recovery Worker\n(Stale Lease Recovery)"]
    end

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

🚦 State Machine

stateDiagram-v2
    [*] --> BLOCKED: Sequence > 1 (Predecessor Pending)
    [*] --> PENDING: Sequence = 1 / Auto Sequence
    
    BLOCKED --> PENDING: Predecessor Status = COMPLETED
    PENDING --> PROCESSING: Worker Claim (Acquire Lease)
    RETRYING --> PROCESSING: Worker Claim (Acquire Lease)
    
    PROCESSING --> COMPLETED: Handler Success (Promote Next Sequence)
    PROCESSING --> RETRYING: Transient Failure (Backoff & Jitter)
    PROCESSING --> FAILED: Terminal Error / Max Attempts Reached
    PROCESSING --> CANCEL_REQUESTED: User Cancel Request
    
    CANCEL_REQUESTED --> CANCELLED: Worker Confirm Cancel
    FAILED --> DEAD_LETTERED: Final DLQ State
    
    COMPLETED --> [*]
    CANCELLED --> [*]
    DEAD_LETTERED --> [*]

[!IMPORTANT] Hanya status COMPLETED yang melepaskan urutan job berikutnya (sequence N+1). Status FAILED, CANCELLED, atau EXPIRED akan memblokir chain secara default demi keamanan data.

🚀 Contoh Penggunaan

Untuk melihat contoh kode aplikasi lengkap yang mencakup integrasi database, pencatatan log terstruktur (log/slog), metrik Prometheus / OpenTelemetry, penanganan error retryable, enkui batch satu transaksi DB, dan propagasi context tracing, silakan kunjungi folder example/.

🛠️ Penanganan Job Gagal (DLQ Operations)

Jika sebuah job berstatus FAILED, chain akan terhenti sampai ada penanganan manual:

Replay Job

Mengembalikan job FAILED atau CANCELLED ke status PENDING untuk dicoba ulang:

err := eng.ReplayJob(ctx, failedJobID)

Skip Job

Menandai job FAILED sebagai COMPLETED agar urutan berikutnya dapat dilanjutkan:

err := eng.SkipJob(ctx, failedJobID)

⚙️ Opsi Konfigurasi Engine

Option Deskripsi Default
WithConcurrency(n) Jumlah worker goroutine 10
WithPollInterval(d) Interval polling fallback (dengan random jitter) 500ms
WithLease(d) Durasi sewa job per worker (heartbeat = lease/3) 60s
WithRetryPolicy(p) Konfigurasi backoff retry (MaxAttempts, BaseDelay, MaxDelay, Jitter) Max 5, Base 1s, Max 5m
WithSlogLogger(l) Integrasi log/slog NoopLogger
WithMetrics(m) Integrasi Prometheus / OpenTelemetry Metrics NoopMetrics

🧪 Testing

Gunakan adapter memory untuk unit testing tanpa database:

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

repo := memory.New()
eng := orderedjob.New(repo, orderedjob.WithConcurrency(5))

Menjalankan test dan linter:

make test
make lint

📄 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")
)

Functions

func AggregateChainStats 

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

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

func ExtractTraceID 

func ExtractTraceID(ctx context.Context) string

ExtractTraceID extracts trace ID from context.

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.

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) 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) 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) 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 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
	Search   string
	OrderBy  string
	OrderDir string
	Offset   int
	Limit    int
}

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()
}

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 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) 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 Option 

type Option func(*Engine)

func WithConcurrency 

func WithConcurrency(n int) 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 WithPollInterval 

func WithPollInterval(d time.Duration) Option

func WithRecoveryInterval 

func WithRecoveryInterval(d time.Duration) Option

func WithRetryPolicy 

func WithRetryPolicy(p retry.Policy) Option

func WithSlogLogger 

func WithSlogLogger(l *slog.Logger) 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 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
	SkipJob(ctx context.Context, id uuid.UUID) 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"`
	Total           int64 `json:"total"`
	ActiveChains    int64 `json:"active_chains"`
}

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.