runnerq

README

RunnerQ (Go)

A robust, scalable activity queue and worker system for Go applications with pluggable storage backends.

Features

• Pluggable backend system - Interface-based storage abstraction; PostgreSQL is built-in
• Priority-based activity processing - Support for Critical, High, Normal, and Low priority levels
• Activity scheduling - Precise timestamp-based scheduling for future execution
• Intelligent retry mechanism - Built-in retry mechanism with exponential backoff
• Dead letter queue - Failed activities are moved to a dead letter queue for inspection
• Concurrent activity processing - Configurable number of concurrent workers (goroutines)
• Graceful shutdown - Proper shutdown handling with OS signal support
• Activity orchestration - Activities can execute other activities for complex workflows
• Comprehensive error handling - Retryable and non-retryable error types
• Activity metadata - Support for custom metadata on activities
• Built-in observability console - Real-time web UI for monitoring and managing activities
• Worker-level activity type filtering - Isolate workloads by restricting each engine to specific activity types
• Parent-await suspension - Optional SuspendOnAwait mode releases a parent's worker slot while it waits for child futures, eliminating the parent-blocking-on-children starvation pattern on recursive fan-out workloads
• Queue statistics - Monitoring capabilities and metrics collection

Storage Backends

Backend	Status	Use Case
PostgreSQL	Built-in	Default. Permanent persistence, SQL-based queries
Custom	Supported	Implement the Storage interface for your own backend

Installation

go get github.com/alob-mtc/runnerq-go

Quick Start

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/alob-mtc/runnerq-go"
	"github.com/alob-mtc/runnerq-go/storage/postgres"
)

// SendEmailHandler implements activity processing for emails.
type SendEmailHandler struct {
	runnerq.DefaultDeadLetterHandler
}

func (h *SendEmailHandler) ActivityType() string {
	return "send_email"
}

func (h *SendEmailHandler) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
	var data map[string]interface{}
	if err := json.Unmarshal(payload, &data); err != nil {
		return nil, runnerq.NewNonRetryError("Invalid payload format")
	}

	to, ok := data["to"].(string)
	if !ok || to == "" {
		return nil, runnerq.NewNonRetryError("Missing 'to' field")
	}

	fmt.Printf("Sending email to: %s\n", to)

	result, _ := json.Marshal(map[string]interface{}{
		"message": fmt.Sprintf("Email sent to %s", to),
		"status":  "delivered",
	})
	return result, nil
}

func main() {
	ctx := context.Background()

	backend, err := postgres.New(ctx, "postgres://localhost/mydb", "my_app")
	if err != nil {
		log.Fatal(err)
	}
	defer backend.Close()

	engine, err := runnerq.Builder().
		Backend(backend).
		QueueName("my_app").
		MaxWorkers(8).
		SchedulePollInterval(30 * time.Second).
		Build()
	if err != nil {
		log.Fatal(err)
	}

	engine.RegisterActivity("send_email", &SendEmailHandler{})

	executor := engine.GetActivityExecutor()

	// Execute an activity with custom options
	emailPayload, _ := json.Marshal(map[string]interface{}{
		"to":      "user@example.com",
		"subject": "Welcome!",
	})
	future, err := executor.Activity("send_email").
		Payload(emailPayload).
		MaxRetries(5).
		Timeout(10 * time.Minute).
		Execute(ctx)
	if err != nil {
		log.Fatal(err)
	}

	// Schedule an activity for future execution (10 seconds from now)
	scheduledPayload, _ := json.Marshal(map[string]interface{}{
		"to":      "user@example.com",
		"subject": "Reminder",
	})
	_, err = executor.Activity("send_email").
		Payload(scheduledPayload).
		MaxRetries(3).
		Timeout(5 * time.Minute).
		Delay(10 * time.Second).
		Execute(ctx)
	if err != nil {
		log.Fatal(err)
	}

	// Tag a cron-triggered activity (useful for observability filtering)
	cronPayload, _ := json.Marshal(map[string]interface{}{
		"job": "nightly_reconcile",
	})
	_, err = executor.Activity("reconcile_accounts").
		Payload(cronPayload).
		Metadata("source", "cron").
		Metadata("cron_job", "nightly_reconcile").
		Execute(ctx)
	if err != nil {
		log.Fatal(err)
	}

	// Execute an activity with default options
	simplePayload, _ := json.Marshal(map[string]interface{}{
		"to": "admin@example.com",
	})
	_, err = executor.Activity("send_email").
		Payload(simplePayload).
		Execute(ctx)
	if err != nil {
		log.Fatal(err)
	}

	// Wait for the email result in a goroutine
	go func() {
		resultCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
		defer cancel()

		result, err := future.GetResult(resultCtx)
		if err != nil {
			fmt.Printf("Failed to get result: %v\n", err)
			return
		}
		fmt.Printf("Email result: %s\n", string(result))
	}()

	// Start the worker engine (blocks until shutdown signal or context cancellation)
	if err := engine.Start(ctx); err != nil {
		log.Fatal(err)
	}
}

Builder Pattern API

RunnerQ provides a fluent builder pattern for both WorkerEngine configuration and activity execution.

WorkerEngine Builder

import (
	"time"

	"github.com/alob-mtc/runnerq-go"
	"github.com/alob-mtc/runnerq-go/storage/postgres"
)

// Basic configuration
engine, err := runnerq.Builder().
	Backend(backend).
	QueueName("my_app").
	MaxWorkers(8).
	SchedulePollInterval(30 * time.Second).
	Build()

// With custom metrics
engine, err := runnerq.Builder().
	Backend(backend).
	QueueName("my_app").
	MaxWorkers(8).
	Metrics(&PrometheusMetrics{}).
	Build()

// Restrict this engine to specific activity types (workload isolation)
engine, err := runnerq.Builder().
	Backend(backend).
	QueueName("my_app").
	ActivityTypes([]string{"send_email", "send_sms"}).
	Build()

Activity Builder

executor := engine.GetActivityExecutor()

// Fluent activity execution
payload, _ := json.Marshal(map[string]interface{}{
	"to":      "user@example.com",
	"subject": "Hello",
})
future, err := executor.Activity("send_email").
	Payload(payload).
	MaxRetries(5).
	Timeout(10 * time.Minute).
	Execute(ctx)

// Schedule activity for future execution
scheduledPayload, _ := json.Marshal(map[string]interface{}{"user_id": 123})
future, err := executor.Activity("send_reminder").
	Payload(scheduledPayload).
	Delay(1 * time.Hour).
	Execute(ctx)

// Attach metadata (for correlation, routing, or cron tagging)
future, err := executor.Activity("reconcile_accounts").
	Payload(payload).
	Metadata("source", "cron").
	MetadataMap(map[string]string{
		"cron_job": "nightly_reconcile",
		"tenant":   "acme",
	}).
	Execute(ctx)

// Simple activity with defaults
simplePayload, _ := json.Marshal(map[string]interface{}{"data": "example"})
future, err := executor.Activity("process_data").
	Payload(simplePayload).
	Execute(ctx)

Activity Types

Activity types in RunnerQ are simple strings that identify different types of activities.

Examples

"send_email"
"process_payment"
"provision_card"
"update_card_status"
"process_webhook_event"

"user.registration"
"email-notification"
"background_sync"

Custom Activity Handlers

Create custom activity handlers by implementing the ActivityHandler interface:

import (
	"encoding/json"
	"fmt"

	"github.com/alob-mtc/runnerq-go"
)

type PaymentHandler struct {
	runnerq.DefaultDeadLetterHandler
	// Add your dependencies here (database connections, external APIs, etc.)
}

func (h *PaymentHandler) ActivityType() string {
	return "process_payment"
}

func (h *PaymentHandler) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
	var data map[string]interface{}
	if err := json.Unmarshal(payload, &data); err != nil {
		return nil, runnerq.NewNonRetryError("Invalid payload format")
	}

	amount, _ := data["amount"].(float64)
	currency, _ := data["currency"].(string)
	if currency == "" {
		currency = "USD"
	}

	fmt.Printf("Processing payment: %.2f %s\n", amount, currency)

	if amount <= 0 {
		return nil, runnerq.NewNonRetryError("Invalid amount")
	}

	result, _ := json.Marshal(map[string]interface{}{
		"transaction_id": "txn_123456",
		"amount":         amount,
		"currency":       currency,
		"status":         "completed",
	})
	return result, nil
}

// Register the handler
engine.RegisterActivity("process_payment", &PaymentHandler{})

Activity Priority and Options

Activities can be configured with priority and execution parameters:

// High priority with custom retry and timeout settings
payload, _ := json.Marshal(map[string]interface{}{"to": "user@example.com"})
future, err := executor.Activity("send_email").
	Payload(payload).
	Priority(runnerq.PriorityCritical). // Highest priority
	MaxRetries(10).                      // Retry up to 10 times
	Timeout(15 * time.Minute).           // 15 minute timeout
	Execute(ctx)

// Default options (Normal priority, 3 retries, 300s timeout)
future, err := executor.Activity("send_email").
	Payload(payload).
	Execute(ctx)

Available priorities:

• runnerq.PriorityCritical - Highest priority (processed first)
• runnerq.PriorityHigh - High priority
• runnerq.PriorityNormal - Default priority
• runnerq.PriorityLow - Lowest priority

Getting Activity Results

Activities can return results that can be retrieved asynchronously:

future, err := executor.Activity("send_email").
	Payload(payload).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}

// Get the result (this will poll until the activity completes or ctx is cancelled)
resultCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()

result, err := future.GetResult(resultCtx)
if err != nil {
	log.Printf("Failed: %v", err)
	return
}

var emailResult struct {
	Message string `json:"message"`
	Status  string `json:"status"`
}
json.Unmarshal(result, &emailResult)
fmt.Printf("Email result: %+v\n", emailResult)

Activity Orchestration

Activities can execute other activities using the ActivityExecutor available in the ActivityContext. This enables powerful workflow orchestration:

type OrderProcessingHandler struct {
	runnerq.DefaultDeadLetterHandler
}

func (h *OrderProcessingHandler) ActivityType() string {
	return "process_order"
}

func (h *OrderProcessingHandler) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
	var data map[string]interface{}
	if err := json.Unmarshal(payload, &data); err != nil {
		return nil, runnerq.NewNonRetryError("Invalid payload")
	}

	orderID, _ := data["order_id"].(string)

	// Step 1: Validate payment using the fluent API
	paymentPayload, _ := json.Marshal(map[string]interface{}{"order_id": orderID})
	_, err := ctx.ActivityExecutor.Activity("validate_payment").
		Payload(paymentPayload).
		Priority(runnerq.PriorityHigh).
		MaxRetries(3).
		Timeout(2 * time.Minute).
		Execute(ctx.Ctx)
	if err != nil {
		return nil, runnerq.NewRetryError(fmt.Sprintf("Failed to enqueue payment validation: %v", err))
	}

	// Step 2: Update inventory
	inventoryPayload, _ := json.Marshal(map[string]interface{}{"order_id": orderID})
	_, err = ctx.ActivityExecutor.Activity("update_inventory").
		Payload(inventoryPayload).
		Execute(ctx.Ctx)
	if err != nil {
		return nil, runnerq.NewRetryError(fmt.Sprintf("Failed to enqueue inventory update: %v", err))
	}

	// Step 3: Schedule delivery notification for later
	notifPayload, _ := json.Marshal(map[string]interface{}{
		"order_id":       orderID,
		"customer_email": data["customer_email"],
	})
	_, err = ctx.ActivityExecutor.Activity("send_delivery_notification").
		Payload(notifPayload).
		Priority(runnerq.PriorityNormal).
		MaxRetries(5).
		Timeout(5 * time.Minute).
		Delay(1 * time.Hour).
		Execute(ctx.Ctx)
	if err != nil {
		return nil, runnerq.NewRetryError(fmt.Sprintf("Failed to schedule notification: %v", err))
	}

	result, _ := json.Marshal(map[string]interface{}{
		"order_id":        orderID,
		"status":          "processing",
		"steps_initiated": []string{"payment_validation", "inventory_update", "delivery_notification"},
	})
	return result, nil
}

Benefits of Activity Orchestration

• Modularity: Break complex workflows into smaller, reusable activities
• Reliability: Each sub-activity has its own retry logic and error handling
• Monitoring: Track progress of individual workflow steps
• Scalability: Sub-activities can be processed by different workers
• Flexibility: Different priority levels and timeouts for different steps
• Scheduling: Schedule activities for future execution
• Fluent API: Clean, readable activity execution with method chaining

Metrics and Monitoring

RunnerQ provides comprehensive metrics collection through the MetricsSink interface, allowing you to integrate with your preferred monitoring system.

Basic Metrics Implementation

import (
	"fmt"
	"time"

	"github.com/alob-mtc/runnerq-go"
)

type LoggingMetrics struct{}

func (m *LoggingMetrics) IncCounter(name string, value uint64) {
	fmt.Printf("METRIC: %s += %d\n", name, value)
}

func (m *LoggingMetrics) ObserveDuration(name string, dur time.Duration) {
	fmt.Printf("METRIC: %s = %v\n", name, dur)
}

// Use with WorkerEngine
engine, err := runnerq.Builder().
	Backend(backend).
	QueueName("my_app").
	Metrics(&LoggingMetrics{}).
	Build()

Available Metrics

The library automatically collects the following metrics:

• activity_completed - Number of activities completed successfully
• activity_retry - Number of activities that requested retry
• activity_failed_non_retry - Number of activities that failed permanently
• activity_timeout - Number of activities that timed out

No-op Metrics

If you don't need metrics collection, the built-in NoopMetrics is used by default:

var metrics runnerq.NoopMetrics

// These calls do nothing
metrics.IncCounter("activities_completed", 1)
metrics.ObserveDuration("activity_execution", 5*time.Second)

Advanced Features

Activity Scheduling

RunnerQ supports scheduling activities for future execution:

executor := engine.GetActivityExecutor()

// Schedule an activity to run in 1 hour
payload, _ := json.Marshal(map[string]interface{}{
	"user_id": 123,
	"message": "Don't forget!",
})
future, err := executor.Activity("send_reminder").
	Payload(payload).
	Delay(1 * time.Hour).
	Execute(ctx)

// Schedule for 2 hours from now
reportPayload, _ := json.Marshal(map[string]interface{}{"report_type": "monthly"})
future, err := executor.Activity("process_report").
	Payload(reportPayload).
	Delay(2 * time.Hour).
	Execute(ctx)

Parent-Await Suspension (SuspendOnAwait)

The problem. When an activity handler spawns child activities and calls fut.GetResult() to wait for them, the parent's goroutine sits blocked inside GetResult while still holding its worker slot. With recursive fan-out (a parent spawns several children, each of which is itself a parent that fans out further), worker slots fill up with suspended parents and the leaf activities they're waiting on can't dequeue. The result is starvation: queue depth climbs, parent awaitCtx deadlines fire, activities fail and retry, and dead-letter grows.

The fix. With Builder.SuspendOnAwait(true) the engine switches to a semaphore-based dispatcher. When a handler calls GetResult, the parent's slot is released for the duration of the wait and reacquired before the call returns. The handler's goroutine stays alive — only the slot moves. Other activities, including the children the parent is waiting on, can dispatch onto that slot immediately.

engine, err := runnerq.Builder().
    Backend(backend).
    QueueName("my_app").
    MaxWorkers(20).
    SuspendOnAwait(true).
    Build()

No handler changes required. The release/reacquire happens transparently inside GetResult. Outside a handler context (e.g. in main()) it's a no-op — behaviour is identical to before.

Reserving slots for leaf activities

When many parents wake up at once they all race to reacquire slots. In the worst case every freed slot is immediately retaken by another waking parent and no leaf can dispatch — a wake-up deadlock. Mitigate by reserving a fraction of slots for designated leaf activity types:

engine, err := runnerq.Builder().
    Backend(backend).
    QueueName("my_app").
    MaxWorkers(20).
    SuspendOnAwait(true).
    ReserveSlotsForLeaves(5, []string{"send_email", "fetch_data", "audit_log"}).
    Build()

This carves the 20 slots into two independent pools:

Pool	Slots	Dequeues
Primary	15 (= MaxWorkers − reserved)	Any activity type (subject to ActivityTypes filter, if set)
Leaf reservation	5	Only the listed leaf types

A separate dispatcher goroutine drains the leaf pool. Leaves can never be starved by waking parents.

Concurrency invariants

MaxWorkers(N) means different things in the two modes:

Mode	Goroutine count	Bound on actively-computing handlers
Default	Fixed = MaxWorkers	The goroutine count itself
SuspendOnAwait(true)	Unbounded (1 dispatcher + 1 per active activity + 1 per suspended parent)	The semaphore buffer capacity (MaxWorkers)

In suspend mode, suspended parents hold zero slots but their goroutines stay alive in GetResult's poll loop. With deep fan-out you can have hundreds of suspended goroutines while only MaxWorkers are actively computing.

Operational notes

• Activity timeouts and leases are unchanged. The handler's timeoutCtx still fires at the activity's declared timeout. The dequeue lease is set to max(defaultLeaseMS, (timeout_seconds+10)*1000) and covers the full execution window — including suspended waits — so no separate lease-extension loop is needed.
• Crash recovery. A process crash loses in-memory parent goroutines just as today; the reaper requeues the row and the new worker re-runs the handler from scratch. Plan handlers for idempotent re-execution.
• GetResult poll storm. GetResult currently polls every 100ms. With many concurrent suspended parents this multiplies the query rate on runnerq_results. Size your DB connection pool to absorb it (rule of thumb: a few connections beyond MaxWorkers plus headroom proportional to expected concurrent waiters). Event-driven futures via the existing event hub are a planned follow-up.
• active_workers KPI semantics. In suspend mode every dispatcher dequeues with the same workerID, so the cluster-wide COUNT(DISTINCT current_worker_id) doesn't reflect actual slot utilization. Treat it as a per-process activity indicator, not per-slot.

Stress-testing the pattern

The examples/perf/suspend_stress example reproduces the parent-blocking starvation pattern at configurable scale and reports drain time vs DLQ growth. Useful for tuning MaxWorkers and ReserveSlotsForLeaves for your specific workload shape:

# Baseline: classical fixed-goroutine pool. Expect stall + DLQ growth.
SUSPEND=0 ROOTS=100 WORKERS=20 go run ./examples/perf/suspend_stress

# Suspend on, no reservation: usually drains; small wake-up risk.
SUSPEND=1 ROOTS=100 WORKERS=20 go run ./examples/perf/suspend_stress

# Suspend on with reservation: clean drain.
SUSPEND=1 LEAVES_RESERVED=4 ROOTS=100 WORKERS=20 go run ./examples/perf/suspend_stress

Workload Isolation (Activity Type Filtering)

By default every worker engine dequeues all activity types from the queue. When you need to isolate workloads — for example, keeping slow report-generation jobs from starving latency-sensitive email sends — you can restrict each engine to specific activity types with .ActivityTypes():

import (
	"github.com/alob-mtc/runnerq-go"
	"github.com/alob-mtc/runnerq-go/storage/postgres"
)

backend, err := postgres.New(ctx, "postgres://localhost/runnerq", "my_app")

// Node 1 — only processes email-related activities
emailEngine, err := runnerq.Builder().
	Backend(backend).
	ActivityTypes([]string{"send_email", "send_sms"}).
	MaxWorkers(4).
	Build()
emailEngine.RegisterActivity("send_email", &SendEmailHandler{})
emailEngine.RegisterActivity("send_sms", &SendSmsHandler{})

// Node 2 — only processes trades
tradeEngine, err := runnerq.Builder().
	Backend(backend).
	ActivityTypes([]string{"execute_trade"}).
	MaxWorkers(8).
	Build()
tradeEngine.RegisterActivity("execute_trade", &TradeHandler{})

// Node 3 — catch-all, processes anything not claimed above
catchallEngine, err := runnerq.Builder().
	Backend(backend).
	MaxWorkers(2).
	Build()
// Register all handlers on the catch-all node

All engines share the same backend and queue. Each engine's Dequeue() only claims activities matching its declared types; an engine with no filter acts as a catch-all.

Startup validation: If ActivityTypes is set and any listed type does not have a registered handler, the engine panics at Start() with a clear error message.

See examples/advanced/activity_filtering/ for a complete working example.

Pluggable Storage Backends

RunnerQ uses an interface-based storage abstraction that allows you to swap out the persistence layer.

PostgreSQL Backend

import "github.com/alob-mtc/runnerq-go/storage/postgres"

backend, err := postgres.New(ctx, "postgres://user:password@localhost/runnerq", "my_queue")
if err != nil {
	log.Fatal(err)
}
defer backend.Close()

engine, err := runnerq.Builder().
	Backend(backend).
	MaxWorkers(8).
	Build()

PostgreSQL Backend Features:

• Permanent Persistence - Activities stored indefinitely (no TTL expiration)
• Multi-node Safe - Uses FOR UPDATE SKIP LOCKED for concurrent job claiming
• Cross-process Events - PostgreSQL LISTEN/NOTIFY for real-time event streaming
• Atomic Idempotency - Separate table with INSERT ... ON CONFLICT for race-safe key claiming
• History Preservation - Never deletes activity records

Schema Tables Created:

• runnerq_activities - Main activity storage
• runnerq_events - Event history timeline
• runnerq_results - Activity execution results
• runnerq_idempotency - Idempotency key mapping

See examples/basic/ for a complete working example, and examples/advanced/activity_filtering/ for workload isolation with multiple engines.

Implementing a Custom Backend

You can implement your own backend by implementing the Storage interface (which combines QueueStorage and InspectionStorage):

import (
	"context"
	"encoding/json"
	"time"

	"github.com/google/uuid"

	"github.com/alob-mtc/runnerq-go/storage"
)

type MyCustomBackend struct {
	// Your backend state (connection pool, config, etc.)
}

// QueueStorage methods

func (b *MyCustomBackend) Enqueue(ctx context.Context, activity storage.QueuedActivity) error {
	// Implement activity enqueuing
	panic("not implemented")
}

func (b *MyCustomBackend) Dequeue(ctx context.Context, workerID string, timeout time.Duration, activityTypes []string) (*storage.QueuedActivity, error) {
	// Implement activity claiming.
	// When activityTypes is non-nil, only claim matching types.
	panic("not implemented")
}

func (b *MyCustomBackend) AckSuccess(ctx context.Context, activityID uuid.UUID, result json.RawMessage, workerID string) error {
	// Mark activity as completed
	panic("not implemented")
}

func (b *MyCustomBackend) AckFailure(ctx context.Context, activityID uuid.UUID, failure storage.FailureKind, workerID string) (bool, error) {
	// Handle activity failure (retry or dead-letter). Returns true if dead-lettered.
	panic("not implemented")
}

// ... implement remaining QueueStorage methods:
//   ProcessScheduled, RequeueExpired, ExtendLease,
//   StoreResult, GetResult, CheckIdempotency, SchedulesNatively

// InspectionStorage methods

func (b *MyCustomBackend) Stats(ctx context.Context) (*storage.QueueStats, error) {
	panic("not implemented")
}

func (b *MyCustomBackend) ListPending(ctx context.Context, offset, limit int) ([]storage.ActivitySnapshot, error) {
	panic("not implemented")
}

// ... implement remaining InspectionStorage methods:
//   ListProcessing, ListScheduled, ListCompleted, ListDeadLetter,
//   GetActivity, GetActivityEvents, EventStream

// Use your custom backend
engine, err := runnerq.Builder().
	Backend(&MyCustomBackend{}).
	MaxWorkers(8).
	Build()

Storage Interface Reference

The storage abstraction consists of two interfaces:

QueueStorage - Core queue operations:

• Enqueue() - Add activity to the queue
• Dequeue() - Claim an activity for processing (PostgreSQL picks up due scheduled/retrying activities directly here)
• AckSuccess() - Mark activity as completed
• AckFailure() - Handle activity failure (retry or dead-letter)
• ProcessScheduled() - Move due scheduled activities to ready queue (PostgreSQL returns 0, nil as it handles this natively)
• RequeueExpired() - Reclaim activities with expired leases
• ExtendLease() - Extend activity processing lease
• StoreResult() / GetResult() - Activity result storage
• CheckIdempotency() - Idempotency key handling
• SchedulesNatively() - Whether the backend handles scheduling in Dequeue() (skips the polling loop if true)

InspectionStorage - Observability operations:

• Stats() - Get queue statistics
• ListPending() / ListProcessing() / ListScheduled() / ListCompleted() - List activities by status
• ListDeadLetter() - List dead-lettered activities
• GetActivity() - Get specific activity details
• GetActivityEvents() - Get activity lifecycle events
• EventStream() - Stream real-time events (multiplexed via fan-out hub in QueueInspector — one connection serves all SSE subscribers)

Graceful Shutdown

The worker engine supports graceful shutdown via OS signals (SIGINT, SIGTERM) or context cancellation:

engine, err := runnerq.Builder().
	Backend(backend).
	MaxWorkers(8).
	Build()

// Option 1: Start in background, stop programmatically
go func() {
	if err := engine.Start(ctx); err != nil {
		log.Printf("Engine error: %v", err)
	}
}()

time.Sleep(10 * time.Second)
engine.Stop() // Graceful shutdown

// Option 2: Start with cancellable context
ctx, cancel := context.WithCancel(context.Background())
go func() {
	time.Sleep(10 * time.Second)
	cancel() // Triggers shutdown
}()
engine.Start(ctx) // Blocks until cancelled

Activity Context and Metadata

Access rich context information in your activity handlers:

func (h *MyHandler) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
	// Access activity metadata
	fmt.Printf("Processing activity %s of type %s\n", ctx.ActivityID, ctx.ActivityType)
	fmt.Printf("This is retry attempt #%d\n", ctx.RetryCount)

	// Check for context cancellation / deadline
	if ctx.Ctx.Err() != nil {
		return nil, runnerq.NewNonRetryError("Activity was cancelled")
	}

	// Access custom metadata
	if correlationID, ok := ctx.Metadata["correlation_id"]; ok {
		fmt.Printf("Correlation ID: %s\n", correlationID)
	}

	// Execute sub-activities via the embedded executor
	subPayload, _ := json.Marshal(map[string]interface{}{"parent": ctx.ActivityID.String()})
	_, err := ctx.ActivityExecutor.Activity("child_task").
		Payload(subPayload).
		Execute(ctx.Ctx)

	result, _ := json.Marshal(map[string]interface{}{"status": "processed"})
	return result, nil
}

Queue Statistics

Monitor queue performance and health using the inspector:

import (
	"github.com/alob-mtc/runnerq-go/observability"
)

inspector := observability.NewQueueInspector(backend).WithMaxWorkers(engine.MaxConcurrentActivities())

stats, err := inspector.Stats(ctx)
if err != nil {
	log.Fatal(err)
}

fmt.Println("Queue stats:")
fmt.Printf("  Pending activities: %d\n", stats.PendingActivities)
fmt.Printf("  Processing activities: %d\n", stats.ProcessingActivities)
fmt.Printf("  Scheduled activities: %d\n", stats.ScheduledActivities)
fmt.Printf("  Dead letter queue size: %d\n", stats.DeadLetterActivities)
fmt.Println("Priority distribution:")
fmt.Printf("  Critical: %d\n", stats.CriticalPriority)
fmt.Printf("  High: %d\n", stats.HighPriority)
fmt.Printf("  Normal: %d\n", stats.NormalPriority)
fmt.Printf("  Low: %d\n", stats.LowPriority)

For a visual dashboard with real-time updates, see the Observability Console section.

Error Handling

The library provides comprehensive error handling with clear separation between retryable and non-retryable errors.

Activity Handler Results

func (h *MyHandler) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
	var data struct {
		ID    string `json:"id"`
		Value string `json:"value"`
	}
	if err := json.Unmarshal(payload, &data); err != nil {
		return nil, runnerq.NewNonRetryError("Invalid data format")
	}

	if data.ID == "" {
		return nil, runnerq.NewNonRetryError("Missing required field: id")
	}

	// Perform operation that might temporarily fail
	result, err := externalAPICall(data)
	if err != nil {
		return nil, runnerq.NewRetryError(fmt.Sprintf("API call failed: %v", err))
	}

	resultJSON, _ := json.Marshal(map[string]interface{}{"result": result})
	return resultJSON, nil
}

Error Types:

• runnerq.NewRetryError(message) - Will be retried with exponential backoff
• runnerq.NewNonRetryError(message) - Will not be retried
• Any error returned that is not an *ActivityError is treated as retryable

Dead Letter Callback

When an activity exhausts all retries, it moves to the dead letter queue. Handle this by implementing OnDeadLetter (or embed DefaultDeadLetterHandler for a no-op):

type MyHandler struct{}

func (h *MyHandler) ActivityType() string { return "my_activity" }

func (h *MyHandler) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
	return nil, nil
}

func (h *MyHandler) OnDeadLetter(ctx runnerq.ActivityContext, payload json.RawMessage, errorMsg string) {
	fmt.Printf("Activity %s dead-lettered: %s\n", ctx.ActivityID, errorMsg)
	// Use for cleanup, notifications, or logging
}

The DefaultDeadLetterHandler provides an empty OnDeadLetter implementation — embed it in your handler struct if you don't need dead letter handling.

Worker Engine Errors

executor := engine.GetActivityExecutor()

payload, _ := json.Marshal(map[string]interface{}{"id": "123", "value": "test"})
future, err := executor.Activity("my_activity").
	Payload(payload).
	Execute(ctx)
if err != nil {
	fmt.Printf("Failed to enqueue activity: %v\n", err)
	return
}

resultCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()

result, err := future.GetResult(resultCtx)
if err != nil {
	if we, ok := runnerq.IsWorkerError(err); ok {
		fmt.Printf("Worker error (retryable=%v): %v\n", we.IsRetryable(), we)
	} else {
		fmt.Printf("Error: %v\n", err)
	}
	return
}
fmt.Printf("Activity completed: %s\n", string(result))

Error Recovery Patterns

func (h *ResilientHandler) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
	switch {
	case ctx.RetryCount <= 2:
		// First few attempts: retry on any error
		return h.processWithRetry(payload)
	case ctx.RetryCount <= 5:
		// Middle attempts: more conservative
		return h.processConservative(payload)
	default:
		// Final attempts: only retry on specific errors
		return h.processFinalAttempt(payload)
	}
}

Default Values

When using the builder pattern, sensible defaults are provided:

// Uses these defaults:
// - queue_name: "default"
// - max_workers: 10
// - schedule_poll_interval: 5 seconds
// - lease_ms: 60000 (60s)
// - reaper_interval: 5 seconds
// - reaper_batch_size: 100
engine, err := runnerq.Builder().
	Backend(backend).
	Build()

Observability Console

RunnerQ includes a built-in web-based observability console for monitoring and managing your activity queues in real-time.

Features

• Real-time Updates - Server-Sent Events (SSE) for instant activity updates
• Live Statistics - Monitor queue health with processing, pending, scheduled, and dead-letter counts
• Priority Distribution - See activity breakdown by priority level (Critical, High, Normal, Low)
• Activity Management - Browse and search activities across all queues (pending, processing, scheduled, completed, dead-letter)
• Activity Results - View execution results and outputs for completed activities
• Event Timeline - Detailed activity lifecycle events with multiple view modes
• Zero Setup - Embedded HTML, no build tools or npm required

SSE Fan-Out Architecture

The real-time event stream uses a fan-out hub inside QueueInspector so that all SSE subscribers share a single backend connection, regardless of how many clients are connected.

                        ┌──────────────┐
  Backend.EventStream() │  notifyHub   │──► SSE client 1
  (1 connection)  ─────►│  (inspector) │──► SSE client 2
                        │   broadcast  │──► SSE client 3
                        └──────────────┘
                         lazy start/stop

How it works:

• The first call to SubscribeEvents() lazily starts a single backend EventStream listener (e.g. one PostgreSQL LISTEN connection).
• Each additional subscriber gets its own buffered channel fed by non-blocking broadcast. If a subscriber's channel is full (slow client), events are dropped for that subscriber only — other subscribers are unaffected.
• When the last subscriber disconnects, the backend connection is released automatically.
• The HTTP layer enforces a concurrent SSE connection limit (default: 64). Requests beyond this limit receive 503 Service Unavailable.

This design means 100 SSE clients consume 1 database connection instead of 100, eliminating connection pool exhaustion as a denial-of-service vector.

Custom backends get fan-out for free — the hub operates on the EventStream() channel returned by any InspectionStorage implementation, so there is nothing backend-specific to implement.

Quick Start

import (
	"net/http"

	"github.com/alob-mtc/runnerq-go"
	"github.com/alob-mtc/runnerq-go/observability"
	"github.com/alob-mtc/runnerq-go/observability/ui"
	"github.com/alob-mtc/runnerq-go/storage/postgres"
)

backend, _ := postgres.New(ctx, "postgres://localhost/mydb", "my_app")

engine, _ := runnerq.Builder().
	Backend(backend).
	QueueName("my_app").
	Build()

inspector := observability.NewQueueInspector(backend).WithMaxWorkers(engine.MaxConcurrentActivities())

// Mount the console — serves the UI and API including SSE
mux := http.NewServeMux()
mux.Handle("/console/", http.StripPrefix("/console", ui.RunnerQUI(inspector)))

fmt.Println("RunnerQ Console: http://localhost:8081/console/")
http.ListenAndServe(":8081", mux)

Integration with Existing Apps

Easily integrate the console into your existing HTTP server:

mux := http.NewServeMux()

// Your existing routes
mux.HandleFunc("/api/users", listUsers)
mux.HandleFunc("/api/posts", listPosts)

// Add the RunnerQ console
mux.Handle("/console/", http.StripPrefix("/console", ui.RunnerQUI(inspector)))

http.ListenAndServe(":8080", mux)

API-Only Mode

If you prefer to build a custom UI, serve just the API:

mux.Handle("/api/observability/", http.StripPrefix("/api/observability", ui.ObservabilityAPI(inspector)))

Example

See the complete examples:

# Start PostgreSQL
docker run -d --name runnerq-postgres \
    -e POSTGRES_PASSWORD=runnerq \
    -e POSTGRES_DB=runnerq \
    -p 5432:5432 \
    postgres:16

# Run the console example
export DATABASE_URL="postgres://postgres:runnerq@localhost:5432/runnerq"
go run ./examples/observability/console_ui

# Open http://localhost:8081/console/

License

This project is licensed under the MIT License - see the LICENSE file for details.

Documentation

Overview

Package runnerq provides a durable activity queue and worker system for Go.

Features

• Priority-based activity processing (Critical, High, Normal, Low)
• Activity scheduling with precise timestamp-based scheduling
• Intelligent retry mechanism with exponential backoff
• Dead letter queue handling for activities exceeding retry limits
• Concurrent activity processing with configurable worker pools
• Graceful shutdown with proper cleanup
• Activity orchestration enabling activities to execute other activities
• Comprehensive error handling with retryable and non-retryable types
• Pluggable storage backends (PostgreSQL built-in)
• Worker-level activity type filtering
• Queue statistics and monitoring
• Web-based observability console

Quick Start

backend, _ := postgres.New(ctx, "postgres://localhost/mydb", "my_app")
engine, _ := runnerq.Builder().
    Backend(backend).
    QueueName("my_app").
    MaxWorkers(8).
    Build()

engine.RegisterActivity("send_email", &SendEmailHandler{})
engine.Start(ctx)

Index

• Constants
• type ActivityBuilder
  • func (b *ActivityBuilder) AsRoot() *ActivityBuilder
  • func (b *ActivityBuilder) Delay(d time.Duration) *ActivityBuilder
  • func (b *ActivityBuilder) Execute(ctx context.Context) (*ActivityFuture, error)
  • func (b *ActivityBuilder) IdempotencyKeyOption(key string, behavior OnDuplicate) *ActivityBuilder
  • func (b *ActivityBuilder) MaxRetries(retries uint32) *ActivityBuilder
  • func (b *ActivityBuilder) MaxRetryDelay(d time.Duration) *ActivityBuilder
  • func (b *ActivityBuilder) Metadata(key, value string) *ActivityBuilder
  • func (b *ActivityBuilder) MetadataMap(metadata map[string]string) *ActivityBuilder
  • func (b *ActivityBuilder) Payload(payload json.RawMessage) *ActivityBuilder
  • func (b *ActivityBuilder) Priority(p ActivityPriority) *ActivityBuilder
  • func (b *ActivityBuilder) Timeout(d time.Duration) *ActivityBuilder
• type ActivityContext
• type ActivityError
  • func NewNonRetryError(msg string) *ActivityError
  • func NewRetryError(msg string) *ActivityError
  • func (e *ActivityError) Error() string
  • func (e *ActivityError) IsRetryable() bool
• type ActivityExecutor
• type ActivityFuture
  • func (f *ActivityFuture) GetResult(ctx context.Context) (json.RawMessage, error)
• type ActivityHandler
• type ActivityOption
• type ActivityPriority
  • func (p ActivityPriority) MarshalJSON() ([]byte, error)
  • func (p ActivityPriority) String() string
  • func (p *ActivityPriority) UnmarshalJSON(data []byte) error
• type ActivityStatus
• type DefaultDeadLetterHandler
  • func (DefaultDeadLetterHandler) OnDeadLetter(_ ActivityContext, _ json.RawMessage, _ string)
• type IdempotencyConfig
• type MetricsSink
• type NoopMetrics
  • func (NoopMetrics) IncCounter(_ string, _ uint64)
  • func (NoopMetrics) ObserveDuration(_ string, _ time.Duration)
• type OnDuplicate
• type ResultState
• type RetryableError
• type WorkerConfig
  • func DefaultWorkerConfig() WorkerConfig
• type WorkerEngine
  • func NewWorkerEngineWithBackend(backend storage.Storage, config WorkerConfig) *WorkerEngine
  • func (e *WorkerEngine) Backend() storage.Storage
  • func (e *WorkerEngine) GetActivityExecutor() ActivityExecutor
  • func (e *WorkerEngine) MaxConcurrentActivities() int
  • func (e *WorkerEngine) RegisterActivity(activityType string, handler ActivityHandler)
  • func (e *WorkerEngine) SetMetrics(sink MetricsSink)
  • func (e *WorkerEngine) Start(ctx context.Context) error
  • func (e *WorkerEngine) Stop()
• type WorkerEngineBuilder
  • func Builder() *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) ActivityTypes(types []string) *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) Backend(backend storage.Storage) *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) Build() (*WorkerEngine, error)
  • func (b *WorkerEngineBuilder) MaxWorkers(max int) *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) Metrics(sink MetricsSink) *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) QueueName(name string) *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) ReserveSlotsForLeaves(n int, leafTypes []string) *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) SchedulePollInterval(interval time.Duration) *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) ShutdownGrace(d time.Duration) *WorkerEngineBuilder
  • func (b *WorkerEngineBuilder) SuspendOnAwait(enabled bool) *WorkerEngineBuilder
• type WorkerEngineWrapper
  • func (w *WorkerEngineWrapper) Activity(activityType string) *ActivityBuilder
• type WorkerError
  • func IsWorkerError(err error) (*WorkerError, bool)
  • func WorkerErrorFromStorage(err error) *WorkerError
  • func (e *WorkerError) Error() string
  • func (e *WorkerError) IsRetryable() bool
  • func (e *WorkerError) Unwrap() error
• type WorkerErrorKind

Constants

const DefaultMaxActivityDepth uint16 = 32

DefaultMaxActivityDepth is the default cap when MaxActivityDepth is unset.

Types

type ActivityBuilder

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

ActivityBuilder builds and executes activities with fluent configuration.

func (*ActivityBuilder) AsRoot

func (b *ActivityBuilder) AsRoot() *ActivityBuilder

AsRoot detaches this spawn from its parent's lineage. The resulting activity becomes a root (no parent, root is itself, depth 0) regardless of the surrounding handler context. Use for fire-and-forget side jobs whose lifecycle is logically independent of the parent (audit logs, async telemetry, etc.).

func (*ActivityBuilder) Delay

func (b *ActivityBuilder) Delay(d time.Duration) *ActivityBuilder

Delay sets the delay before execution.

func (*ActivityBuilder) Execute

func (b *ActivityBuilder) Execute(ctx context.Context) (*ActivityFuture, error)

Execute enqueues the activity and returns an ActivityFuture.

func (*ActivityBuilder) IdempotencyKeyOption

func (b *ActivityBuilder) IdempotencyKeyOption(key string, behavior OnDuplicate) *ActivityBuilder

IdempotencyKeyOption sets the idempotency key and behavior for duplicate detection.

func (*ActivityBuilder) MaxRetries

func (b *ActivityBuilder) MaxRetries(retries uint32) *ActivityBuilder

MaxRetries sets the maximum number of retry attempts (0 for unlimited).

func (*ActivityBuilder) MaxRetryDelay

func (b *ActivityBuilder) MaxRetryDelay(d time.Duration) *ActivityBuilder

MaxRetryDelay caps the exponential backoff delay between retries. Defaults to 1 hour if not set. Non-positive durations are ignored. Values below 1 second are rounded up to 1 second.

func (*ActivityBuilder) Metadata

func (b *ActivityBuilder) Metadata(key, value string) *ActivityBuilder

Metadata sets a single metadata key/value pair on the activity.

func (*ActivityBuilder) MetadataMap

func (b *ActivityBuilder) MetadataMap(metadata map[string]string) *ActivityBuilder

MetadataMap sets multiple metadata fields on the activity. Values are merged with existing metadata and overwrite duplicate keys.

func (*ActivityBuilder) Payload

func (b *ActivityBuilder) Payload(payload json.RawMessage) *ActivityBuilder

Payload sets the JSON payload for the activity.

func (*ActivityBuilder) Priority

func (b *ActivityBuilder) Priority(p ActivityPriority) *ActivityBuilder

Priority sets the priority level.

func (*ActivityBuilder) Timeout

func (b *ActivityBuilder) Timeout(d time.Duration) *ActivityBuilder

Timeout sets the maximum execution time.

type ActivityContext

type ActivityContext struct {
	// ActivityID is the unique identifier for this activity instance.
	ActivityID uuid.UUID

	// ActivityType is the type of activity being executed.
	ActivityType string

	// RetryCount is the current retry attempt number (0 for first execution).
	RetryCount uint32

	// Metadata is custom metadata associated with the activity.
	Metadata map[string]string

	// Ctx is the Go context for cancellation and deadline propagation.
	Ctx context.Context

	// ActivityExecutor allows executing other activities from within a handler.
	// Spawns made through this executor are tagged as children of this activity.
	ActivityExecutor ActivityExecutor

	// ParentActivityID is the direct parent of this activity, if any.
	ParentActivityID *uuid.UUID

	// RootActivityID is the root of the lineage tree this activity belongs to.
	// For root activities, RootActivityID equals ActivityID.
	RootActivityID uuid.UUID

	// Depth is the lineage depth of this activity (0 for roots).
	Depth uint16
}

ActivityContext is provided to activity handlers during execution.

type ActivityError

type ActivityError struct {
	Retryable bool
	Message   string
}

ActivityError represents an error from activity handler execution. Retryable indicates whether the activity should be retried.

func NewNonRetryError

func NewNonRetryError(msg string) *ActivityError

NewNonRetryError creates an ActivityError that should not be retried.

func NewRetryError

func NewRetryError(msg string) *ActivityError

NewRetryError creates an ActivityError that triggers a retry.

func (*ActivityError) Error

func (e *ActivityError) Error() string

func (*ActivityError) IsRetryable

func (e *ActivityError) IsRetryable() bool

IsRetryable implements the RetryableError interface.

type ActivityExecutor

type ActivityExecutor interface {
	Activity(activityType string) *ActivityBuilder
}

ActivityExecutor allows executing activities, enabling orchestration.

type ActivityFuture

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

ActivityFuture represents a pending activity result that can be awaited.

func (*ActivityFuture) GetResult

func (f *ActivityFuture) GetResult(ctx context.Context) (json.RawMessage, error)

GetResult waits for and returns the completed activity result. It polls every 100ms. The caller should use context for timeout control.

When called from inside an activity handler with SuspendOnAwait enabled, the host activity's worker slot is released for the duration of the wait and reacquired before returning. This prevents the parent-blocking-on- children starvation that pins worker slots on recursive fan-out workloads — see WorkerConfig.SuspendOnAwait. Outside that context the release/reacquire is a no-op and behaviour is identical to before.

type ActivityHandler

type ActivityHandler interface {
	// ActivityType returns the activity type string this handler processes.
	ActivityType() string

	// Handle processes the activity with the given payload and context.
	// Returns:
	//   (result, nil)       - completed successfully, result may be nil
	//   (nil, RetryError)   - failed but should be retried
	//   (nil, NonRetryError)- failed permanently
	Handle(ctx ActivityContext, payload json.RawMessage) (json.RawMessage, error)

	// OnDeadLetter is called when an activity enters the dead letter state.
	// The default behavior (if not overridden by embedding DefaultDeadLetterHandler) is a no-op.
	OnDeadLetter(ctx ActivityContext, payload json.RawMessage, errorMsg string)
}

ActivityHandler is the interface that all activity handlers must implement. Implementations should be safe for concurrent use.

type ActivityOption

type ActivityOption struct {
	Priority             *ActivityPriority
	MaxRetries           uint32
	TimeoutSeconds       uint64
	MaxRetryDelaySeconds *uint64
	DelaySeconds         *uint64
	IdempotencyKey       *IdempotencyConfig
	Metadata             map[string]string
}

ActivityOption configures an activity's execution parameters.

type ActivityPriority

type ActivityPriority int

ActivityPriority determines the order of execution. Higher priority activities are processed first.

const (
	PriorityLow      ActivityPriority = 1
	PriorityNormal   ActivityPriority = 2
	PriorityHigh     ActivityPriority = 3
	PriorityCritical ActivityPriority = 4
)

func (ActivityPriority) MarshalJSON

func (p ActivityPriority) MarshalJSON() ([]byte, error)

func (ActivityPriority) String

func (p ActivityPriority) String() string

func (*ActivityPriority) UnmarshalJSON

func (p *ActivityPriority) UnmarshalJSON(data []byte) error

type ActivityStatus

type ActivityStatus string

ActivityStatus tracks the activity lifecycle.

const (
	StatusPending    ActivityStatus = "Pending"
	StatusRunning    ActivityStatus = "Running"
	StatusCompleted  ActivityStatus = "Completed"
	StatusFailed     ActivityStatus = "Failed"
	StatusRetrying   ActivityStatus = "Retrying"
	StatusDeadLetter ActivityStatus = "DeadLetter"
)

type DefaultDeadLetterHandler

type DefaultDeadLetterHandler struct{}

DefaultDeadLetterHandler provides a no-op OnDeadLetter implementation. Embed this in your handler struct if you don't need dead letter handling.

func (DefaultDeadLetterHandler) OnDeadLetter

func (DefaultDeadLetterHandler) OnDeadLetter(_ ActivityContext, _ json.RawMessage, _ string)

OnDeadLetter is a no-op implementation.

type IdempotencyConfig

type IdempotencyConfig struct {
	Key      string
	Behavior OnDuplicate
}

IdempotencyConfig holds idempotency key and its behavior.

type MetricsSink

type MetricsSink interface {
	IncCounter(name string, value uint64)
	ObserveDuration(name string, dur time.Duration)
}

MetricsSink allows collecting metrics about activity processing. Implement this interface to integrate with your preferred metrics system.

type NoopMetrics

type NoopMetrics struct{}

NoopMetrics is the default metrics sink that discards all metrics.

func (NoopMetrics) IncCounter

func (NoopMetrics) IncCounter(_ string, _ uint64)

func (NoopMetrics) ObserveDuration

func (NoopMetrics) ObserveDuration(_ string, _ time.Duration)

type OnDuplicate

type OnDuplicate int

OnDuplicate defines behavior when an activity with the same idempotency key exists.

const (
	// AllowReuse always creates a new activity, updating the idempotency record.
	AllowReuse OnDuplicate = iota
	// ReturnExisting returns the existing ActivityFuture if key exists.
	ReturnExisting
	// AllowReuseOnFailure allows reuse only if the previous activity failed.
	AllowReuseOnFailure
	// NoReuse returns an error if the key exists.
	NoReuse
)

type ResultState

type ResultState int

ResultState indicates whether an activity result is success or failure.

const (
	ResultOk ResultState = iota
	ResultErr
)

type RetryableError

type RetryableError interface {
	IsRetryable() bool
}

RetryableError is an interface for errors that know if they are retryable. Handlers can return any error implementing this interface to control retry behavior.

type WorkerConfig

type WorkerConfig struct {
	// QueueName is used as a prefix to avoid conflicts between different applications.
	QueueName string `json:"queue_name"`

	// MaxConcurrentActivities is the maximum number of activities processed concurrently.
	MaxConcurrentActivities int `json:"max_concurrent_activities"`

	// SchedulePollIntervalSeconds is the interval for polling scheduled activities.
	// When nil, defaults to 5 seconds.
	// Only effective for backends that don't handle scheduling natively in Dequeue().
	SchedulePollIntervalSeconds *uint64 `json:"schedule_poll_interval_seconds,omitempty"`

	// LeaseMS is the lease duration in milliseconds for claimed activities.
	// Defaults to 60000 ms (60s).
	LeaseMS *uint64 `json:"lease_ms,omitempty"`

	// ReaperIntervalSeconds is how often the reaper scans for expired leases.
	// Defaults to 5 seconds.
	ReaperIntervalSeconds *uint64 `json:"reaper_interval_seconds,omitempty"`

	// ReaperBatchSize is the max number of expired items to requeue per reaper tick.
	// Defaults to 100.
	ReaperBatchSize *int `json:"reaper_batch_size,omitempty"`

	// ActivityTypes restricts this engine to only dequeue specific activity types.
	// When nil, workers dequeue all activity types.
	ActivityTypes []string `json:"activity_types,omitempty"`

	// MaxActivityDepth caps how deep the parent/child activity tree can grow.
	// A handler attempting to spawn a child beyond this depth will receive ErrDepthExceeded.
	// When zero, defaults to 32.
	MaxActivityDepth uint16 `json:"max_activity_depth,omitempty"`

	// SuspendOnAwait, when true, switches the engine to a semaphore-based
	// concurrency model where ActivityFuture.GetResult releases this
	// activity's worker slot while it waits for the child future to resolve.
	// Eliminates the parent-blocking-on-children starvation pattern that
	// pins worker slots on recursive fan-out workloads. Default false to
	// preserve the existing fixed-goroutine behaviour.
	SuspendOnAwait bool `json:"suspend_on_await,omitempty"`

	// SuspendLeafActivityTypes, when non-empty together with SuspendOnAwait,
	// reserves SuspendLeavesReserved slots for these "leaf" types — i.e. when
	// the free-slot count drops to the reservation, the dispatcher only
	// dequeues leaf types. Prevents the wake-up deadlock where every freed
	// slot is immediately taken by another awakening parent and no leaf can
	// run. Mirrors the parent/leaf split from the WORKER_MODE example
	// pattern, but in-process.
	SuspendLeafActivityTypes []string `json:"suspend_leaf_activity_types,omitempty"`

	// SuspendLeavesReserved is the number of slots set aside for leaf
	// activity types when at-pressure. Only meaningful with SuspendOnAwait
	// and SuspendLeafActivityTypes set. Default 0 (no reservation).
	SuspendLeavesReserved int `json:"suspend_leaves_reserved,omitempty"`

	// ShutdownGraceSeconds bounds the entire shutdown drain — worker loops,
	// dispatchers, in-flight activity goroutines, result-storage goroutines,
	// and worker-pool deregistration all run in parallel under this single
	// budget. When the budget expires, Start() returns even if some
	// goroutines are still in flight (those are then orphaned for the
	// remaining process lifetime, which is fine on a SIGTERM). Default 30s.
	ShutdownGraceSeconds *uint64 `json:"shutdown_grace_seconds,omitempty"`
}

WorkerConfig controls queue behavior and resource usage.

func DefaultWorkerConfig

func DefaultWorkerConfig() WorkerConfig

DefaultWorkerConfig returns a WorkerConfig with sensible defaults.

type WorkerEngine

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

WorkerEngine is the main activity processing engine.

func NewWorkerEngineWithBackend

func NewWorkerEngineWithBackend(backend storage.Storage, config WorkerConfig) *WorkerEngine

NewWorkerEngineWithBackend creates a WorkerEngine from a custom backend.

func (*WorkerEngine) Backend

func (e *WorkerEngine) Backend() storage.Storage

Inspector returns a QueueInspector for observability operations. Import the observability package and use NewQueueInspector(backend) instead for decoupled usage.

func (*WorkerEngine) GetActivityExecutor

func (e *WorkerEngine) GetActivityExecutor() ActivityExecutor

GetActivityExecutor returns an ActivityExecutor for orchestrating activities. Spawns made through the returned executor are roots (no parent lineage).

func (*WorkerEngine) MaxConcurrentActivities

func (e *WorkerEngine) MaxConcurrentActivities() int

MaxConcurrentActivities returns the max workers config for inspector use.

func (*WorkerEngine) RegisterActivity

func (e *WorkerEngine) RegisterActivity(activityType string, handler ActivityHandler)

RegisterActivity registers an activity handler for a given activity type.

func (*WorkerEngine) SetMetrics

func (e *WorkerEngine) SetMetrics(sink MetricsSink)

SetMetrics sets the metrics sink.

func (*WorkerEngine) Start

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

Start starts the worker engine and blocks until shutdown or error.

func (*WorkerEngine) Stop

func (e *WorkerEngine) Stop()

Stop initiates a graceful shutdown of the worker engine.

type WorkerEngineBuilder

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

WorkerEngineBuilder provides fluent configuration for WorkerEngine.

func Builder

func Builder() *WorkerEngineBuilder

Builder creates a new WorkerEngineBuilder.

func (*WorkerEngineBuilder) ActivityTypes

func (b *WorkerEngineBuilder) ActivityTypes(types []string) *WorkerEngineBuilder

ActivityTypes restricts this engine to only dequeue the specified types.

func (*WorkerEngineBuilder) Backend

func (b *WorkerEngineBuilder) Backend(backend storage.Storage) *WorkerEngineBuilder

Backend sets the storage backend. Required.

func (*WorkerEngineBuilder) Build

func (b *WorkerEngineBuilder) Build() (*WorkerEngine, error)

Build creates the WorkerEngine with configured settings.

func (*WorkerEngineBuilder) MaxWorkers

func (b *WorkerEngineBuilder) MaxWorkers(max int) *WorkerEngineBuilder

MaxWorkers sets the maximum concurrent workers.

func (*WorkerEngineBuilder) Metrics

func (b *WorkerEngineBuilder) Metrics(sink MetricsSink) *WorkerEngineBuilder

Metrics sets the metrics sink.

func (*WorkerEngineBuilder) QueueName

func (b *WorkerEngineBuilder) QueueName(name string) *WorkerEngineBuilder

QueueName sets the queue name.

func (*WorkerEngineBuilder) ReserveSlotsForLeaves

func (b *WorkerEngineBuilder) ReserveSlotsForLeaves(n int, leafTypes []string) *WorkerEngineBuilder

ReserveSlotsForLeaves reserves `n` slots out of MaxWorkers for the listed "leaf" activity types when SuspendOnAwait is on. Prevents the wake-up deadlock where every freed slot is immediately retaken by a waking parent and no leaf can run. No effect when SuspendOnAwait is off.

func (*WorkerEngineBuilder) SchedulePollInterval

func (b *WorkerEngineBuilder) SchedulePollInterval(interval time.Duration) *WorkerEngineBuilder

SchedulePollInterval sets the interval for polling scheduled activities.

func (*WorkerEngineBuilder) ShutdownGrace

func (b *WorkerEngineBuilder) ShutdownGrace(d time.Duration) *WorkerEngineBuilder

ShutdownGrace bounds the entire shutdown drain — workers, dispatchers, in-flight activity goroutines, result-storage goroutines, and pool deregistration all run in parallel under this single budget. When the budget expires Start() returns even if some goroutines remain in flight. Defaults to 30s when not set; clamped to >= 1s.

func (*WorkerEngineBuilder) SuspendOnAwait

func (b *WorkerEngineBuilder) SuspendOnAwait(enabled bool) *WorkerEngineBuilder

SuspendOnAwait toggles the semaphore-based dispatcher that releases a parent activity's worker slot while it waits for child futures. Default false (existing fixed-goroutine pool). See WorkerConfig.SuspendOnAwait for the operational notes.

type WorkerEngineWrapper

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

WorkerEngineWrapper provides activity execution capabilities. It implements ActivityExecutor.

func (*WorkerEngineWrapper) Activity

func (w *WorkerEngineWrapper) Activity(activityType string) *ActivityBuilder

Activity creates a fluent activity builder.

type WorkerError

type WorkerError struct {
	Kind    WorkerErrorKind
	Message string
	Cause   error
}

WorkerError represents an error from the worker engine.

func IsWorkerError

func IsWorkerError(err error) (*WorkerError, bool)

IsWorkerError extracts a *WorkerError from err (if any).

func WorkerErrorFromStorage

func WorkerErrorFromStorage(err error) *WorkerError

WorkerErrorFromStorage converts a StorageError to a WorkerError.

func (*WorkerError) Error

func (e *WorkerError) Error() string

func (*WorkerError) IsRetryable

func (e *WorkerError) IsRetryable() bool

IsRetryable returns true if this error may be resolved by retrying.

func (*WorkerError) Unwrap

func (e *WorkerError) Unwrap() error

type WorkerErrorKind

type WorkerErrorKind int

WorkerErrorKind classifies worker engine errors.

const (
	ErrCustom WorkerErrorKind = iota
	ErrQueue
	ErrSerializationW
	ErrTimeoutW
	ErrExecution
	ErrHandlerNotFound
	ErrBackend
	ErrDatabase
	ErrConfiguration
	ErrShutdown
	ErrAlreadyRunning
	ErrScheduling
	ErrDuplicateActivityW
	ErrIdempotencyConflictW
	ErrDepthExceeded
	ErrUnknown
)