Documentation
¶
Overview ¶
Package worker implements the claim→heartbeat→execute→complete loop.
Example ¶
Embedding go-taskqueue in a Go binary: open a store, register executors, run a worker pool, enqueue work. Ctrl-C drains gracefully.
package main
import (
"context"
"log"
"log/slog"
"os"
"os/signal"
"github.com/larsartmann/go-taskqueue/internal/executor"
"github.com/larsartmann/go-taskqueue/internal/queue"
"github.com/larsartmann/go-taskqueue/internal/task"
"github.com/larsartmann/go-taskqueue/internal/worker"
)
func main() {
store, err := queue.OpenSQLite("tasks.db")
if err != nil {
log.Fatal(err)
}
defer store.Close()
reg := executor.NewRegistry()
reg.Register("sh", executor.NewCommandExecutor(""))
pool := worker.New(store, worker.Config{
Owner: "my-app",
Concurrency: 4,
Executors: reg,
}, slog.Default())
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()
if _, err := queue.New(store).Enqueue(ctx, task.New{
Project: "demo",
Type: "sh",
Payload: []byte(`{"cmd":"go test ./..."}`),
}); err != nil {
log.Fatal(err)
}
if err := pool.Start(ctx); err != nil {
log.Fatal(err)
}
}
Output:
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ExpBackoff ¶
ExpBackoff returns an exponential retry delay for the given attempt number.
Types ¶
type Config ¶
type Config struct {
Owner string // lease owner identity (default: auto)
Concurrency int // parallel task executions (default 2)
PollInterval time.Duration // idle poll gap (default 250ms)
Lease time.Duration // claim lease length (default 2m)
Heartbeat time.Duration // heartbeat cadence, < Lease (default 30s)
TaskTimeout time.Duration // per-task execution cap (default 10m)
Executors *executor.Registry
Backoff func(attempt int) time.Duration // retry backoff (default exp)
// PreflightBackoff delays re-claiming after a preflight refusal
// (dirty repo, missing autonomy config). Default 2m. No attempt is
// burned by preflight refusals.
PreflightBackoff time.Duration
}
Config controls one worker pool.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool runs N concurrent claim-execute loops against a store.
func (*Pool) Start ¶
Start launches the pool; it blocks the caller until ctx is cancelled, then drains in-flight tasks and returns. Tasks execute under a context that is NOT cancelled by pool shutdown (bounded only by TaskTimeout), so a graceful stop lets agents finish; terminal store writes use that same uncancellable context so a cancelled parent cannot orphan a running task's outcome. Stop waiting with a second Ctrl-C (SIGKILL) if truly urgent.
Source Files