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/queue/sqlite"
"github.com/larsartmann/go-taskqueue/internal/task"
"github.com/larsartmann/go-taskqueue/internal/worker"
)
func main() {
store, err := sqlite.Open("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 ¶
const EnvStreakCode = "env-streak"
EnvStreakCode marks breaker-burned attempts in the journal: a task.failed fact whose reason carries this code was killed by the environmental-requeue circuit breaker (sustained preflight/gate refusals), not by a judged execution failure.
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
// EnvRequeueBurn bounds the environmental-requeue ladder (R3 option
// (a), docs/planning/2026-10-01_11-19_env-requeue-circuit-breaker.md):
// after this many consecutive preflight/gate refusals the attempt
// burns (store.Fail, exponential NotBefore escalation) instead of
// requeueing again, so a sustained-sick gate cannot churn claims
// forever. Default 3; negative disables (pure requeue behavior).
EnvRequeueBurn int
// Budget is the claim-time spend gate: before a claimed task's
// executor runs, the hook decides whether the pool may start another
// paid turn at all. A blocked task is requeued WITHOUT burning an
// attempt, parked until the returned delay passes (a daily cap
// resolves at local midnight), so work enqueued before the cap bit
// cannot burn money after it. The reason rides the task.requeued
// fact with requeue class "budget". Nil = ungated (historical
// behavior); budget refusals never ride the environmental
// streak/burn ladder — an exhausted budget is policy, not a sick
// environment.
Budget func(ctx context.Context) (blocked bool, reason string, retryIn time.Duration)
}
Config controls one worker pool.
Example (Budget) ¶
This example arms the claim-time budget gate: the Budget hook runs BEFORE any executor spawns, so an agent pool never spends a paid turn once the day's cap is spent. A blocked claim requeues WITHOUT burning an attempt and parks until the hook's delay (midnight for a daily cap), so work enqueued before the cap bit stays queued instead of costing money after it.
package main
import (
"context"
"fmt"
"time"
)
func main() {
countEnqueuedToday := func(ctx context.Context) (int, error) {
// Production: store.CountFacts(ctx, journal.Enqueued, localMidnight)
return 3, nil //nolint:goerr113 // example stub
}
guard := func(ctx context.Context) (blocked bool, reason string, retryIn time.Duration) {
const dailyCap = 40
spent, err := countEnqueuedToday(ctx)
if err != nil {
return false, "", 0 // fail open: the claim gate is a cost brake, not a correctness gate
}
if spent < dailyCap {
return false, "", 0
}
return true, "daily agent budget spent", time.Until(time.Now().AddDate(0, 0, 1).Truncate(24 * time.Hour))
}
_ = guard // hand to worker.Config.Budget
fmt.Println("paid turns stop at the cap; attempts never burn")
}
Output: paid turns stop at the cap; attempts never burn
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