README
¶
task
Manage a group of long-running background tasks that all stop when any one of them stops.
Stability
v1.x releases make no breaking changes to exported APIs. New functionality may be added in minor releases; patches are bug fixes, or administrative work only.
Installation
Go 1.26.1 or later.
go get github.com/wood-jp/task
Task interface
Any type that satisfies the Task interface can be managed by a Manager:
type Task interface {
Run(context.Context) error
Name() string
}
Run should block until the context is cancelled or the task can no longer continue. Run must return nil when the context is cancelled — a non-nil error is treated as a failure and causes the manager to cancel all other tasks. In particular, do not return ctx.Err(): context.Canceled is a non-nil error and will be treated as a failure. Name provides a human-friendly label used in log output.
Manager
Manager runs a group of tasks concurrently. When any task stops, whether due to an error or a clean exit, the manager cancels the shared context so all other tasks know to stop. Ephemeral tasks (registered via RunEphemeral) are the exception: a clean exit from an ephemeral task does not trigger shutdown.
m := task.NewManager(
task.WithLogger(logger),
)
if err := m.Run(taskA, taskB, taskC); err != nil {
// ErrManagerStopped if the manager has already stopped
}
if err := m.Wait(); err != nil {
// first task error, plus any cleanup errors attached as CleanupErrors
}
Running tasks
Run starts one or more tasks immediately. The tasks share the manager's internal context. If any task returns an error, the context is cancelled and the error is propagated through Wait.
m.Run(taskA, taskB)
Ephemeral tasks
RunEphemeral starts tasks that are expected to finish on their own without triggering shutdown of the rest of the group. Unlike Run, a clean exit from an ephemeral task does not cancel the manager context.
m.RunEphemeral(migrationTask)
Cleanup
Cleanup registers a function to run after all tasks have stopped, similar to defer. Functions are called in reverse registration order. Errors are collected and attached to the Wait return value as CleanupErrors; retrieve them with xerrors.Extract:
m.Cleanup(db.Close)
m.Cleanup(cache.Flush)
if err := m.Wait(); err != nil {
if cleanupErrs, ok := xerrors.Extract[task.CleanupErrors](err); ok {
for _, ce := range cleanupErrs {
logger.Error("cleanup failed", slog.Any("err", ce))
}
}
}
Waiting and stopping
Wait blocks until all tasks finish and all cleanup functions have run. Repeated or concurrent calls all return the same result.
err := m.Wait()
Stop cancels the context immediately and then calls Wait:
err := m.Stop()
Timeout behaviour:
| Situation | Error returned |
|---|---|
| Tasks do not stop within the shutdown timeout | ErrShutdownTimeout |
| Cleanup functions do not complete within the cleanup timeout | ErrCleanupTimeout |
| One or more cleanup functions returned an error | ErrCleanupFailed (base) with CleanupErrors attached |
Options
| Option | Default | Description |
|---|---|---|
WithLogger(logger) |
discard | Logger for task start/stop/error events |
WithContext(ctx) |
context.Background() |
Parent context; cancellation triggers shutdown |
WithShutdownTimeout(d) |
30s | How long to wait for tasks to stop after cancellation |
WithCleanupTimeout(d) |
10s | Total time budget for all cleanup functions |
Guard
Guard prevents a task's Run method from being called more than once. Embed it in any Task struct and call TryStart at the top of Run:
type MyTask struct {
task.Guard
// ...
}
func (t *MyTask) Run(ctx context.Context) error {
if err := t.Guard.TryStart(); err != nil {
return err // returns task.ErrAlreadyStarted on the second call
}
// ...
}
TryStart returns nil on the first call and a wrapped ErrAlreadyStarted on all subsequent calls. ErrAlreadyStarted is a sentinel error in the root task package and can be tested with errors.Is.
Subpackages
ossignal
github.com/wood-jp/task/ossignal
A Task implementation that listens for OS signals and returns nil when one is received, triggering an orderly shutdown of the manager.
Signal capture begins at construction time, so no signals are missed between NewTask and Run. Note however that only the first caught signal will trigger the task termination, even if captured before Run is called. In such a case Run will return immediately.
sig := ossignal.NewTask(
ossignal.WithLogger(logger),
)
m := task.NewManager(task.WithLogger(logger))
m.Run(sig, taskA, taskB)
if err := m.Wait(); err != nil {
log.Fatal(err)
}
By default, ossignal.NewTask listens for SIGINT, SIGTERM, and SIGQUIT. Override with WithSignals:
sig := ossignal.NewTask(
ossignal.WithSignals(syscall.SIGUSR1),
)
Other options:
| Option | Default | Description |
|---|---|---|
WithLogger(logger) |
discard | Logger for signal receipt |
WithSignalLogLevel(level) |
slog.LevelInfo |
Log level used when a signal is received |
WithSignals(signals...) |
SIGINT, SIGTERM, SIGQUIT | Signals to listen for |
WithOnSignal(fn) |
none | Callback invoked after the signal is logged |
Action-based subpackages
loop, poll, and sigtrigger are three variations of the same idea: wrap a task.Action in a Task that calls it repeatedly under different triggering conditions. The action signature is the same in all three:
func(ctx context.Context) error
The only difference is what causes the action to fire:
| Package | Trigger |
|---|---|
loop |
The previous action completed (completion-based) |
poll |
A clock tick fired (time-based) |
sigtrigger |
An OS signal was received (event-based) |
All three share the same error behaviour: by default an action error propagates and triggers manager shutdown; WithContinueOnError logs the error and keeps running instead.
loop
github.com/wood-jp/task/loop
Calls the action in a tight loop: as soon as one call returns nil, the next begins (after an optional delay). Use this for work that should run continuously, restarting immediately after each completion.
worker := loop.NewTask(
func(ctx context.Context) error {
return processNextBatch(ctx, db)
},
"worker",
loop.WithLogger(logger),
)
m := task.NewManager(task.WithLogger(logger))
m.Run(sig, worker)
if err := m.Wait(); err != nil {
log.Fatal(err)
}
The delay between runs is measured from the completion of one call to the start of the next, so calls never overlap:
worker := loop.NewTask(action, "worker",
loop.WithDelay(5*time.Second), // sleep between runs
loop.WithInitialDelay(), // also sleep before the first run
)
Options:
| Option | Default | Description |
|---|---|---|
WithLogger(logger) |
discard | Logger for per-run start/complete events |
WithDelay(d) |
0 | Sleep between runs (context-aware) |
WithInitialDelay() |
false | Also apply the delay before the first run |
WithContinueOnError() |
false | Log action errors and keep looping instead of propagating |
poll
github.com/wood-jp/task/poll
Calls the action on a fixed clock interval using a ticker. The interval is time-based: ticks fire regardless of how long the action takes. If the action takes longer than the interval, the next tick fires immediately after it completes (Go's ticker coalesces missed ticks).
poller := poll.NewTask(
func(ctx context.Context) error {
return syncState(ctx, db)
},
"state-sync",
30*time.Second,
poll.WithLogger(logger),
)
m := task.NewManager(task.WithLogger(logger))
m.Run(sig, poller)
if err := m.Wait(); err != nil {
log.Fatal(err)
}
Use WithRunAtStart to call the action immediately when Run is called, before the first tick:
poller := poll.NewTask(action, "state-sync", 30*time.Second,
poll.WithRunAtStart(),
)
Options:
| Option | Default | Description |
|---|---|---|
WithLogger(logger) |
discard | Logger used when WithContinueOnError is active |
WithRunAtStart() |
false | Call the action immediately before the first tick |
WithContinueOnError() |
false | Log action errors and keep ticking instead of propagating |
sigtrigger
github.com/wood-jp/task/sigtrigger
Calls the action each time a configured OS signal is received. Unlike ossignal, which exits on the first signal, sigtrigger stays alive and re-runs the action on every delivery. Signal capture begins at construction time, so no signals are missed between NewTask and Run.
The signal channel has a buffer of one. A signal received while the action is running queues up and triggers another run immediately after the current one completes. Additional signals beyond that one are dropped (standard Go signal delivery behaviour).
trig := sigtrigger.NewTask(
func(ctx context.Context) error {
return reloadConfig(ctx)
},
sigtrigger.WithLogger(logger),
)
m := task.NewManager(task.WithLogger(logger))
m.Run(sig, trig, server)
if err := m.Wait(); err != nil {
log.Fatal(err)
}
By default, sigtrigger.NewTask listens for SIGHUP. Override with WithSignals:
trig := sigtrigger.NewTask(action,
sigtrigger.WithSignals(syscall.SIGUSR1),
)
Options:
| Option | Default | Description |
|---|---|---|
WithLogger(logger) |
discard | Logger for signal receipt and (with WithContinueOnError) action errors |
WithSignals(signals...) |
SIGHUP | Signals to listen for |
WithContinueOnError() |
false | Log action errors and keep running instead of propagating |
Contributing
See CONTRIBUTING.md.
Security
See SECURITY.md.
Attribution
This library is a simplified fork of one written by wood-jp at Zircuit. The original code is available here: zkr-go-common-public/task
Documentation
¶
Overview ¶
Package task provides wrappers for simplified management of async functions.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ErrAlreadyStarted = errors.New("task already started")
ErrAlreadyStarted is returned by any task's Run method if called more than once.
var ErrCleanupFailed = errors.New("one or more cleanup functions failed")
ErrCleanupFailed is used as the base error when cleanup functions return errors but no task error occurred.
var ErrCleanupTimeout = errors.New("cleanup timed out")
ErrCleanupTimeout is returned by Wait when cleanup functions do not complete within the cleanup timeout. If tasks also failed, the task error is provided as the base and cleanup errors are attached; see CleanupErrors.
var ErrManagerStopped = errors.New("manager already stopped")
ErrManagerStopped is returned when Run or RunEphemeral is called after the manager has stopped.
var ErrShutdownTimeout = errors.New("shutdown timed out waiting for tasks to stop")
ErrShutdownTimeout is returned by Wait when tasks do not stop within the shutdown timeout.
Functions ¶
This section is empty.
Types ¶
type Action ¶
Action is a function that performs a unit of work. It must return nil when the context is cancelled.
type CleanupErrors ¶
type CleanupErrors []error
CleanupErrors holds errors returned by registered cleanup functions. Retrieve it from a Wait error using xerrors.ExtractCleanupErrors.
func (CleanupErrors) LogValue ¶
func (e CleanupErrors) LogValue() slog.Value
LogValue implements slog.LogValuer.
type Guard ¶
type Guard struct {
// contains filtered or unexported fields
}
Guard prevents a task's Run method from being called more than once. Embed it in a Task struct and call TryStart at the top of Run.
Example ¶
ExampleGuard demonstrates the Guard's double-run prevention.
package main
import (
"errors"
"fmt"
"github.com/wood-jp/task"
)
func main() {
var g task.Guard
fmt.Println(g.TryStart() == nil)
fmt.Println(errors.Is(g.TryStart(), task.ErrAlreadyStarted))
}
Output: true true
func (*Guard) TryStart ¶
TryStart returns nil on the first call and a wrapped ErrAlreadyStarted on all subsequent calls.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager manages a group of tasks that should all stop when any one of them stops.
Example ¶
ExampleManager demonstrates creating a Manager, running a task, and stopping cleanly.
package main
import (
"context"
"log"
"github.com/wood-jp/task"
)
// contextTask is a Task that blocks until its context is cancelled.
type contextTask struct{ name string }
func (t *contextTask) Name() string { return t.name }
func (t *contextTask) Run(ctx context.Context) error {
<-ctx.Done()
return nil
}
func main() {
m := task.NewManager()
if err := m.Run(&contextTask{name: "worker"}); err != nil {
log.Fatal(err)
}
// Stop cancels all tasks and waits for them to finish.
if err := m.Stop(); err != nil {
log.Fatal(err)
}
}
Output:
func (*Manager) Cleanup ¶
Cleanup registers a function that runs after all tasks are stopped. Similar to defer, cleanup functions are executed in the reverse order in which they were registered. Any errors returned are collected and attached to the Wait return value as CleanupErrors.
Example ¶
ExampleManager_Cleanup demonstrates registering a cleanup function that runs after all tasks have stopped.
package main
import (
"context"
"fmt"
"log"
"github.com/wood-jp/task"
)
// contextTask is a Task that blocks until its context is cancelled.
type contextTask struct{ name string }
func (t *contextTask) Name() string { return t.name }
func (t *contextTask) Run(ctx context.Context) error {
<-ctx.Done()
return nil
}
func main() {
m := task.NewManager()
m.Cleanup(func() error {
fmt.Println("cleanup: closing database")
return nil
})
if err := m.Run(&contextTask{name: "worker"}); err != nil {
log.Fatal(err)
}
_ = m.Stop()
}
Output: cleanup: closing database
func (*Manager) Run ¶
Run immediately starts all of the given tasks. Returns ErrManagerStopped if the manager has already stopped.
func (*Manager) RunEphemeral ¶
RunEphemeral immediately starts all of the given tasks. These tasks are expected to terminate without error while others continue running. Returns ErrManagerStopped if the manager has already stopped.
func (*Manager) Stop ¶
Stop cancels the context immediately and waits for all running tasks to complete.
func (*Manager) Wait ¶
Wait blocks until all tasks are complete, then executes all registered cleanup functions. It returns the first encountered task error, with any cleanup errors attached as CleanupErrors (retrieve via xerrors.Extract). If tasks do not stop within the shutdown timeout, Wait returns ErrShutdownTimeout. If cleanup functions do not complete within the cleanup timeout, Wait returns ErrCleanupTimeout. Concurrent or repeated calls all return the same result.
type Option ¶
type Option func(options *options)
Option is an option func for NewManager.
func WithCleanupTimeout ¶
WithCleanupTimeout sets the maximum total duration for all cleanup functions to complete. Defaults to 10 seconds.
func WithContext ¶
WithContext sets a parent context for the manager. When the parent context is cancelled, the manager will begin shutting down.
func WithLogger ¶
WithLogger sets the logger to be used.
func WithShutdownTimeout ¶
WithShutdownTimeout sets the maximum duration to wait for tasks to stop after the manager context is cancelled. Defaults to 30 seconds.
type Task ¶
type Task interface {
// Run must execute the work of this service and block until the context
// is cancelled, or until the service is unable to continue due to an error.
// Run must return nil when the context is cancelled; a non-nil error signals
// a failure and will cause the manager to cancel all other tasks. In
// particular, do not return ctx.Err(): context.Canceled is a non-nil error
// and will be treated as a failure.
Run(context.Context) error
// Name provides a human-friendly name for use in logging.
Name() string
}
Task represents a background service.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
examples
|
|
|
graceful-shutdown
command
Package main demonstrates graceful shutdown using the task package.
|
Package main demonstrates graceful shutdown using the task package. |
|
Package loop provides a Task that repeatedly executes an action function.
|
Package loop provides a Task that repeatedly executes an action function. |
|
Package ossignal provides a Task that listens for signals from the operating system.
|
Package ossignal provides a Task that listens for signals from the operating system. |
|
Package poll provides a Task that periodically executes an action function.
|
Package poll provides a Task that periodically executes an action function. |
|
Package sigtrigger provides a Task that executes an action each time a configured OS signal is received.
|
Package sigtrigger provides a Task that executes an action each time a configured OS signal is received. |