snerd

package module
v0.2.5 Latest Latest
Warning

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

 Go to latest Published: Aug 19, 2026 License: MIT Imports: 25 Imported by: 0

README

Snerd-Go Logo

⚙️ snerd-go v0.2.5

A blazingly fast, brutally simple, zero-infrastructure embedded background job engine for Go.

Go Reference Docs

If you are tired of wrestling with heavy, bloated background job frameworks like Redis, Postgres tables, or RabbitMQ just to send a few emails in the background... well, you are in the right place.

snerd-go is an embedded, high-performance background task queue that lives entirely in a single, perfectly OS-locked, append-only .log file on your file system. It was designed to bring aggressive concurrency and a lightweight footprint to your Go microservices.

No databases. No external daemons. No nonsense.

🔥 Features

  • Zero External Infrastructure: You don't need a Redis cluster. Your tasks are persisted directly to .snerdata/tasks/tasks.log using standard filesystem I/O.
  • Built-in Web Dashboard: A one-line StartDashboard(port) serves a live React UI with queue stats, job table, and a real-time progress stream.
  • Bulletproof File Locks: Safely scales across multiple processes! We utilize OS-level file-locking boundaries to guarantee that your tasks are never corrupted.
  • Smart API Rate-Limiting: Natively tracks rateLimitGroup execution velocity to prevent 429 "Too Many Requests" API errors.
  • Payload-Hashing Deduplication: Automatically computes cryptographic hashes to drop duplicate tasks instantly.
  • Dynamic Float Prioritization: A native Binary Max-Heap bypasses standard FIFO rules for high urgency tasks.
  • Cron, Webhooks & Hard Timeouts: Recurring schedules, serverless HTTP execution, and per-task execution timeouts.
  • Progress Streaming: Handlers can emit live progress events that stream straight into the dashboard.
  • Dead-Letter Queue (DLQ): Built-in maxRetries limits and hooks to elegantly catch and bury poison-pill tasks.

📦 Installation

go get github.com/speed-nerd/snerd-go

🚀 Quickstart (Basic)

It takes roughly 3 lines of code to spin up a queue and start firing background jobs.

package main

import (
	"context"
	"fmt"
	"time"

	snerd "github.com/speed-nerd/snerd-go"
)

func main() {
	// 1. Create the Queue (name, max size, processor poll interval)
	// For local development this persists to ./.snerdata/tasks/tasks.log
	queue := snerd.NewAnyQueue("my-fast-queue", 10, 2*time.Second)

	// Need a custom location instead? (durable network-drive storage, per-server
	// isolation, or keeping tests out of .snerdata)
	// queue := snerd.NewAnyQueueWithStorage("my-fast-queue", 10, 2*time.Second, "/var/data/snerd/tasks.log")

	// 2. Register your Task Handler (the closure that does the actual work)
	snerd.RegisterTaskHandler("generate_ai_image", func(ctx context.Context, parameters string) error {
		fmt.Printf("Generating image with payload: %s\n", parameters)

		// do your heavy lifting here!
		// return fmt.Errorf("...") to trigger a retry!
		return nil
	})

	// 3. (Optional) Register a Dead-Letter Handler for when retries run out
	snerd.RegisterMaxRetryHandler("generate_ai_image", func(ctx context.Context, parameters string) error {
		fmt.Printf("Task permanently failed! Payload: %s\n", parameters)
		return nil
	})

	// 4. Enqueue a task! (maxRetries=3, retryAfterHours=1.0)
	task, _ := snerd.NewSnerdTask(
		"unique-task-id-123",                    // Unique task ID
		"generate_ai_image",                     // Task type (matches handler)
		map[string]string{"prompt": "A crab in space"}, // JSON payload
		3,    // Max retries
		1.0,  // Delay in hours before a failed task is retried
	)
	queue.EnqueueSnerdTask(task)

	// Keep your app alive — jobs run in background goroutines
	select {}
}

⚙️ Advanced Task Configuration

To power complex workflows, tasks can be configured with advanced orchestration parameters via NewSnerdTaskAdvanced (pass nil for anything you don't need):

rateLimitGroup := "openai_api"
maxPerMinute := 50
autoDedupe := true
urgencyScore := 0.95
cronStr := "1h"

task, _ := snerd.NewSnerdTaskAdvanced(
	"unique-task-id-123",  // Unique task ID
	"generate_ai_image",   // Task type (matches handler)
	map[string]string{"prompt": "A crab in space"}, // JSON payload
	3,                     // Max retries
	1.0,                   // Delay in hours before a failed task is retried
	&rateLimitGroup,       // Rate limit group
	&maxPerMinute,         // Max executions per minute for that group
	&autoDedupe,           // Auto-dedupe identical payloads
	&urgencyScore,         // Urgency score (higher floats to the front)
	nil,                   // executeAt — RFC3339 timestamp for delayed execution
	&cronStr,              // Cron — recurring job, runs every 1 hour
	nil,                   // webhookUrl — HTTP execution instead of a local handler
	nil,                   // maxExecutionSeconds — hard timeout
)
queue.EnqueueSnerdTask(task)
Parameter Type Default Description
maxRetries int — How many times a failed task is retried before hitting the Dead Letter Queue.
retryAfterHours float64 — Backoff in hours before a failed task is retried (e.g. 0.001 ≈ seconds).
autoDedupe *bool nil If true, a cryptographic hash of taskType + parameters is computed. If an identical payload is already pending, the new task is silently dropped.
urgencyScore *float64 nil A value (e.g. 0.99) used to bypass the standard FIFO queue. A Binary Max-Heap floats high urgency tasks to the front.
rateLimitGroup *string nil A custom string (e.g. "openai_api") that groups tasks together for backpressure control.
maxPerMinute *int nil Used with rateLimitGroup. If the group exceeds this limit in a 60-second rolling window, further tasks in the group pause for a minute — natively preventing 429 errors.
executeAt *string nil An RFC3339 timestamp of when the job should first run (delayed execution).
cron *string nil A cron expression for recurring jobs: standard 5-field ("0 * * * *"), 6-field with seconds ("*/10 * * * * *"), or shorthands "30s", "10m", "2h", "1d".
webhookUrl *string nil Optional webhook URL — the payload is dispatched via HTTP POST instead of a local handler.
maxExecutionSeconds *int nil Optional hard timeout in seconds (see below).

⏱️ Note on Hard Timeouts (maxExecutionSeconds)

When maxExecutionSeconds is provided, the engine executes your handler with a context.WithTimeout. If the task takes longer than the timeout, the context is cancelled. If your handler respects context cancellation (select on ctx.Done()), it will terminate early and the execution is marked as failed and retried:

snerd.RegisterTaskHandler("slow-job", func(ctx context.Context, parameters string) error {
	select {
	case <-time.After(10 * time.Minute): // the actual work
		return nil
	case <-ctx.Done():
		return ctx.Err() // gives up promptly when the timeout trips
	}
})

🌐 HTTP Webhooks (Serverless Execution)

You can configure a task to execute externally via an HTTP POST request. By setting a webhookUrl, the background processor skips any registered handlers and directly invokes the HTTP endpoint with the payload and the header X-SnerdMQ-Event: Execute.

If the endpoint returns a non-2xx status code, it triggers a retry. If it permanently fails (reaches maxRetries), the Dead Letter Queue event is automatically fired via a final HTTP POST to the same webhookUrl with the header X-SnerdMQ-Event: MaxRetriesReached.

🕒 Cron Jobs vs. Retryable Jobs

When using the scheduling features, it is important to understand the difference between Cron and Retry behaviors:

  • A Cron Job is a Repeatable Job that executes again only after a success, on a fixed schedule.
  • A Retryable Job is a Recovery Job that executes again only after a failure, attempting to recover using the retryAfterHours backoff.
  • Combined: If a Cron Job fails, it temporarily uses retryAfterHours to retry until it recovers. Once it succeeds, it goes back to ticking on its standard cron schedule!

☠️ Dead Letter Queue (Handling Permanent Failures)

The DLQ captures tasks that have exhausted all maxRetries. Define a custom handler with snerd.RegisterMaxRetryHandler(taskType, handler) — critical for alerting or manual intervention when a background process consistently fails.

📁 Custom Storage Location

By default, tasks persist to .snerdata/tasks/tasks.log. To use a different file — isolating queues per concern, pointing at a network drive (EFS/NFS) for durable storage, or keeping tests clean — use NewAnyQueueWithStorage:

// Same semantics as NewAnyQueue, plus an explicit task log path
queue := snerd.NewAnyQueueWithStorage("image-processing", 10, 2*time.Second, "/mnt/efs/image-jobs/tasks.log")

The rate limiter state file (rate_limits.json) is stored alongside the task log, so two queues on different paths are fully independent — including their dashboards.

One queue instance per storage file. Each queue takes an exclusive OS-level lock on its task log (e.g. tasks.log.lock) at creation. A second queue on the same file fails fast with a panic instead of racing it and double-executing tasks — so register all your task types on a single queue, or give each queue its own path.

📊 Live Dashboard

snerd-go ships with a built-in React UI dashboard served directly by the library — no extra services or dependencies required. It gives you a real-time window into your queue:

  • Live stats: total enqueued, processed, and failed jobs
  • Recent Jobs table: per-task status (queued, active, completed, failed, dead_letter), retry counts, and badges showing which features a task uses (cron / webhook / timeout)
  • Real-time Progress Stream: live output from YieldProgress calls in your handlers
queue := snerd.NewAnyQueue("my-queue", 10, 2*time.Second)

// Start the built-in dashboard on http://localhost:9090
queue.StartDashboard(9090)

Then open http://localhost:9090 in your browser. The page polls a small JSON API exposed by the library — also handy if you want to build your own tooling on top:

Endpoint Returns
/api/stats {"enqueued": N, "processed": N, "failed": N}
/api/tasks All jobs with status, retries, cron, webhook, timeout info
/api/progress The last 100 progress events ({ts, task_id, data})

Serving the UI: the dashboard page is the single file static/index.html, resolved relative to your process's working directory. The bundle ships with this repo under static/ — run your binary from the directory that contains the static/ folder (or copy the folder next to your binary).

Note: StartDashboard only serves the UI — your jobs keep running whether or not the dashboard is open.

📡 Progress Reporting

Long-running handlers can stream live updates to the Dashboard's Progress Stream (ideal for streaming LLM tokens or multi-step ETL work):

snerd.RegisterTaskHandler("generate_report", func(ctx context.Context, parameters string) error {
	for step := 1; step <= 10; step++ {
		doWork(step)
		queue.YieldProgress("report-task-1", fmt.Sprintf("Step %d/10 complete", step))
	}
	return nil
})

You can also subscribe to the raw progress feed from your own code (each message is a JSON string with task_id and data):

for msg := range queue.SubscribeProgress() {
	fmt.Println("progress:", msg)
}

🧩 Queue Topology: One Queue or Many?

The recommended pattern is one queue instance per application: register every job type on it and serve a single shared dashboard:

package main

import (
	"context"
	"fmt"
	"time"

	snerd "github.com/speed-nerd/snerd-go"
)

func main() {
	// ONE queue for the whole app (persists to ./.snerdata/tasks/tasks.log)
	queue := snerd.NewAnyQueue("main", 10, 2*time.Second)

	// Job type #1: image processing
	snerd.RegisterTaskHandler("process_image", func(ctx context.Context, data string) error {
		fmt.Printf("Processing image: %s\n", data)
		return nil
	})

	// Job type #2: OTP emails — same queue
	snerd.RegisterTaskHandler("send_otp_email", func(ctx context.Context, data string) error {
		fmt.Printf("Sending OTP: %s\n", data)
		return nil
	})

	// Both job types flow through the exact same queue
	imgTask, _ := snerd.NewSnerdTask("img-1", "process_image", map[string]string{"image_id": "abc123"}, 3, 0.5)
	queue.EnqueueSnerdTask(imgTask)

	otpTask, _ := snerd.NewSnerdTask("otp-1", "send_otp_email", map[string]string{"to": "john@wick.com"}, 3, 0.5)
	queue.EnqueueSnerdTask(otpTask)

	// ONE dashboard shows every job type
	queue.StartDashboard(9090)

	select {} // keep the process alive — jobs run in background goroutines
}

All job types share everything: the same persistent job log, retry/DLQ pipeline, rate-limit state, stats — and one dashboard at http://localhost:9090 showing all of them.

🚫 Same storage twice = fails fast

Each queue takes an exclusive OS-level lock on its task log at creation. A second queue on the same storage fails instead of silently double-executing your tasks:

first := snerd.NewAnyQueue("main", 10, 2*time.Second)   // ✅ owns .snerdata/tasks/tasks.log
second := snerd.NewAnyQueue("other", 10, 2*time.Second) // ❌ panics:
// "[Snerd] ERROR: Another queue instance is already running on storage ..."

This applies across processes too — a second process pointed at the same log file also fails to start its queue.

🔀 Need multiple queues? Give each one its own storage

images := snerd.NewAnyQueueWithStorage("images", 10, 2*time.Second, "./.snerdata-images/tasks.log")
emails := snerd.NewAnyQueueWithStorage("emails", 10, 500*time.Millisecond, "./.snerdata-emails/tasks.log")

images.StartDashboard(9090) // separate dashboards, so separate ports
emails.StartDashboard(9091)

Now you have two fully independent engines: separate job logs, separate rate-limit state, separate dashboards. Only split when you actually need isolation (different cadence, different retention, independent monitoring) — otherwise the singleton is simpler and recommended.

🌍 Advanced: Distributed Scaling

A queue instance exclusively owns its storage file: it takes an OS-level lock (<tasks.log>.lock) at creation and holds it for its lifetime. A second instance pointed at the same file — in the same process or on another server — fails fast instead of racing it and double-executing tasks.

Scaling out therefore means one queue per server, each with its own storage. Your load balancer routes requests across servers, and every server processes the tasks it enqueued:

// Each server runs its own queue on its own log file (local disk works fine)
queue := snerd.NewAnyQueueWithStorage("worker-server-1", 10, 2*time.Second, "/var/data/snerd/tasks.log")

A shared network drive (AWS EFS or NFS) is still a good home for that log when a single instance needs durable storage — e.g. a container that restarts but must keep its queue state. OS-level file locking keeps writes safe — no Redis required.

🔧 Queue API Reference

API Description
snerd.NewAnyQueue(args ...interface{}) Create a queue. Variadic options: string = name (default "default-queue"), int = max size (default 100), time.Duration = processor poll interval (default 10s). Persists to .snerdata/tasks/tasks.log. Panics if another queue instance already owns that file.
snerd.NewAnyQueueWithStorage(name, maxSize, interval, storePath) Create a queue with an explicit task log file location instead of the default .snerdata/tasks/tasks.log. Panics if another queue instance already owns that file.
queue.EnqueueSnerdTask(task) / queue.Enqueue(task) Enqueue a task. Due tasks execute immediately in background goroutines; the rest are picked up by the processor loop.
snerd.RegisterTaskHandler(type, handler) Register func(ctx context.Context, parameters string) error for a task type.
snerd.RegisterMaxRetryHandler(type, handler) Register the Dead-Letter handler for a task type.
queue.StartDashboard(port int) Serve the built-in dashboard UI on the given port.
queue.YieldProgress(taskID, data) Emit a progress event (dashboard Progress Stream / SubscribeProgress).
queue.SubscribeProgress() Receive progress events as JSON strings on a channel.
queue.StopProcessor() Stop the background polling loop.
queue.ProcessDueTasks() Manually trigger one processing sweep.
queue.Name(), queue.Size(), queue.RemainingCapacity(), queue.TotalEnqueued(), queue.TotalProcessed() Queue inspection helpers.

🧠 Architecture Details

snerd-go utilizes an Append-Only Log Model to achieve massive write speeds. Instead of updating rows in a database, every time a task is enqueued, updated, or deleted, a brand new JSON line is instantly appended to the end of the log file.

When the queue wakes up on its polling interval, it scans the log, maps out the absolute latest state of every task, and spawns parallel goroutines for anything that is currently due (executeAt <= now and retryAfterTime <= now).

If your file ever grows too large, snerd-go atomically clones, shrinks, and replaces the file in the background (Log Compaction) to keep disk space minimal.

🤝 License

MIT License. Do whatever you want with it, just don't let your tasks die unhandled.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func CalculateDynamicQueueSize 

func CalculateDynamicQueueSize() int

CalculateDynamicQueueSize returns an optimal queue size based on system resources. It uses available CPU and memory to auto-scale queue capacity.

func DecodeTaskData 

func DecodeTaskData(encoded string, target interface{}) error

Helper method to decode task data from JSON

func DeleteTask 

func DeleteTask(taskId string) error

DeleteTask removes a task from the database by its TaskID

func EncodeTaskData 

func EncodeTaskData(data interface{}) (string, error)

Helper method to encode task data into JSON and save it

func EnsureTaskTypesRegistered 

func EnsureTaskTypesRegistered()

EnsureTaskTypesRegistered ensures that all known task types are registered This is called before processing tasks to prevent the "no factory registered" error

func GetRegisteredTaskFactory 

func GetRegisteredTaskFactory(retryableTask RetryableTask) (func(id string, data string) (Task, error), bool)

GetRegisteredTaskFactory returns a task factory for a specific task type

func LogQueueUsage 

func LogQueueUsage(ctx context.Context, queue *AnyQueue, wg *sync.WaitGroup)

LogQueueUsage monitors a queue and logs its stats periodically. Logging stops automatically when the queue is empty or when the context is canceled.

func ProcessInMemoryQueue 

func ProcessInMemoryQueue(ctx context.Context, queue *AnyQueue, wg *sync.WaitGroup)

ProcessInMemoryQueue processes tasks in an in-memory queue with proper rate limiting This is specifically for NON-retryable, in-memory tasks only

func ProcessRetryQueue 

func ProcessRetryQueue(ctx context.Context, queue *AnyQueue, wg *sync.WaitGroup, interval time.Duration)

ProcessRetryQueue starts a background goroutine that periodically processes retryable tasks. It polls for due tasks at the specified interval and processes them using the provided queue. ProcessRetryQueue starts a goroutine that processes a queue at regular intervals.

func RegisterInitFunction 

func RegisterInitFunction(fn InitFunction)

RegisterInitFunction registers a function to be called when ensuring task types are registered This allows packages to register their task types without circular dependencies

func RegisterMaxRetryHandler 

func RegisterMaxRetryHandler(taskType string, handler OnMaxRetryHandler)

RegisterMaxRetryHandler registers a handler for when a task reaches max retries

func RegisterTaskHandler 

func RegisterTaskHandler(taskType string, handler TaskHandler)

RegisterTaskHandler registers a handler for a specific task type

func RegisterTaskType 

func RegisterTaskType[P any](
	taskType string,
	creator func(id string, payload P) Task,
)

RegisterTaskType registers a task type with a factory function This allows tasks to be recreated from their saved data

func SerializeTaskPayload 

func SerializeTaskPayload[P any](payload P) (string, error)

SerializeTaskPayload serializes a task payload for storage This allows client code to create task data without knowing serialization details

Types

type AnyQueue 

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

AnyQueue is a thread-safe queue that manages SnerdTask execution, retry logic, and statistics

func NewAnyQueue 

func NewAnyQueue(args ...interface{}) *AnyQueue

NewAnyQueue creates a new queue with the given parameters

func NewAnyQueueWithStorage added in v0.2.5 

func NewAnyQueueWithStorage(name string, maxSize int, processingInterval time.Duration, taskStorePath string) *AnyQueue

NewAnyQueueWithStorage creates a new queue that persists tasks to a custom file location instead of the default ./.snerdata/tasks/tasks.log. This is useful for isolating queues per concern, pointing at a shared network drive (e.g. EFS/NFS) for cross-process queue sharing, or test isolation.

func (*AnyQueue) Enqueue 

func (q *AnyQueue) Enqueue(task Task) error

Enqueue adds a task to the queue

func (*AnyQueue) EnqueueSnerdTask 

func (q *AnyQueue) EnqueueSnerdTask(task *SnerdTask) error

EnqueueSnerdTask adds a parameter-based SnerdTask to the queue for execution This is the preferred method for adding new tasks as it uses the parameter-based approach that doesn't require client-side task registration

func (*AnyQueue) Name 

func (q *AnyQueue) Name() string

func (*AnyQueue) ProcessDueTasks 

func (q *AnyQueue) ProcessDueTasks()

ProcessDueTasks processes all tasks that are due for execution (retry time has passed)..

func (*AnyQueue) RemainingCapacity 

func (q *AnyQueue) RemainingCapacity() int

RemainingCapacity returns the number of additional tasks that can be enqueued before reaching maxSize..

func (*AnyQueue) Size 

func (q *AnyQueue) Size() int

Size returns the number of active tasks currently in the queue.

func (*AnyQueue) StartDashboard 

func (q *AnyQueue) StartDashboard(port int)

StartDashboard starts the built-in dashboard UI on the given port.

The dashboard is a single-page React app (served from ./static/index.html relative to the process working directory) that shows live queue stats, a Recent Jobs table, and a real-time Progress Stream fed by YieldProgress. Updates are delivered via HTTP polling of the JSON API (/api/stats, /api/tasks, /api/progress).

The dashboard only serves the UI — jobs keep running whether or not it is open.

func (*AnyQueue) StopProcessor 

func (q *AnyQueue) StopProcessor()

StopProcessor stops the background task processor

func (*AnyQueue) SubscribeProgress 

func (q *AnyQueue) SubscribeProgress() <-chan string

SubscribeProgress returns a channel that receives real-time task progress JSON chunks.

func (*AnyQueue) TotalEnqueued 

func (q *AnyQueue) TotalEnqueued() int

TotalEnqueued returns the total number of tasks that have been enqueued.

func (*AnyQueue) TotalProcessed 

func (q *AnyQueue) TotalProcessed() int

TotalProcessed returns the total number of tasks that have been processed and dequeued.

func (*AnyQueue) YieldProgress 

func (q *AnyQueue) YieldProgress(taskID string, data string)

YieldProgress broadcasts a progress update to all subscribers.

type FileStore 

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

FileStore provides persistent storage for retryable tasks. It manages task log files, compaction, and metadata tracking for tasks.

func NewFileStore 

func NewFileStore(path string) (*FileStore, error)

NewFileStore creates a new FileStore for the given file path. It rebuilds metadata from the existing log file if present.

func (*FileStore) Compact 

func (fs *FileStore) Compact() error

func (*FileStore) CreateTask 

func (fs *FileStore) CreateTask(task *RetryableTask) error

CreateTask appends a new retryable task to the log file and updates internal counters.

func (*FileStore) DeleteTask 

func (fs *FileStore) DeleteTask(taskID string) error

func (*FileStore) GetLatestTask 

func (fs *FileStore) GetLatestTask(taskID string) (*RetryableTask, error)

func (*FileStore) ReadDueTasks 

func (fs *FileStore) ReadDueTasks() ([]*RetryableTask, error)

func (*FileStore) ReadTasks 

func (fs *FileStore) ReadTasks() ([]*RetryableTask, error)

func (*FileStore) RebuildMetaData 

func (fs *FileStore) RebuildMetaData() error

RebuildMetaData scans the log file and rebuilds internal counters for tasks and deletions.

func (*FileStore) UpdateTaskRetryConfig 

func (fs *FileStore) UpdateTaskRetryConfig(taskID string, taskErr error) error

type InitFunction 

type InitFunction func()

InitFunction is a function that registers task types

type JobErrorReturn 

type JobErrorReturn struct {
	ErrorObj    error
	ErrorString string `json:"error"` // Used for JSON serialization
	RetryWorthy bool   `json:"retry_worthy"`
}

JobErrorReturn contains error information from task execution JobErrorReturn holds error information for a failed job and implements custom JSON marshaling/unmarshaling to handle the error type

func (JobErrorReturn) MarshalJSON 

func (j JobErrorReturn) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler interface

func (*JobErrorReturn) UnmarshalJSON 

func (j *JobErrorReturn) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler interface

type OnMaxRetryHandler 

type OnMaxRetryHandler func(ctx context.Context, parameters string) error

OnMaxRetryHandler is a function that handles when a task reaches max retries

type PriorityQueue 

type PriorityQueue []*RetryableTask

PriorityQueue implements heap.Interface and holds RetryableTasks

func (PriorityQueue) Len 

func (pq PriorityQueue) Len() int

func (PriorityQueue) Less 

func (pq PriorityQueue) Less(i, j int) bool

func (*PriorityQueue) Pop 

func (pq *PriorityQueue) Pop() interface{}

func (*PriorityQueue) Push 

func (pq *PriorityQueue) Push(x interface{})

func (PriorityQueue) Swap 

func (pq PriorityQueue) Swap(i, j int)

type RateLimitEntry 

type RateLimitEntry struct {
	Count     int       `json:"count"`
	WindowEnd time.Time `json:"window_end"`
}

type RateLimiter 

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

func NewRateLimiter 

func NewRateLimiter(storageDir string) *RateLimiter

func (*RateLimiter) CheckLimit 

func (r *RateLimiter) CheckLimit(group string, maxPerMinute int) bool

CheckLimit returns true if the task is allowed to execute, false if it should be rate-limited

type RetryableTask 

type RetryableTask struct {
	TaskID          string    `json:"taskId"`
	RetryCount      int       `json:"retryCount"`
	MaxRetries      int       `json:"maxRetries"`
	RetryAfterHours float64   `json:"retryAfterHours"`
	RetryAfterTime  time.Time `json:"retryAfterTime"`
	TaskData        string    `json:"taskData"` // JSON string to store task-specific data
	TaskType        string    `json:"taskType"` // For diagnostic purposes only
	RateLimitGroup  *string   `json:"rate_limit_group,omitempty"`
	MaxPerMinute    *int      `json:"max_per_minute,omitempty"`
	AutoDedupe      *bool     `json:"autoDedupe,omitempty"`
	PayloadHash     *string   `json:"payloadHash,omitempty"`
	UrgencyScore    *float64  `json:"urgency_score,omitempty"`
	// Fields to store error information for OnMaxRetryReached
	LastErrorObj        error
	LastJobError        *JobErrorReturn
	ExecuteAt           time.Time  `json:"executeAt"`
	CronExpr            *string    `json:"cronExpression,omitempty"`
	WebhookUrl          *string    `json:"webhookUrl,omitempty"`
	MaxExecutionSeconds *int       `json:"maxExecutionSeconds,omitempty"`
	CreatedAt           time.Time  `json:"-"`
	UpdatedAt           time.Time  `json:"-"`
	DeletedAt           *time.Time `json:"deletedAt,omitempty"`
	// Embedded task object - this is the actual task that will be executed
	EmbeddedTask Task `json:"-"`
}

Task with retries

func CreateTaskWithPayload 

func CreateTaskWithPayload[P any](
	taskID string,
	taskType string,
	payload P,
	maxRetries int,
	retryAfterHours int,
) (*RetryableTask, error)

CreateTaskWithPayload creates a new task with the provided payload

func FetchDueTasks 

func FetchDueTasks() ([]RetryableTask, error)

FetchDueTasks gets all tasks that are due for execution based on RetryAfter time

func (*RetryableTask) Execute 

func (t *RetryableTask) Execute(ctx context.Context) error

func (*RetryableTask) GenerateRandomString 

func (t *RetryableTask) GenerateRandomString(length int) (string, error)

func (*RetryableTask) GetMaxRetries 

func (t *RetryableTask) GetMaxRetries() int

func (*RetryableTask) GetRetryAfterHours 

func (t *RetryableTask) GetRetryAfterHours() float64

func (*RetryableTask) GetRetryAfterTime 

func (t *RetryableTask) GetRetryAfterTime() time.Time

func (*RetryableTask) GetRetryCount 

func (t *RetryableTask) GetRetryCount() int

func (*RetryableTask) GetTaskID 

func (t *RetryableTask) GetTaskID() string

func (*RetryableTask) MarshalJSON 

func (t *RetryableTask) MarshalJSON() ([]byte, error)

MarshalJSON implements the json.Marshaler interface to ensure proper serialization of RetryableTask

func (*RetryableTask) Save 

func (t *RetryableTask) Save() error

Save a task to the database

func (*RetryableTask) UnmarshalJSON 

func (t *RetryableTask) UnmarshalJSON(data []byte) error

UnmarshalJSON implements the json.Unmarshaler interface to properly deserialize a RetryableTask

func (*RetryableTask) UpdateTaskRetryConfig 

func (t *RetryableTask) UpdateTaskRetryConfig(taskId string) error

UpdateTaskRetryConfig updates a task's retry configuration in the database

func (*RetryableTask) UpdateTaskRetryConfigWithError 

func (t *RetryableTask) UpdateTaskRetryConfigWithError(taskId string, errorObj error) error

UpdateTaskRetryConfigWithError updates a task's retry configuration and stores error information

type SnerdTask 

type SnerdTask struct {
	// Core Task Identification
	TaskID string `json:"taskId"` // Unique identifier for the task

	// Retry Configuration
	RetryCount      int       `json:"retryCount"`      // Current retry count
	MaxRetries      int       `json:"maxRetries"`      // Maximum number of retries allowed
	RetryAfterHours float64   `json:"retryAfterHours"` // Hours to wait before retrying
	RetryAfterTime  time.Time `json:"retryAfterTime"`  // Timestamp for next retry attempt

	// Task Execution Data
	TaskType   string `json:"taskType"`   // Type of task (maps to registered handler)
	Parameters string `json:"parameters"` // JSON-encoded parameters for the task

	RateLimitGroup *string  `json:"rate_limit_group,omitempty"`
	MaxPerMinute   *int     `json:"max_per_minute,omitempty"`
	AutoDedupe     *bool    `json:"autoDedupe,omitempty"`
	PayloadHash    *string  `json:"payloadHash,omitempty"`
	UrgencyScore   *float64 `json:"urgency_score,omitempty"`

	LastErrorObj error           `json:"lastErrorObj"` // Last error that occurred
	LastJobError *JobErrorReturn `json:"lastJobError"` // Detailed error information

	ExecuteAt           time.Time `json:"executeAt"`
	CronExpr            *string   `json:"cronExpression,omitempty"`
	WebhookUrl          *string   `json:"webhookUrl,omitempty"`
	MaxExecutionSeconds *int      `json:"maxExecutionSeconds,omitempty"`

	// Timestamps for record-keeping
	CreatedAt time.Time  `json:"-"`                   // When the task was created
	UpdatedAt time.Time  `json:"-"`                   // When the task was last updated
	DeletedAt *time.Time `json:"deletedAt,omitempty"` // Soft deletion timestamp
}

SnerdTask is a retryable task that stores parameters instead of implementations

func CreateTask 

func CreateTask(taskID string, taskType string, parameters interface{}, maxRetries int, retryAfterHours float64) (*SnerdTask, error)

CreateTask is a convenience function that creates a new task with the given parameters. This is the simplified client API function for creating parameter-based tasks

func FromRetryableTask 

func FromRetryableTask(rt *RetryableTask) *SnerdTask

FromRetryableTask creates a SnerdTask from a RetryableTask This is used when loading tasks from the file store

func NewSnerdTask 

func NewSnerdTask(
	taskID string,
	taskType string,
	parameters interface{},
	maxRetries int,
	retryAfterHours float64,
) (*SnerdTask, error)

NewSnerdTask creates a new task with the specified parameters

func NewSnerdTaskAdvanced 

func NewSnerdTaskAdvanced(
	taskID string,
	taskType string,
	parameters interface{},
	maxRetries int,
	retryAfterHours float64,
	rateLimitGroup *string,
	maxPerMinute *int,
	autoDedupe *bool,
	urgencyScore *float64,
	executeAtOpt *string,
	cronOpt *string,
	webhookUrl *string,
	maxExecutionSeconds *int,
) (*SnerdTask, error)

NewSnerdTaskAdvanced creates a new task with advanced parameters

func (*SnerdTask) Execute 

func (t *SnerdTask) Execute(ctx context.Context) error

Execute runs the task by invoking the registered handler

func (*SnerdTask) GetMaxRetries 

func (t *SnerdTask) GetMaxRetries() int

GetMaxRetries returns the maximum retry count

func (*SnerdTask) GetRetryAfterHours 

func (t *SnerdTask) GetRetryAfterHours() float64

GetRetryAfterHours returns the retry interval in hours

func (*SnerdTask) GetRetryAfterTime 

func (t *SnerdTask) GetRetryAfterTime() time.Time

GetRetryAfterTime returns the time when the task should be retried

func (*SnerdTask) GetRetryCount 

func (t *SnerdTask) GetRetryCount() int

GetRetryCount returns the current retry count

func (*SnerdTask) GetTaskID 

func (t *SnerdTask) GetTaskID() string

GetTaskID returns the task ID

func (*SnerdTask) OnMaxRetryReached 

func (t *SnerdTask) OnMaxRetryReached(ctx context.Context, contextProvider func() interface{}) error

func (*SnerdTask) String 

func (t *SnerdTask) String() string

func (*SnerdTask) ToRetryableTask 

func (t *SnerdTask) ToRetryableTask() *RetryableTask

ToRetryableTask wraps the SnerdTask in a RetryableTask for storage compatibility

func (*SnerdTask) UpdateRetryConfig 

func (t *SnerdTask) UpdateRetryConfig(errorObj error)

UpdateRetryConfig updates the retry configuration after a failed execution

type Task 

type Task interface {
	// GetTaskID returns the unique identifier for the task.
	GetTaskID() string
	// GetRetryCount returns the number of times this task has been retried.
	GetRetryCount() int
	// Execute runs the task's logic. Return an error if the task fails and should be retried.
	Execute(ctx context.Context) error
}

Task represents a unit of work that can be processed by the queue system.

type TaskFactory 

type TaskFactory func(id string, data string) (Task, error)

TaskFactory creates a Task from its stored data. The factory function is responsible for reconstructing a Task instance, including unmarshaling any stored data.

func CreateTaskFactoryWithDecoder 

func CreateTaskFactoryWithDecoder[P any](creator func(taskID string, payload P) Task) TaskFactory

CreateTaskFactoryWithDecoder creates a factory function that can reconstruct tasks from stored data This helps client code avoid having to deal with marshaling/unmarshaling

type TaskHandler 

type TaskHandler func(ctx context.Context, parameters string) error

TaskHandler is a function that processes parameters to execute a task

type TaskWithData 

type TaskWithData interface {
	Task
	// GetTaskType returns a unique identifier for this task type.
	// This is used for debugging and monitoring, not for type-based dispatch.
	GetTaskType() string
	// MarshalData serializes the task data to JSON.
	MarshalData() ([]byte, error)
	// UnmarshalData deserializes the task data from JSON.
	UnmarshalData([]byte) error
	// Clone creates a new instance of this task with the same type but no data.
	// This will be populated via UnmarshalData when reconstructing tasks.
	Clone() TaskWithData
}

TaskWithData extends Task to support saving and retrieving task-specific data. Implement this interface if your task needs to persist additional fields.

type TaskWithMaxRetryCallback 

type TaskWithMaxRetryCallback interface {
	// OnMaxRetryReached is called when the task reaches its maximum retry count.
	OnMaxRetryReached(ctx context.Context, contextProvider func() interface{}) error
}

TaskWithMaxRetryCallback allows a task to handle the case where it has reached its maximum number of retries.

Source Files

View all Source files

Directories

Path Synopsis
cmd
dashboarddemo command

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.