dureq

dureq

A distributed, actor-based job scheduling and workflow orchestration system built in Go.

dureq is short for durable redis execution queue.

dureq is a full rewrite of the original pull-based model, now using a push-based actor model powered by Hollywood for inter-node communication. It supports one-time and recurring job scheduling, DAG-based workflows, batch processing, automatic retries, distributed locking, leader election, and real-time event streaming — all backed by Redis. Heavily inspired by Asynq, gocron, and river.

Features

• Flexible Scheduling — Immediate, one-time, interval-based, cron, daily, weekly, and monthly schedules
• Workflow Orchestration — DAG-based task dependencies with automatic state advancement
• Batch Processing — Concurrent item execution with optional preprocessing, chunking, and failure policies
• Actor Model — Cluster-wide singleton actors (scheduler, dispatcher, orchestrator) with per-node workers
• Mux Handler — Pattern-based task type routing (festival.*), global and per-handler middleware
• Monitoring APIs — HTTP REST and gRPC APIs for full cluster observability and management
• Web Dashboard — React-based UI (durequi) for visual monitoring
• Terminal UI — Interactive TUI (dureqctl) for real-time cluster monitoring
• Distributed Locking — Redis-based locks with auto-extension for safe concurrent execution
• Leader Election — Automatic singleton failover across nodes
• Priority Queues — Four priority levels (Low, Normal, High, Critical) with configurable tiers
• Retry Policies — Exponential backoff with jitter, error classification (retryable, non-retryable, rate-limited)
• Overlap Policies — Control concurrent runs for recurring jobs (allow all, skip, buffer one, buffer all)
• Catchup / Backfill — Recover missed executions within a configurable window
• Middleware — Composable handler middleware chains (global + per-handler)
• Progress Reporting — In-flight progress updates from workers
• WebSocket Live Updates — Real-time event push to web dashboard via WebSocket, with automatic polling fallback
• Event Streaming — Real-time lifecycle events via Redis Pub/Sub and Streams
• Payload Search — JSONPath-based payload search across jobs, workflows, and batches
• Unique Keys — Deduplication via unique keys with lookup and manual deletion
• Multi-Tenancy — Key prefix isolation per tenant
• Client SDKs — Go

Architecture

                    ┌─────────────────────┐
                    │     durequi (UI)    │
                    │ React Web Dashboard │
                    └─────────┬───────────┘
                              │
               HTTP REST ─────┤───── gRPC
                              │
Client (Go)          │
  │                 ┌─────────▼───────────┐
  │  enqueue        │      dureqd         │
  │  via Redis      │  (server daemon)    │
  │                 │                     │
  │                 │  :8080 HTTP API     │
  │                 │  :9090 gRPC API     │
  │                 └─────────┬───────────┘
  │                           │
  ▼                           ▼
┌──────────────────────────────────────────────────────┐
│                      Redis                           │
│  Jobs · Schedules · Streams · Locks · Events         │
└──────────────────────────────────────────────────────┘
  │                         ▲
  ▼  Pub/Sub notify         │  persist results
┌──────────────────────────────────────────────────────┐
│  Node (Hollywood Actor Engine)                       │
│                                                      │
│  ┌─────────────┐  ┌──────────────┐  ┌────────────┐   │
│  │  Notifier   │→ │  Dispatcher  │→ │  Worker    │   │
│  │  (per-node) │  │  (singleton) │  │  Supervisor│   │
│  └─────────────┘  └──────────────┘  │  (per-node)│   │
│                          ▲          └─────┬──────┘   │
│  ┌─────────────┐         │                │          │
│  │  Scheduler  │─────────┘          ┌─────▼──────┐   │
│  │  (singleton)│                    │  Worker(s) │   │
│  └─────────────┘                    │  (per-job) │   │
│                                     └─────┬──────┘   │
│  ┌──────────────┐  ┌──────────────┐       │          │
│  │ Orchestrator │← │ EventBridge  │←──────┘          │
│  │  (singleton) │  │  (per-node)  │                  │
│  └──────────────┘  └──────────────┘                  │
│                                                      │
│  ┌───────────────┐  ┌───────────────┐                │
│  │   Heartbeat   │  │ ClusterGuard  │                │
│  │  (per-node)   │  │  (per-node)   │                │
│  └───────────────┘  └───────────────┘                │
└──────────────────────────────────────────────────────┘
         ▲
         │  HTTP client
┌────────┴────────┐
│    dureqctl     │
│  (terminal UI)  │
└─────────────────┘

Singleton actors run on the elected leader node only:

• SchedulerActor — scans due schedules every second, dispatches jobs, detects orphaned runs, backfills missed executions
• DispatcherActor — routes jobs to nodes based on capacity, task-type affinity, and priority
• OrchestratorActor — advances workflow DAG and batch state machines on job completion/failure

Per-node actors:

• WorkerSupervisorActor — manages a pool of WorkerActors, enforces concurrency limits
• WorkerActor — executes a single job handler, manages heartbeats and progress
• EventBridgeActor — persists results to Redis, forwards domain events to OrchestratorActor
• NotifierActor — subscribes to Redis Pub/Sub for new job notifications
• HeartbeatActor — maintains node liveness
• ClusterGuardActor — monitors singleton health, triggers re-election if needed

Getting Started

Prerequisites

• Go 1.25+
• Redis 7+
• Protocol Buffers compiler (for regenerating proto files)

Installation

go get github.com/FDK0901/dureq

Quick Example

Server:

package main

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

    "github.com/FDK0901/dureq/internal/server"
    "github.com/FDK0901/dureq/pkg/types"
)

func main() {
    srv, err := server.New(
        server.WithRedisURL("redis://localhost:6379"),
        server.WithRedisDB(15),
        server.WithNodeID("node-1"),
        server.WithMaxConcurrency(10),
    )
    if err != nil {
        log.Fatal(err)
    }

    srv.RegisterHandler(types.HandlerDefinition{
        TaskType:    "email.send",
        Concurrency: 5,
        Handler: func(ctx context.Context, payload json.RawMessage) error {
            fmt.Printf("Sending email: %s\n", payload)
            return nil
        },
        RetryPolicy: types.DefaultRetryPolicy(),
    })

    ctx := context.Background()
    if err := srv.Start(ctx); err != nil {
        log.Fatal(err)
    }
}

Client (Go):

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    client "github.com/FDK0901/dureq/clients/go"
    "github.com/FDK0901/dureq/pkg/types"
)

func main() {
    cli, err := client.New(client.WithRedisURL("redis://localhost:6379"))
    if err != nil {
        log.Fatal(err)
    }

    ctx := context.Background()

    // Immediate job
    job, _ := cli.EnqueueImmediate(ctx, &client.EnqueueRequest{
        TaskType: "email.send",
        Payload:  []byte(`{"to":"user@example.com","subject":"Hello"}`),
    })
    fmt.Println("Enqueued:", job.ID)

    // Recurring job (every 30 seconds)
    start := time.Now().Add(time.Minute)
    end := start.Add(24 * time.Hour)
    interval := types.Duration(30 * time.Second)
    cli.EnqueueScheduled(ctx, &client.EnqueueRequest{
        TaskType: "report.generate",
        Payload:  []byte(`{}`),
        Schedule: types.Schedule{
            Type:     types.ScheduleDuration,
            Interval: &interval,
            StartsAt: &start,
            EndsAt:   &end,
        },
    })
}

dureqd — Server Daemon

dureqd is the main server process that runs the full dureq cluster node. It starts the actor engine, exposes HTTP and gRPC monitoring APIs, and handles graceful shutdown.

go run ./cmd/dureqd

Configuration

dureqd reads configuration from a YAML file. Set the path via DUREQ_CONFIG_PATH environment variable (default: configs/config.yaml).

dureqd:
  nodeId: ""              # Auto-generated if empty
  apiAddress: ":8080"     # HTTP monitoring API
  grpcAddress: ":9090"    # gRPC monitoring API
  concurrency: 100        # Max concurrent workers
  prefix: ""              # Redis key prefix (multi-tenancy)

redis:
  url: "redis://localhost:6379"
  password: ""
  db: 0
  poolSize: 100

What dureqd provides

dureqctl — Terminal UI

dureqctl is an interactive terminal UI for monitoring and managing dureq clusters in real-time. Built with Bubble Tea.

go run ./cmd/dureqctl

Flags

Pages

Keyboard Controls

durequi — Web Dashboard

A React-based web dashboard for visual monitoring and management of dureq clusters. Connects to the dureqd HTTP and WebSocket APIs.

• Real-time updates via WebSocket (/api/ws) — events push to the browser and invalidate React Query caches, so data refreshes instantly without polling
• Automatic fallback — if WebSocket disconnects, the dashboard seamlessly falls back to configurable polling (default 5s)
• Connection indicator — sidebar shows live connection status (green = Live, yellow = Connecting, gray = Polling)
• Configurable — WebSocket can be toggled on/off from the Settings page

Repository: github.com/FDK0901/durequi

Monitoring API

Both HTTP and gRPC APIs expose the same set of operations. The HTTP API serves the web dashboard and CLI, while the gRPC API provides type-safe access with protobuf definitions.

Endpoints

Jobs:

Schedules:

Nodes:

Runs:

History:

Workflows:

Batches:

Bulk Operations:

Bulk request body: {"ids": ["id1", "id2"]} or {"status": "failed"}

Queues & Groups:

Payload Search (JSONPath):

Unique Keys:

Statistics & Health:

DLQ (Dead Letter Queue):

WebSocket (Real-time Events):

The WebSocket endpoint subscribes to the Redis Pub/Sub {prefix}:events channel and fans out every event to all connected clients. Events include all 29+ types (job, workflow, batch, node, schedule, leader). The durequi dashboard uses this to invalidate React Query caches for instant UI updates, falling back to polling when disconnected.

Pagination — query params: limit (default 10, max 100), offset, sort (newest or oldest)

Scheduling

Overlap Policies

For recurring jobs, control what happens when a new execution is due while a previous one is still running:

Workflows

Define DAG-based workflows where tasks execute in dependency order:

cli.CreateWorkflow(ctx, &types.WorkflowDefinition{
    Name: "user-onboarding",
    Tasks: []types.WorkflowTask{
        {Name: "create-account", TaskType: "account.create", Payload: payload},
        {Name: "send-welcome",   TaskType: "email.welcome",  DependsOn: []string{"create-account"}},
        {Name: "setup-profile",  TaskType: "profile.setup",  DependsOn: []string{"create-account"}},
        {Name: "notify-admin",   TaskType: "admin.notify",   DependsOn: []string{"send-welcome", "setup-profile"}},
    },
})

The orchestrator automatically validates the DAG (no cycles, all dependencies exist), dispatches root tasks first, and advances downstream tasks as dependencies complete.

Batches

Process collections of items concurrently with optional preprocessing:

cli.CreateBatch(ctx, &types.BatchDefinition{
    Name:            "process-images",
    OnetimeTaskType: types.TaskTypePtr("image.prepare"),  // optional preprocessing
    ItemTaskType:    "image.resize",
    Items: []types.BatchItem{
        {ID: "img-1", Payload: json.RawMessage(`{"url":"..."}`)},
        {ID: "img-2", Payload: json.RawMessage(`{"url":"..."}`)},
    },
    ChunkSize:     100,
    FailurePolicy: types.BatchFailureContinueOnError,
})

Failure policies:

• CONTINUE_ON_ERROR — continue processing remaining items even if some fail
• FAIL_ON_ERROR — stop the entire batch on first failure

Error Handling & Retries

// Default retry policy: 3 attempts, 5s initial delay, 5m max, 2x multiplier, 10% jitter
types.DefaultRetryPolicy()

// Custom retry policy
&types.RetryPolicy{
    MaxAttempts:  5,
    InitialDelay: 10 * time.Second,
    MaxDelay:     10 * time.Minute,
    Multiplier:   3.0,
    Jitter:       0.2,
}

Classify errors to control retry behavior:

// Retryable (default) — transient failures, will retry
return &types.RetryableError{Err: err}

// Non-retryable — permanent failures, skip remaining retries
return &types.NonRetryableError{Err: err}

// Rate-limited — retry after a specific duration
return &types.RateLimitedError{Err: err, RetryAfter: 30 * time.Second}

Handler Context

Inside a handler, access job metadata via context helpers:

func handler(ctx context.Context, payload json.RawMessage) error {
    jobID    := types.GetJobID(ctx)
    runID    := types.GetRunID(ctx)
    attempt  := types.GetAttempt(ctx)
    maxRetry := types.GetMaxRetry(ctx)
    taskType := types.GetTaskType(ctx)
    priority := types.GetPriority(ctx)
    nodeID   := types.GetNodeID(ctx)
    headers  := types.GetHeaders(ctx)

    // Report progress
    types.ReportProgress(ctx, map[string]any{"percent": 50})

    return nil
}

Middleware

Middleware can be applied globally via srv.Use() or per-handler via HandlerDefinition.Middlewares. Global middleware runs first, then per-handler middleware, forming a chain around the handler.

// Global middleware — applies to ALL handlers.
srv.Use(func(next types.HandlerFunc) types.HandlerFunc {
    return func(ctx context.Context, payload json.RawMessage) error {
        log.Printf("Starting job %s", types.GetJobID(ctx))
        err := next(ctx, payload)
        log.Printf("Finished job %s (err=%v)", types.GetJobID(ctx), err)
        return err
    }
})

// Per-handler middleware — applied after global middleware.
srv.RegisterHandler(types.HandlerDefinition{
    TaskType:    "email.send",
    Handler:     emailHandler,
    Middlewares: []types.MiddlewareFunc{retryAwareMiddleware},
})

Pattern-Based Routing (Mux)

Register a single handler for multiple task types using glob patterns:

srv.RegisterHandler(types.HandlerDefinition{
    TaskType: "festival.*",  // matches festival.query_wait_time, festival.send_alert, etc.
    Handler:  festivalHandler,
})

Inside the handler, use types.GetTaskType(ctx) to distinguish the actual task type and dispatch accordingly.

Configuration

Server Options

Client Options

Event Types

dureq emits lifecycle events via Redis Streams for monitoring and integration:

Project Structure

dureq/
├── cmd/
│   ├── dureqd/             # Server daemon (actor engine + HTTP/gRPC APIs)
│   │   ├── main.go
│   │   └── config/         # YAML config loading (Koanf)
│   └── dureqctl/           # Interactive terminal UI (Bubble Tea)
│       ├── main.go
│       └── teamodel/       # TUI model, views, and key handling
├── clients/
│   └── go/                 # Go client SDK
├── pkg/
│   ├── types/              # Public domain types (Job, Schedule, Workflow, Batch, etc.)
│   └── dureq/              # Package-level convenience API
├── internal/
│   ├── server/             # Actor-based server entry point and options
│   ├── actors/             # All actor implementations
│   ├── handler/            # Handler registry (task type → handler mapping)
│   ├── monitor/            # HTTP REST + gRPC monitoring APIs, WebSocket hub
│   ├── api/                # HTTP client for dureqctl
│   ├── store/              # Redis persistence layer (Lua scripts, streams, pagination)
│   ├── messages/           # Internal message types
│   ├── scheduler/          # Standalone scheduler component
│   ├── workflow/           # DAG validation & topological sort
│   ├── lock/               # Distributed locking
│   ├── cache/              # In-memory schedule cache
│   └── provider/           # Service providers (Redis client factory)
├── proto/
│   └── dureq/
│       ├── messages.proto          # Inter-actor message definitions
│       └── monitor/v1/             # gRPC monitoring service definitions
│           ├── service.proto       # 37 RPC methods
│           ├── job.proto
│           ├── workflow.proto
│           ├── batch.proto
│           ├── bulk.proto
│           ├── node.proto
│           ├── queue.proto
│           ├── stats.proto
│           └── common.proto
├── gen/dureq/              # Generated protobuf code
├── examples/               # Runnable examples
├── benchmarks/             # Performance benchmarks
├── Makefile                # Proto generation
└── go.mod

Examples

The examples/ directory contains runnable demos:

cd examples/festival
go run main.go

Building

# Build all packages
go build ./...

# Run the server daemon
go run ./cmd/dureqd

# Run the terminal UI
go run ./cmd/dureqctl -api http://localhost:8080

# Run tests
go test ./...

# Run benchmarks
cd benchmarks && go test -bench=. ./...

# Regenerate protobuf code
make genproto

Dependencies