Help us shape the future of Go! Take the 2026 Go Developer Survey

work

package
v0.14.0 Latest Latest
Warning

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

Go to latest
Published: Oct 23, 2026 License: AGPL-3.0 Imports: 4 Imported by: 0

Documentation

Overview

Package work runs work held in Postgres on a bounded number of goroutines (issue #72, O-32).

The work itself — a deploy, a detection — is a row. A replica claims rows with FOR UPDATE SKIP LOCKED when it has room and runs each on a goroutine of its own, never more at once than its limit; every replica does the same, so the install's throughput grows with its replicas and none of them takes on more than it can run. A row nobody has claimed waits for the first replica with room, which is what lets queued work survive a restart.

This is not scheduling (R-010): nothing here decides where an app runs. It decides which of Pando's own processes does a piece of Pando's own work.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Each

func Each[T any](ctx context.Context, n int, items []T, fn func(ctx context.Context, item T))

Each calls fn for every item, at most n at a time, and returns when all have returned. For a leader job's pass over many apps — rolling backups, the auto-deploy poll — where one slow app must not hold up the rest and every app at once would swamp the host (issue #72).

An item not yet started when ctx ends is skipped.

Types

type Pool

type Pool[T any] struct {
	// Name is what the pool is called in log lines.
	Name string

	// Limit is the most items this replica runs at once. At least one.
	Limit int

	// LimitFunc, when set, is read instead of Limit every time the pool looks
	// for work, so the limit can change while the pool runs (issue #93). A
	// lower limit takes effect as running items finish: nothing is stopped.
	LimitFunc func(ctx context.Context) int

	// Poll is how often the pool looks for work when it was not told there is
	// some (Kick). Work queued on another replica is found this way. Two
	// seconds when zero.
	Poll time.Duration

	// Claim takes up to n items for this replica.
	Claim func(ctx context.Context, n int) ([]T, error)

	// Run runs one item to completion. Its context ends when the pool stops,
	// and Run must return soon after: a deploy's adapters honor cancellation.
	Run func(ctx context.Context, item T)

	// Release puts back items whose Run was still going when the pool
	// stopped, so another replica takes them at once rather than after this
	// one is taken for dead. An item that finished in the meantime must be
	// left alone; the stores' Release only touches work still in flight.
	Release func(ctx context.Context, items []T)

	// Drain is how long a stopping pool waits for its items to return before
	// releasing them anyway. Ten seconds when zero.
	Drain time.Duration

	Logger *zap.Logger
	// contains filtered or unexported fields
}

Pool claims and runs items of type T.

func (*Pool[T]) Kick

func (p *Pool[T]) Kick()

Kick tells the pool there may be work, so it claims now rather than at its next poll. Never blocks.

func (*Pool[T]) Running

func (p *Pool[T]) Running() int

Running is how many items this replica is running now.

func (*Pool[T]) Serve

func (p *Pool[T]) Serve(ctx context.Context)

Serve claims and runs work until ctx ends, then stops what is running, waits up to Drain for it, and releases what did not finish.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL