stm

STM - Stupid Task Manager

Distributed task queue with crash recovery and multi-instance coordination.

STM v2 is a persistent task queue designed for distributed Go applications that need:

• Task survival across server restarts
• Multi-instance coordination (multiple workers processing the same queue)
• Per-server concurrency limits
• Priority-based task scheduling
• Automatic retry with exponential backoff

Two backends available:

• PostgreSQL: Durable storage, ACID guarantees, transaction pooling compatible
• Redis: In-memory speed (~3x faster), simpler architecture, automatic TTL cleanup

Choose based on your persistence requirements and throughput needs.

Features

Core (Both Backends)

• Crash Recovery: Tasks survive server restarts - crashed tasks automatically return to pending state
• Multi-Instance Safe: Atomic operations prevent duplicate processing across multiple workers
• Per-Server Concurrency: Limit parallel tasks per backend server (e.g., SSR workers, API endpoints)
• Priority Scheduling: Higher priority tasks processed first within each provider
• Hot Config Reload: Update server lists and concurrency limits without restarting workers
• Scheduled Tasks: Delayed and recurring task execution with retry logic

PostgreSQL Backend

• ACID Guarantees: Full transaction support with rollback capability
• Pgbouncer Compatible: Simple protocol mode works with transaction pooling (no prepared statement errors)
• Advisory Locks: Database-level coordination with FOR UPDATE SKIP LOCKED
• Persistent Indexes: Optimized composite indexes for high-throughput task grabbing
• Throughput: 20,000+ tasks/sec sustained

Redis Backend

• In-Memory Speed: 67,000+ tasks/sec throughput (~3x faster than PostgreSQL)
• Automatic TTL: Tasks auto-delete after 7 days, scheduled tasks after 2 months
• Simpler Operations: Sorted sets + atomic counters instead of transactions
• Pub/Sub Ready: Easy to add instant worker notification (vs polling)
• No Migrations: Schema-less, just connect and run

Architecture

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Instance A │     │  Instance B │     │  Instance C │
│  (Leader)   │     │  (Reader)   │     │  (Reader)   │
└──────┬──────┘     └──────┬──────┘     └──────┬──────┘
       │                   │                   │
       └───────────────────┼───────────────────┘
                           │
                  ┌────────▼────────┐
                  │   PostgreSQL    │
                  │   or Redis      │
                  │                 │
                  │  PostgreSQL:    │
                  │  - stm_tasks    │
                  │  - stm_config   │
                  │  - advisory     │
                  │    locks        │
                  │                 │
                  │  Redis:         │
                  │  - sorted sets  │
                  │  - hashes       │
                  │  - atomic       │
                  │    counters     │
                  └─────────────────┘

Leader Mode: Creates schema/config, initializes provider config, starts workers Reader Mode: Connects to existing config, starts workers

All instances coordinate through backend storage - tasks grabbed atomically, server capacity tracked globally.

Quick Start

Installation

go get github.com/coffyg/stm

Basic Usage (Leader Instance)

package main

import (
    "time"
    "github.com/rs/zerolog"
    "github.com/coffyg/stm"
)

func main() {
    logger := zerolog.New(os.Stdout).With().Timestamp().Logger()

    // Define providers and their servers
    providers := []stm.ProviderConfig{
        {
            Name: "ssr_rendering",
            Servers: []string{
                "http://ssr-worker-1:8080",
                "http://ssr-worker-2:8080",
            },
        },
        {
            Name: "api_calls",
            Servers: []string{"http://api-backend:3000"},
        },
    }

    // Create leader instance (initializes schema)
    tm, err := stm.NewStupidTaskManager(
        "postgres://user:pass@localhost/mydb",
        "worker-instance-1",  // Unique runner ID
        providers,
        &logger,
        getTimeout,  // func(callbackName, provider string) time.Duration
        25,          // Max open connections
        10,          // Max idle connections
    )
    if err != nil {
        panic(err)
    }
    defer tm.Shutdown()

    // Set server concurrency limits
    tm.SetServerMaxParallel("http://ssr-worker-1:8080", 4)
    tm.SetServerMaxParallel("http://ssr-worker-2:8080", 4)
    tm.SetServerMaxParallel("http://api-backend:3000", 50)

    // Start workers
    tm.Start()

    // Add tasks
    task := &MyTask{
        id: "task-123",
        provider: &MyProvider{name: "ssr_rendering"},
        priority: 10,
    }
    tm.AddTask(task)

    // Keep running
    select {}
}

func getTimeout(callbackName, provider string) time.Duration {
    if provider == "ssr_rendering" {
        return 2 * time.Minute
    }
    return 30 * time.Second
}

Adding Reader Instances

// Additional instances connect as readers
tm, err := stm.NewStupidTaskManagerReader(
    "postgres://user:pass@localhost/mydb",
    "worker-instance-2",  // Different runner ID
    &logger,
    getTimeout,
    25, 10,
)
tm.Start()  // Joins the worker pool automatically

Redis Backend Usage

package main

import (
    "time"
    "github.com/rs/zerolog"
    "github.com/coffyg/stm"
)

func main() {
    logger := zerolog.New(os.Stdout).With().Timestamp().Logger()

    // Define providers (same as PostgreSQL)
    providers := []stm.ProviderConfig{
        {
            Name: "ssr_rendering",
            Servers: []string{
                "http://ssr-worker-1:8080",
                "http://ssr-worker-2:8080",
            },
        },
    }

    // Create Redis task manager (leader mode)
    tm, err := stm.NewRedisTaskManager(
        "localhost:6379",  // Redis address
        0,                 // Database index (0-15)
        "worker-instance-1",
        providers,
        &logger,
        getTimeout,
    )
    if err != nil {
        panic(err)
    }
    defer tm.Shutdown()

    // Set server concurrency limits (same API as PostgreSQL)
    tm.SetServerMaxParallel("http://ssr-worker-1:8080", 4)
    tm.SetServerMaxParallel("http://ssr-worker-2:8080", 4)

    // Start workers
    tm.Start()

    // Add tasks (same API as PostgreSQL)
    task := &MyTask{
        id: "task-123",
        provider: &MyProvider{name: "ssr_rendering"},
        priority: 10,
    }
    tm.AddTask(task)

    select {}
}

Redis Reader Instances:

// Additional instances connect as readers
tm, err := stm.NewRedisTaskManagerReader(
    "localhost:6379",
    0,  // Same database index
    "worker-instance-2",  // Different runner ID
    &logger,
    getTimeout,
)
tm.Start()  // Joins the worker pool automatically

Redis Scheduled Tasks:

// Create Redis scheduler
scheduler := stm.NewRedisScheduledTaskScheduler(
    "localhost:6379",
    0,  // Database index
    logger,
)
scheduler.Start()
defer scheduler.Shutdown()

// Register callbacks (same API as PostgreSQL)
scheduler.RegisterCallback(&CleanupCallback{})

// Schedule tasks (same API)
taskID, _ := scheduler.RunIn(3*time.Hour, "cleanup_old_files", map[string]interface{}{
    "directory": "/tmp/uploads",
})

Task Interface

Implement the ITask interface:

type ITask interface {
    GetID() string
    GetProvider() IProvider
    GetPriority() int
    GetRetries() int
    GetMaxRetries() int
    UpdateRetries(int)
    GetTimeout() time.Duration
    GetCallbackName() string
    MarkAsSuccess(totalTime int64)
    MarkAsFailed(totalTime int64, err error)
    OnComplete()
}

Scheduled Tasks

STM includes a built-in scheduler for delayed and recurring task execution:

// Create scheduler
scheduler := stm.NewScheduledTaskScheduler(db, logger)
scheduler.Start()
defer scheduler.Shutdown()

// Implement callback interface
type CleanupCallback struct{}

func (c *CleanupCallback) Name() string {
    return "cleanup_old_files"
}

func (c *CleanupCallback) Handle(payload map[string]interface{}) error {
    directory := payload["directory"].(string)
    deleted, err := cleanupOldFiles(directory)
    fmt.Printf("Cleaned up %d files\n", deleted)
    return err
}

func (c *CleanupCallback) OnComplete(payload map[string]interface{}) {
    // Called after successful execution
}

func (c *CleanupCallback) OnFail(payload map[string]interface{}, err error) {
    // Called after max retries exhausted
}

// Register callback
scheduler.RegisterCallback(&CleanupCallback{})

// Schedule one-time delayed execution
taskID, _ := scheduler.RunIn(3*time.Hour, "cleanup_old_files", map[string]interface{}{
    "directory": "/tmp/uploads",
})

// Schedule recurring execution
taskID, _ := scheduler.RunEvery(24*time.Hour, "cleanup_old_files", map[string]interface{}{
    "directory": "/tmp/cache",
})

// Cancel before execution
scheduler.CancelScheduled(taskID)

// Permanently delete
scheduler.DeleteScheduled(taskID)

Features:

• Crash recovery: Scheduled tasks survive server restarts
• Multi-instance safe: Only one instance executes each scheduled task
• Automatic retry: Exponential backoff (10s, 20s, 40s, 60s max)
• Recurring tasks: Run at fixed intervals indefinitely
• Callback registry: Register functions at startup, reference by name

Use cases:

• Delayed user actions ("delete my account in 30 days")
• Recurring cleanup jobs
• Reminder escalation chains
• Rate-limited API calls
• Gradual feature rollouts

Performance

Benchmark Results

PostgreSQL Backend (PG 17, 25/10 connection pool):

• 20,663 tasks/sec sustained throughput (1000 concurrent goroutines × 10 tasks)
• Crash recovery overhead: ~100-200ms per task (database INSERT + SELECT)
• ACID guarantees: Full transaction support

Redis Backend (Redis 8.2.1):

• 67,740 tasks/sec sustained throughput (~3.3x faster than PostgreSQL)
• 10,000 tasks in 147ms (stress test)
• In-memory operations: Sub-millisecond latency
• Automatic cleanup: TTL-based expiration (7 days for tasks, 2 months for scheduled)

Compare to in-memory (SMT v1): 1M+ tasks/sec, but no persistence or multi-instance coordination.

Backend Selection Guide

Use PostgreSQL when:

• ACID guarantees required (transactions, rollback)
• Long-term task history needed (>7 days)
• Already using PostgreSQL infrastructure
• Pgbouncer transaction pooling required
• Complex queries on task metadata needed

Use Redis when:

• Maximum throughput critical (50k+ tasks/sec)
• Tasks can expire after 7 days
• Already using Redis infrastructure
• Simpler operations preferred (sorted sets vs SQL)
• Instant TTL cleanup desired

Use in-memory queues when:

• Single instance only
• Tasks are cheap to regenerate
• No crash recovery needed
• Maximum throughput critical (>100k tasks/sec)

Configuration

Environment Variables

PostgreSQL:

DATABASE_URL=postgres://user:pass@localhost/dbname
RUNNER_ID=instance-1
MAX_OPEN_CONNS=25
MAX_IDLE_CONNS=10

Redis:

REDIS_ADDR=localhost:6379
REDIS_DB=0
RUNNER_ID=instance-1

Dynamic Configuration

Update provider servers without restart:

tm.UpdateProviderServers("ssr_rendering", []string{
    "http://ssr-worker-1:8080",
    "http://ssr-worker-3:8080",  // New server
})

Update concurrency limits:

tm.SetServerMaxParallel("http://ssr-worker-1:8080", 8)  // Increase to 8 parallel

Config changes picked up within 5 seconds (automatic polling).

Storage Architecture

PostgreSQL Backend

STM creates the following tables:

• stm_tasks - Main task queue with priority scheduling
• stm_scheduled_tasks - Delayed and recurring task execution
• stm_providers - Provider configuration (servers per provider)
• stm_server_limits - Per-server concurrency limits
• stm_server_usage - Active task tracking for capacity control
• stm_config - Version tracking for hot reload
• stm_checkpoints - Optional task checkpoint storage
• stm_task_groups - Task batching and grouping

See stm_schema.sql for full schema.

Optimized indexes for task grabbing:

-- Worker task grabbing (composite index)
CREATE INDEX idx_tasks_pending ON stm_tasks
(runner_hash, provider_name, priority DESC, created_at ASC)
WHERE status = 'pending';

-- Provider discovery (Start())
CREATE INDEX idx_tasks_runner_providers ON stm_tasks
(runner_hash, provider_name);

-- Scheduled task polling
CREATE INDEX idx_scheduled_ready ON stm_scheduled_tasks (next_run_at)
WHERE status = 'pending';

Partial indexes (WHERE clauses) minimize write overhead - only pending/running tasks maintain index entries.

Redis Backend

Redis uses the following key patterns:

Task Queue:

• stm:tasks:{runnerID}:{provider}:pending - Sorted set (score = priority * 1e12 + timestamp)
• stm:tasks:{runnerID}:{provider}:meta:{taskID} - Hash (task metadata, 7 day TTL)

Scheduled Tasks:

• stm:scheduled:pending - Sorted set (score = next_run_at timestamp)
• stm:scheduled:meta:{taskID} - Hash (task metadata, 2 month TTL)

Configuration:

• stm:config:providers - Hash (provider → servers JSON)
• stm:config:limits - Hash (server_prefix → max_parallel)
• stm:config:version - Integer (hot reload version tracking)

Concurrency Control:

• stm:server:{server}:slots - Atomic counter (current active tasks)

Advantages:

• No schema migrations required
• Automatic TTL cleanup
• Atomic operations (ZPOPMAX, INCR, DECR)
• In-memory speed

Acknowledgments

Scheduled tasks feature inspired by Absurd Workflows: PostgreSQL Enables Ridiculous Scheduling Patterns - exploring what becomes possible when task queues persist in PostgreSQL instead of memory.

License

MIT