Skip to content

readme.md

Latest commit

 

History

History
250 lines (191 loc) · 8.25 KB

readme.md

File metadata and controls

250 lines (191 loc) · 8.25 KB
Raw

backstage

A standalone, zero-external-dependency Go package for managing background jobs and scheduled tasks.

Features

  • Named queues — worker pools backed by a pluggable Store; ships with an in-memory buffered channel, drop in your own for durable persistence
  • Schedules — Every, Daily, and standard 5-field Cron expressions, all implemented in-house (no external cron library)
  • Scheduled tasks — fire directly in their own goroutine or dispatch onto a named queue
  • Retries — per-job retry count and optional delay; panics are never retried
  • Job IDs — optional identifier for deduplication and idempotent dispatch
  • Structured logging — optional *slog.Logger; falls back to slog.Default(). All log records carry a backstage key for easy filtering
  • Graceful shutdown — configurable drain timeout; in-flight jobs finish before Serve returns

Quick start

import "lowbit.dev/backstage"

// Create a supervisor (uses slog.Default() for logging).
sv := backstage.New("myapp")

// Register a named worker pool.
sv.RegisterQueue("default", backstage.QueueConfig{
    Workers: 2,
    Buffer:  64,
})

// Dispatch a one-off job with retry on transient failure.
if err := sv.Dispatch("default", backstage.Job{
    ID:         "welcome-" + userID, // optional; used by stores for deduplication
    Name:       "send-welcome-email",
    Retries:    3,
    RetryDelay: 5 * time.Second,
    Run: func(ctx context.Context) error {
        return sendWelcome(ctx, userID)
    },
}); err != nil {
    // err is backstage.ErrQueueFull or backstage.ErrQueueNotFound,
    // or a custom error returned by a user-supplied Store.
    log.Println("dispatch failed:", err)
}

// Schedule a recurring task on the queue.
sv.ScheduleOnQueue("default", backstage.MustCron("0 3 * * *"), backstage.Job{
    Name: "archive-expired-events",
    Run:  archiveExpiredEvents,
})

// Schedule a task that runs directly (no queue).
sv.Schedule(backstage.Every(10*time.Minute), backstage.Job{
    Name: "refresh-cache",
    Run:  refreshCache,
})

// Run until the context is cancelled.
sv.Serve(ctx)

Jobs

type Job struct {
    ID         string        // optional; logged and passed to the Store for deduplication
    Name       string        // human-readable label for log output
    Run        func(ctx context.Context) error
    Retries    int           // additional attempts after the first failure; 0 = try once
    RetryDelay time.Duration // pause between attempts; 0 = retry immediately
}

Retries apply to errors returned by Run. A job that panics is not retried — a panic is a programmer error and the honest response is to stop, log it, and move on. The context passed to Run is the supervisor's context; if it is cancelled during a retry delay the job stops without further attempts.

When ID is set it appears as job_id in every log record for that execution. Custom Store implementations receive the full Job value on every Push, so they can use ID for deduplication or to enforce idempotency before persisting.

Queues

// Default: in-memory store, 100-slot buffer, 3 workers.
sv.RegisterQueue("emails", backstage.QueueConfig{
    Workers: 3,
    Buffer:  100,
})

if err := sv.Dispatch("emails", backstage.Job{Name: "...", Run: fn}); err != nil {
    // inspect with errors.Is(err, backstage.ErrQueueFull)
}

Dispatch returns an error when the job cannot be accepted. Two sentinel values are defined:

Error Meaning
ErrQueueNotFound No queue with that name has been registered
ErrQueueFull The store rejected the job (in-memory buffer full, or custom store at capacity)

When an error occurs a WARN log record is emitted with queue, job, queue_depth, and error attributes before the error is returned to the caller.

Custom stores

By default jobs live in an in-memory buffered channel and are lost if the process restarts. For work that must survive a restart — billing events, audit records, anything with delivery guarantees — implement the Store interface and pass it via QueueConfig.Store:

type Store interface {
    Push(job Job) error              // non-blocking; return ErrQueueFull when at capacity
    Pop(ctx context.Context) (Job, error) // block until a job is available or ctx is cancelled
    Len() int
}
sv.RegisterQueue("billing", backstage.QueueConfig{
    Workers: 2,
    Store:   mydb.NewJobStore(db, "billing"),
})

The Buffer field is ignored when Store is set. Workers call Pop in a blocking loop; cancelling the supervisor context causes Pop to return and the worker to exit cleanly. The store implementation is responsible for its own concurrency safety.

Schedules

Every

backstage.Every(15 * time.Minute)

Fires at a fixed interval measured from when the task fires, not when it finishes. If a job runs longer than the interval, the next invocation starts while the previous one is still in progress. For tasks where concurrent execution is unsafe, either use a queue with Workers: 1, or guard the implementation with a mutex or sync/atomic flag.

Daily

backstage.Daily(3, 0, time.UTC)  // 03:00 UTC every day

Fires once per day at the given hour:minute in the provided *time.Location. Pass nil to use time.UTC.

Cron

schedule, err := backstage.Cron("0 3 * * *") // 03:00 UTC every day
schedule     := backstage.MustCron("*/5 * * * *") // every 5 minutes

Standard 5-field cron: minute hour day-of-month month day-of-week.

Supported syntax per field:

Syntax Meaning
* every value
N specific value
N-M inclusive range
*/N every Nth value across the full range
N-M/N every Nth value within range N–M
N,M comma-separated list of any of above

All cron times are computed in UTC. Day-of-month and day-of-week use AND semantics: both fields must match for a time slot to fire.

MustCron panics on an invalid expression. It is intended for package-level var declarations where the expression is a compile-time constant — a bad literal is a programmer error, and panic is the honest signal. Do not pass user-supplied or config-sourced strings to MustCron; use Cron and handle the error instead.

Logging

// Default — uses slog.Default()
sv := backstage.New("myapp")

// Custom logger
sv := backstage.New("myapp", backstage.WithLogger(
    slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level: slog.LevelDebug,
    })),
))

Every log record carries backstage: "<name>" so output from multiple supervisor instances is always identifiable. Queue-level records additionally carry queue: "<name>".

Level Events
INFO Supervisor start/stop, queue registered, task scheduled, shutdown
DEBUG Worker start/stop, job picked up/finished (with duration), scheduler tick, next fire time
WARN Job dropped (queue full), job returned a non-nil error
ERROR Job panicked (recovered), dispatch to unknown queue

Options

backstage.WithLogger(l *slog.Logger)        // custom logger
backstage.WithDrainTimeout(d time.Duration) // default: 30s

Graceful shutdown

When the context passed to Serve is cancelled:

  1. All worker goroutines are signalled to stop — they finish their current job and exit; no new jobs are picked up from the queue buffer.
  2. The supervisor waits up to DrainTimeout for all workers to exit.
  3. Serve returns nil.

Jobs should respect ctx.Done() for long-running work to enable clean shutdown within the drain window.