orderedjob

package module
v1.12.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: 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 antar-chain.
Dilengkapi storage engine terdistribusi aman untuk locking, ordering, retry, recovery, dan observabilitas OpenTelemetry.

Daftar Isi

Instalasi

go get github.com/semmidev/orderedjob@latest

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

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

Panduan Penggunaan Fitur Lengkap

[!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).

Arsitektur Teknis & Siklus Hidup Eksekusi

Bagian ini menjelaskan secara mendalam arsitektur teknis, model data, siklus hidup eksekusi (execution lifecycle), algoritma penguncian (locking), serta penanganan kondisi khusus (edge cases) pada orderedjob.

1. Filosofi & Jaminan Sistem

orderedjob dirancang untuk menyelesaikan tantangan mendasar pada sistem terdistribusi: menjamin urutan eksekusi pekerjaan secara ketat (Strict FIFO per Chain) tanpa mengorbankan throughput secara keseluruhan.

Jaminan Utama (Core Guarantees)
  1. Strict Per-Chain FIFO: Job dengan Sequence N pada ChainID X hanya dapat dieksekusi setelah Job Sequence N-1 berstatus COMPLETED (kecuali dikonfigurasi dengan opsi kebijakan non-strict).
  2. Inter-Chain Concurrency: Pekerjaan pada chain X dan chain Y dieksekusi secara independen dan paralel tanpa saling mengunci (zero cross-chain lock contention).
  3. Sub-5ms Claim Latency: Memanfaatkan transmisi terstruktur PostgreSQL LISTEN/NOTIFY (atau polling jitter) untuk membangunkan worker goroutine seketika saat job baru hadir.
  4. Exactly-Once Semantics Guard (Fencing Token): Menggunakan lease_generation (fencing token) untuk mencegah efek samping ganda akibat worker yang mengalami lagging atau network partition.
  5. No Data Loss on Crashes: Jika worker mati mendadak saat memproses pekerjaan, mekanisme Stale Lease Recovery akan otomatis memulihkan pekerjaan tersebut.
  6. Multi-Database Support: Adapter terintegrasi penuh untuk PostgreSQL (SKIP LOCKED), Microsoft SQL Server (READPAST, UPDLOCK), dan In-Memory engine.

2. Model Data & Skema Database

Seluruh state penyimpanan dikelola di tabel ordered_jobs. Skema dirancang dengan indeks parsial untuk memaksimalkan performa kueri klaim.

DDL Skema Tabel PostgreSQL (ordered_jobs)
CREATE TABLE IF NOT EXISTS ordered_jobs (
    id              UUID PRIMARY KEY,
    chain_id        VARCHAR(255) NOT NULL,
    sequence        BIGINT NOT NULL,
    job_type        VARCHAR(255) NOT NULL,
    payload         JSONB NOT NULL,
    status          VARCHAR(50) NOT NULL,
    attempt         INT NOT NULL DEFAULT 0,
    max_attempts    INT NOT NULL DEFAULT 5,
    available_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    deadline_at     TIMESTAMPTZ,
    worker_id       VARCHAR(255),
    lease_until     TIMESTAMPTZ,
    lease_generation INT NOT NULL DEFAULT 0,
    last_error      TEXT,
    tenant_id       VARCHAR(255),
    idempotency_key VARCHAR(255),
    trace_id        VARCHAR(255),
    ordering_mode   VARCHAR(50) NOT NULL DEFAULT 'strict',
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    started_at      TIMESTAMPTZ,
    completed_at    TIMESTAMPTZ,
    failed_at       TIMESTAMPTZ,

    CONSTRAINT uq_ordered_jobs_chain_sequence UNIQUE (chain_id, sequence)
);

- - Indeks Parsial Utama untuk Klaim Cepat (High-Throughput)
CREATE INDEX IF NOT EXISTS idx_ordered_jobs_claimable
ON ordered_jobs (available_at, created_at)
WHERE status IN ('PENDING', 'RETRYING');

CREATE INDEX IF NOT EXISTS idx_ordered_jobs_chain_sequence
ON ordered_jobs (chain_id, sequence);

CREATE INDEX IF NOT EXISTS idx_ordered_jobs_tenant
ON ordered_jobs (tenant_id) WHERE tenant_id IS NOT NULL;

CREATE INDEX IF NOT EXISTS idx_ordered_jobs_idempotency
ON ordered_jobs (idempotency_key) WHERE idempotency_key IS NOT NULL;
DDL Skema Tabel Microsoft SQL Server (ordered_jobs)
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'ordered_jobs')
BEGIN
    CREATE TABLE ordered_jobs (
        id              UNIQUEIDENTIFIER PRIMARY KEY,
        chain_id        NVARCHAR(255) NOT NULL,
        sequence        BIGINT NOT NULL,
        job_type        NVARCHAR(255) NOT NULL,
        payload         NVARCHAR(MAX) NOT NULL,
        status          NVARCHAR(50) NOT NULL,
        attempt         INT NOT NULL DEFAULT 0,
        max_attempts    INT NOT NULL DEFAULT 5,
        available_at    DATETIMEOFFSET NOT NULL DEFAULT SYSDATETIMEOFFSET(),
        deadline_at     DATETIMEOFFSET NULL,
        worker_id       NVARCHAR(255) NULL,
        lease_until     DATETIMEOFFSET NULL,
        lease_generation INT NOT NULL DEFAULT 0,
        last_error      NVARCHAR(MAX) NULL,
        tenant_id       NVARCHAR(255) NULL,
        idempotency_key NVARCHAR(255) NULL,
        trace_id        NVARCHAR(255) NULL,
        ordering_mode   NVARCHAR(50) NOT NULL DEFAULT 'strict',
        created_at      DATETIMEOFFSET NOT NULL DEFAULT SYSDATETIMEOFFSET(),
        updated_at      DATETIMEOFFSET NOT NULL DEFAULT SYSDATETIMEOFFSET(),
        started_at      DATETIMEOFFSET NULL,
        completed_at    DATETIMEOFFSET NULL,
        failed_at       DATETIMEOFFSET NULL,

        CONSTRAINT uq_ordered_jobs_chain_sequence UNIQUE (chain_id, sequence)
    );
END;

3. Siklus Hidup Eksekusi Data

[Client Enqueue] ──> (1. Ingestion) ──> [DB Table: PENDING/BLOCKED]
                                             │
                                       (2. LISTEN/NOTIFY Wakeup)
                                             ▼
[Worker Pool] <── (3. FOR UPDATE SKIP LOCKED / READPAST) ── [DB Table: PROCESSING]
      │
 (4. Heartbeat)
      │
 (5. Execution Result & Panic Guard)
      ├── SUCCESS  ──> Update COMPLETED  ──> Unblock Sequence N+1 (BLOCKED -> PENDING)
      ├── RETRY    ──> Update RETRYING   ──> Recalculate Backoff + Jitter
      └── FAILURE  ──> Update FAILED/DLQ ──> Handle per Ordering Policy (Strict/Skip/Continue)
Fase 1: Ingesti Data (Enqueue Pipeline)

Ketika aplikasi memanggil eng.Enqueue(ctx, req) atau eng.EnqueueBatch(ctx, reqs):

  1. Validasi Input: Memastikan ChainID dan Type tidak kosong dan menolak sequence bernilai negatif.
  2. Deduplikasi Idempotensi (Idempotency Key Check): Jika IdempotencyKey diisi, repository memeriksa apakah job dengan key tersebut sudah pernah di-enqueue. Jika ditemukan, orderedjob mengembalikan objek Job yang sudah ada tanpa insert ganda.
  3. Penanganan Auto-Sequence (Sequence = 0): Mengeksekusi penguncian transaksi khusus (pg_advisory_xact_lock di PG atau sp_getapplock di MSSQL) dan mengambil urutan tertinggi saat ini (MAX(sequence) + 1).
  4. Penentuan Status Awal (PENDING vs BLOCKED):Jika Sequence == 1, status di-set ke PENDING. Jika Sequence > 1, repository mengecek status pendahulu (Sequence - 1). Jika pendahulu sudah COMPLETED, job baru berstatus PENDING; jika belum, berstatus BLOCKED.
  5. Disparasi Notifikasi Event: Mengirimkan sinyal NOTIFY ordered_jobs, 'chain_id' (PostgreSQL) atau memicu listener internal.
Fase 2: Notifikasi & Klaim Pekerjaan (Claim Pipeline)
  1. Event-Driven Listener (LISTEN Stream): Mendengarkan sinyal LISTEN ordered_jobs untuk membangunkan worker goroutine seketika (latensi klaim < 5ms).
  2. Fallback Polling Scanner: Polling interval periodik dengan random jitter untuk menjamin resiliensi jika notifikasi terlewat.
  3. Klaim Atomik dengan Fencing Token: Memanfaatkan kueri klaim atomik FOR UPDATE SKIP LOCKED (PostgreSQL) atau WITH (UPDLOCK, READPAST) (SQL Server):
WITH target AS (
    SELECT j.id
    FROM ordered_jobs j
    WHERE j.status IN ('PENDING', 'RETRYING')
      AND j.available_at <= NOW()
      AND (j.sequence = 1 OR EXISTS (
          SELECT 1 FROM ordered_jobs p
          WHERE p.chain_id = j.chain_id
            AND p.sequence = j.sequence - 1
            AND p.status = 'COMPLETED'
      ))
    ORDER BY j.available_at ASC, j.created_at ASC
    FOR UPDATE OF j SKIP LOCKED
    LIMIT 1
)
UPDATE ordered_jobs u
SET status = 'PROCESSING',
    worker_id = $1,
    lease_until = NOW() + $2::interval,
    lease_generation = lease_generation + 1,
    attempt = attempt + 1,
    started_at = COALESCE(started_at, NOW()),
    updated_at = NOW()
FROM target
WHERE u.id = target.id
RETURNING u.*;
Fase 3: Eksekusi Handler & Heartbeat (Execution Pipeline)
  1. Spawning Heartbeat Manager: Goroutine heartbeat internal dinyalakan secara asinkron setiap leaseDuration / 3 untuk memperpanjang lease_until di database.
  2. Panic Recovery Guard: Worker membungkus eksekusi handler dengan defer recover() untuk menangkap unhandled panic dan mencatat stack trace tanpa menghentikan worker process.
  3. Context Injection & Generics: Inject TraceID dan deadline context ke dalam handler. Deserialisasi payload JSON secara otomatis via RegisterTyped[T].
Fase 4: Penyelesaian & Promosi Urutan (Completion Pipeline)
  1. Eksekusi Sukses (err == nil): Status diperbarui menjadi COMPLETED. Promosi Atomik: Memperbarui status job Sequence N+1 dari BLOCKED menjadi PENDING.
  2. Kegagalan Transien (orderedjob.Retryable(err)): Mengaplikasikan exponential backoff + jitter. Status diperbarui menjadi RETRYING dengan available_at = NOW() + backoff.
  3. Error Permanen / Max Attempts Reached: Status diperbarui menjadi FAILED / DEAD_LETTERED. Perilaku promosi Sequence N+1 ditentukan oleh Ordering Mode:
    • strict: Chain tetap terhenti (BLOCKED).
    • skip-on-failure: Job Sequence N+1 otomatis dipromosikan ke PENDING.
    • dead-letter-and-continue: Job gagal dipindahkan ke DLQ dan Sequence N+1 dibuka.

4. Diagram Arsitektur & State Machine

Diagram Komponen & Alur Data
graph TD
    Client["Client Application"] -->|"1. Enqueue(req)"| DB[("Storage Database\n(PostgreSQL / SQL Server / Memory)")]

    subgraph Engine["OrderedJob Core Engine"]
        Listener["LISTEN Stream Listener\n(pg_notify / internal)"]
        Claimer["SKIP LOCKED / UPDLOCK Worker\n(Atomic Fencing Token)"]
        Heartbeat["Lease Heartbeat Loop\n(lease_generation + 1)"]
        Recovery["Stale Lease Recovery Worker\n(Cron Loop)"]
        Registry["Handler Registry & Panic Guard\n(Generics / Struct Validation)"]
        Webhook["Webhook Dispatcher\n(Slack / HTTP Callbacks)"]
    end

    DB -->|"2. NOTIFY event"| Listener
    Listener -->|"3. Wakeup trigger"| Claimer
    DB -->|"4. FOR UPDATE SKIP LOCKED"| Claimer
    Claimer -->|"5. Spawn Heartbeat"| Heartbeat
    Heartbeat -->|"Periodical Extend"| DB
    Claimer -->|"6. Dispatch Job"| Registry
    Registry -->|"7. Execute Logic"| Ext["External Services / APIs"]
    Registry -->|"8. Promote N+1 (COMPLETED)"| DB
    Registry -->|"9. Emit Event"| Webhook
    Recovery -->|"10. Reclaim Stale Locks"| DB
Diagram State Machine Pekerjaan
stateDiagram-v2
    [*] --> BLOCKED: Sequence > 1 (Predecessor Pending)
    [*] --> PENDING: Sequence = 1 / Auto Sequence

    BLOCKED --> PENDING: Predecessor Status = COMPLETED / Unblocked by Policy
    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 --> [*]

5. Penanganan Edge Cases & Resiliensi Sistem

Skenario Edge Case Potensi Masalah Solusi & Mekanisme Resiliensi orderedjob
Worker Mati Mendadak (Crash / Power Loss) Job menggantung di status PROCESSING selamanya. Stale Lease Recovery Loop: Memindai job PROCESSING dengan lease_until < NOW(), lalu memulihkan statusnya ke PENDING/RETRYING dan menaikkan lease_generation.
Worker Lambat (Network Lag / GC Pause) Worker lama terbangun dan mencoba menyelesaikan job yang sudah di-reclaim worker lain. Fencing Token (lease_generation): Kueri UPDATE status = 'COMPLETED' memverifikasi WHERE id = $1 AND lease_generation = $2. Jika generation berbeda, perubahan ditolak.
Klaim Bersamaan pada Chain Sama (Concurrent Enqueue) Penomoran urutan bertabrakan (race condition). Advisory Locking: pg_advisory_xact_lock (PG) / sp_getapplock (MSSQL) memastikan alokasi urutan otomatis (Sequence = 0) dilakukan secara serial per chain.
Unhandled Panic pada Handler Goroutine Worker goroutine mati mendadak dan aplikasi crash. Panic Recovery Guard: Menangkap panic via recover(), mencatat stack trace lengkap, dan menandai job sebagai FAILED/RETRYING dengan aman.
Enkui Melompati Urutan (Sequence Gap) Pengirim mengenkui Sequence 5 padahal Sequence 1..4 belum ada. Strict Gap Checking: Database menolak enkui yang melompati urutan dan mengembalikan ErrSequenceGap.
Duplikasi Enkui akibat Retry Client HTTP Job yang sama di-enqueue berulang kali. Idempotency Key Deduplication: Kueri SELECT ... WHERE idempotency_key = $1 mengembalikan objek Job yang sudah ada tanpa insert ganda.
Pemulihan Manual Job Gagal (DLQ) Chain terhenti akibat 1 job gagal. Bulk DLQ API & Web UI: Operator dapat memicu BulkReplayDLQ, BulkSkipDLQ, atau BulkPurgeDLQ dengan filter waktu, job type, dan error message.

6. Policy-Driven Execution & Dynamic Ordering

Pustaka orderedjob menyediakan tiga moda urutan eksekusi (ordering modes) yang dapat disesuaikan per engine atau per job request:

  1. OrderingModeStrict ("strict"): Mode default. Job Sequence N wajib menunggu hingga Sequence N-1 berstatus COMPLETED. Jika N-1 gagal (FAILED/DEAD_LETTERED), chain akan terhenti sampai diintervensi manual.
  2. OrderingModeSkipOnFailure ("skip-on-failure"): Jika job Sequence N-1 mengalami kegagalan terminal (DLQ), engine secara otomatis menandai kegagalan tersebut dan mempromosikan Sequence N ke status PENDING.
  3. OrderingModeDeadLetterAndContinue ("dead-letter-and-continue"): Job yang gagal dipindahkan ke DLQ untuk inspeksi pasca-mortem, namun urutan berikutnya tetap dibuka agar tidak menghambat aliran data global.
Dynamic Strategy Resolver

Anda dapat menentukan strategi urutan secara dinamis berdasarkan properti job request (misal per tenant_id atau job_type):

eng := orderedjob.New(repo,
    orderedjob.WithOrderingStrategy(func(req orderedjob.EnqueueRequest) string {
        if req.Type == "AuditLog" {
            return orderedjob.OrderingModeSkipOnFailure
        }
        return orderedjob.OrderingModeStrict
    }),
)

7. Worker Panic Isolation & Resiliency Guard

Untuk mencegah kegagalan fatal (unhandled panics) pada logika bisnis user yang dapat menghentikan worker pool, orderedjob menyertakan Panic Recovery Guard bawaan:

  • Pengisolasi Panic: Setiap eksekusi handler dibungkus dalam blok pemulihan isolasi.
  • Pencatatan Stacktrace: Menangkap stack trace lengkap dan mencatatnya ke log terstruktur (log/slog).
  • Opsi Recovery & Retry:
eng := orderedjob.New(repo,
    orderedjob.WithPanicHandler(func(ctx context.Context, job orderedjob.Job, panicVal any, stack []byte) error {
        // Custom logging atau alerting...
        return nil
    }),
    orderedjob.WithRetryOnPanic(true), // Mengizinkan retry jika terjadi panic
)

8. Arsitektur Observabilitas (Metrik & Tracing)

Standar Metrik OpenTelemetry & Prometheus

orderedjob mengimplementasikan pengumpulkan metrik tanpa overhead performa menggunakan operasi atomic:

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()
}
Propagasi Context Tracing (Distributed Tracing)

Trace ID disisipkan saat Enqueue dan secara otomatis dipropagasikan ke dalam context.Context worker handler:

// Menyisipkan Trace ID pada Client
ctx = orderedjob.WithTraceID(ctx, "trace-abc-12345")
eng.Enqueue(ctx, req)

// Membaca Trace ID di Worker Handler
orderedjob.RegisterTyped(eng, "ProcessOrder", func(ctx context.Context, job orderedjob.Job, payload OrderPayload) error {
    traceID := orderedjob.ExtractTraceID(ctx)
    // Log terintegrasi dengan Trace ID...
    return nil
})

9. Real-Time Web UI Architecture

Dashboard Web UI bawaan (ui/) dibangun dengan arsitektur modern tanpa dependensi berat:

  • Server-Sent Events (SSE): Streaming metrik dan perubahan status job secara real-time melalui endpoint /ui/events.
  • Search & Detail Modal: Pencarian job berdasarkan trace_id, chain_id, tenant_id, atau status.
  • Payload Inspector & Editor: Tampilan JSON payload terformat dengan sintaks yang rapi serta modal perbaikan payload secara interaktif.
  • Filtering & Sorting: Fitur pagination, penataan kolom, dan status filter antrean.
  • Auth Protection & Security: Dukungan bawaan untuk HTTP Basic Auth (WithBasicAuth), Bearer Token (WithTokenAuth), HS256 JWT Token (WithJWTAuth), serta custom HTTP middleware (WithAuthMiddleware).

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.

Advanced Operations: DLQ, Struct Validation & Webhooks

orderedjob menyediakan API dan fitur operasional tingkat lanjut untuk mengelola antrean job yang bermasalah (DLQ), validasi struct, dan notifikasi webhook:

Bulk Operations API (DLQ)

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

// Skip masal job di DLQ
skipped, err := eng.BulkSkipDLQ(ctx, orderedjob.JobFilter{
    CreatedBefore: time.Now().Add(-24 * time.Hour),
})

Struct Validation & Payload Editing

Pendaftaran handler mendukung pengujian skema struct atau validator kustom sebelum payload diproses:

orderedjob.RegisterTypedWithValidator(eng, "UpdateUser",
    func(ctx context.Context, job orderedjob.Job, payload UserPayload) error {
        return nil
    },
    func(payload UserPayload) error {
        if payload.Email == "" {
            return errors.New("email required")
        }
        return nil
    },
)

Event Listeners & Webhook Dispatcher

Integrasi callback untuk peristiwa penting seperti kegagalan job atau penumpukan DLQ:

eng := orderedjob.New(repo,
    orderedjob.WithWebhook(orderedjob.WebhookConfig{
        URL:     "https://hooks.slack.com/services/...",
        Secret:  "whsec_xxx",
        Timeout: 5 * time.Second,
    }),
)

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.