conveyor

README

The lib proposes a new way of building pipelines in Go orthogonal to classical goroutines + channels approach. You code the path of your data item through the steps instead of coding loops that execute the same steps for all items.

It's optimized and well suited for batch processing of messages from brokers (Kafka, RabbitMQ, etc.) but can be used for any pull-based processing.

go get github.com/cardinalby/go-conveyor

Concept

Basic example

Suppose we have a typical 3-step pipeline that:

• read stage: consumes messages from Kafka / RabbitMQ and collects them in a batch
• write stage: writes some data from the batch to a database
• commit stage: acknowledges the message / commits offsets on the broker

1. A gopher who doesn't know about pipelines would

Spin 1 goroutine with fn in loop that executes steps one by one:

// (1) read messages
// (2) write to DB                    (nobody reads messages now - underutilized)
// (3) commit offsets / ack messages  (nobody writes to DB now - underutilized)

• fn is responsible for processing 1 batch of messages (1 "item")
• Easy to read and write, but:
  • ⚠️ While one of the steps is busy, the other steps are idle and resources are underutilized.

2. A good gopher would:

• spin 3 goroutines with 3 functions, each:
  • is responsible for one stage
  • loops over the input
• connect goroutines with channels between stages
• don't forget to provide the proper shutdown logic:
  • wait for all goroutines to finish
  • close the channels in the right order

The picture shows the saturated pipeline:

• batch1 reached "commit" stage
• batch2 reached "write" stage
• batch3 entered "read" stage

3. A "conveyor" user would:

• create Conveyor, define stages (steps)
• modify fn (from 1st example) by adding stage.MoveTo calls between stages
• delegate the dispatching, goroutines and shutdown management to the conveyor using this single "ItemProcessor" fn

c := conveyor.NewConveyor()
// "read" stage is implicitly created with items limit = 1
writeStage := c.AddStage()
commitStage := c.AddStage()

itemProcessor := func(ctx context.Context) error {
    // itemProcessor is started only once "read" stage is available 
    
    // (1) read messages:
    // batch := ...
    
    // Atomic:
    // - wait and enter "write" stage (once available) 
    // - exit "read" stage making it available for the next batch
    if err := writeStage.MoveTo(ctx); err != nil {
        // happens on ctx cancellation or conveyor shutdown
        return err 
    }
    
    // (2) write to DB or return an error
    // for item := range batch { ... }
    
    // - enter "commit", 
    // - release "write" stage
    if err := commitStage.MoveTo(ctx); err != nil {
        return err
    }
    
    // (3) commit offsets / ack messages or return an error:
    // broker.Ack(batch)
    
    return nil
    // Conveyor can reuse the goroutine for the next processor call
}

// - starts 1st itemProcessor immediately
// - starts new processors once "read" stage is released
shutdownErr := c.Run(ctx, itemProcessor)

Stages are smart locks with ordered admission:

• MoveTo is atomic: it waits for the stage to be available and releases the previous stage providing backpressure
• Deadlock-free: you can move only forward to the next stage, not backward

You still have a single function that processes a single batch, but the Conveyor takes care of:

• dispatching new items (batches) to the next available goroutine from the dynamic pool
• graceful shutdown of the pipeline
• observability, dynamic concurrency limits and queue sizes for stages, etc.

Pros

1. Code the path of a single batch ("item") through the stages:
   • easier to write / read / debug / trace / reason about
   • less wiring (no channels, no goroutines, no loops in user code)
   • easy to upgrade from existing "1-goroutine" version (just add MoveTo calls)
2. If "stage1" produces some data for "stage3", use it directly at "stage3":
   • you don't need to pass it through "stage2" like you would do in channels
   • you are inside a single function for a single item/batch and keep the batch processing state
3. You can skip some stages with natural control flow (if/else, return):
   • with channels, you would need to send a "skip" signal through the channels
   • with channels, it's complicated to send data from "stage1" to "stage3" if "stage2" is a fan-out stage
4. Supports deadlock-free scatter-gather stages whose work may grow while it runs: schedule in rounds, or let a task schedule its own follow-ups
   • SetBackpressure chooses when an item entering a fan-out releases the previous stage: on entry, at the first task start, or once every branch of its initial work has started a task (or ran out of tasks)
5. Dynamic concurrency limits and queue sizes for stages (gracefully adjusted in runtime)
6. Goroutines pool is dynamically adjusted (goroutines are reused for new items)
7. Graceful shutdown happens naturally, just respect the context:
   • If item N fails:
     • the conveyor stops spawning new items
     • items with number > N receive a context cancellation error and can't enter any stage
     • all items that are already in the pipeline are allowed to finish
   • If the context passed to c.Run() is canceled:
     • all items that are already in the pipeline are allowed to finish
     • new items are not spawned
   • Force shutdown support
8. Pull-based observability

Cons

• New paradigm and API to learn
• Not all topologies (possible with channels) can be expressed (windowing, feedback loops, splits without join, etc.)
• Worse performance compared to channels for CPU-bound scenarios
  • The lib is optimized for IO-bound scenarios (especially if each item is a batch of messages)
  • Any significant difference is only observable if your stages take less than 100 microseconds to process an item / batch.

Vocabulary

Operation	Meaning
MoveTo	Advance the item into a node; leaving a fan-out joins its work.
Schedule	Register parallel work for a fan-out, before or after entering it.
FanOut.Wait	Join the work scheduled so far without leaving the fan-out (rounds).
Retain	Let unfinished work keep the current node occupied while the item moves on.
Wave.Wait	Wait for retained work later in the item's path.

Most processors need only MoveTo and Schedule.

Features

See the docs for guides on these features:

• Retain previous stage longer
• Queues
• Shared Stages
• Fan-out (scatter/gather): pools, Schedule / Wait rounds, tasks that spawn follow-ups, Retain and Wave.Wait
• Lanes: when a branch is a pipeline
• Conditional MoveTo
• Observability
• Design FAQ

★ Interactive demo

• Build different conveyor topologies
• Simulate them in browser (against lib's WebAssembly build)

Documentation

Overview

Package conveyor moves items through an ordered series of nodes, keeping their relative order while managing capacity, backpressure, and completion. The path of one item is written as a single function, the ItemProcessor.

Everyday vocabulary:

• MoveTo advances the item into a node.
• Schedule registers parallel work for a fan-out, before or after entering it.
• FanOut.Wait joins the work scheduled so far without leaving the fan-out.
• Retain (on a Stage or a FanOut) lets unfinished work keep the node while the item moves on.
• Wave.Wait waits for retained work later in the item's path.

Most processors need only MoveTo and Schedule. Adaptive rounds add FanOut.Wait; overlapping work across stages adds Retain.

Index

• Variables
• func ItemNoFromContext(ctx context.Context) (no int64, ok bool)
• type AnyUnitOption
• type Branch
• type Conveyor
  • func NewConveyor(options ...Option) Conveyor
• type FanOut
• type FanOutBackpressure
  • func (m FanOutBackpressure) String() string
• type Gauge
• type ItemProcessor
• type Lane
• type OptName
• type Option
  • func OptShutdownContext(factory ShutdownContextFactory) Option
• type Pool
• type ShutdownContextFactory
• type ShutdownError
• type Stage
• type Stats
• type Task
• type TaskFunc
• type Unit
• type UnitOccupants
• type UnitStat
• type Wave

Variables

var (
	// ErrConveyorAlreadyRunning is returned by Run when the conveyor is already running.
	ErrConveyorAlreadyRunning = errors.New("conveyor is already running")

	// ErrInvalidContext classifies a context that cannot drive a node transition. Match it, or one of the two
	// variants below, with errors.Is.
	ErrInvalidContext = errors.New("invalid context")

	// ErrForeignContext indicates the context was never derived from an item's ctx.
	ErrForeignContext = fmt.Errorf("%w: not derived from a conveyor item's ctx", ErrInvalidContext)

	// ErrStaleContext indicates the context's owner has already finished: the item, a lane child whose callback
	// returned, or a pool task whose wave has finished.
	ErrStaleContext = fmt.Errorf("%w: the item or work it belongs to has finished", ErrInvalidContext)
)

Returned errors, inspectable with errors.Is on the value returned by Run and the node methods (MoveTo, TryMoveTo, FanOut.Schedule, FanOut.Wait), or on Wave.Err.

Functions

func ItemNoFromContext

func ItemNoFromContext(ctx context.Context) (no int64, ok bool)

ItemNoFromContext returns the number of the item this context belongs to, for logging and tracing. A child item's work reports the number of the item that scheduled it. ok is false if ctx did not originate from a conveyor item.

Types

type AnyUnitOption

type AnyUnitOption interface {
	// contains filtered or unexported methods
}

AnyUnitOption configures a node at creation (AddStage, AddFanOut, AddPool, AddLane). OptName is the only one; capacity is set separately with SetLimit and SetQueueSize.

type Branch

type Branch interface {
	Unit

	// NewTask creates a task of one callback — one slot's worth of work on this branch. A nil fn panics.
	NewTask(fn TaskFunc) Task

	// NewTasks creates a task of count independent callbacks fn(ctx, 0)..fn(ctx, count-1), each one slot's worth
	// of work on this branch. A count <= 0 yields a no-op task; a nil fn with count > 0 panics.
	NewTasks(count int, fn func(ctx context.Context, index int) error) Task

	// NewTasksGen creates a task whose callbacks are produced by gen, pulled one by one as slots free up.
	NewTasksGen(gen iter.Seq[TaskFunc]) Task

	// NewTasksChan creates a task whose callbacks are received from ch, pulled one by one as slots free up.
	NewTasksChan(ch <-chan TaskFunc) Task
}

Branch is what a FanOut's work is scheduled on: a Pool or a Lane. It is the task-construction surface the two share, for code that builds work without caring which kind it got.

FanOut.Branches returns them in creation order. Type-assert to Pool or Lane for the kind's own methods.

type Conveyor

type Conveyor interface {
	// AddStage adds a Stage to the end of the conveyor, admitting one item at a time by default. Chain SetLimit and
	// SetQueueSize to adjust its capacity, and pass OptName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddStage(opts ...AnyUnitOption) Stage

	// AddFanOut adds a FanOut to the end of the conveyor: a node whose work runs in parallel on branches added with
	// AddPool or AddLane. It admits one item at a time by default; chain SetLimit and SetQueueSize to adjust its
	// capacity, and pass OptName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddFanOut(opts ...AnyUnitOption) FanOut

	// Run starts the conveyor, creating one item per itemProcessor invocation until ctx is canceled or an item
	// fails. It blocks until every in-flight item has finished, then returns the first item error, or else ctx's
	// cancellation cause.
	//
	// Build all nodes before calling Run; the topology is frozen from the first Run on. Run may be called again
	// after it returns, but a concurrent second call returns ErrConveyorAlreadyRunning.
	Run(ctx context.Context, itemProcessor ItemProcessor) error

	// SetItemsLimit caps how many items may be in flight across the whole conveyor at once. One item is one
	// ItemProcessor call, from creation to completion, so this also bounds the number of worker goroutines. A limit <= 0
	// means unlimited, the default. It returns the conveyor for chaining.
	//
	// It is a global bound on top of the nodes' own limits and does not change any of them.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor. Lowering it never evicts an item in
	// flight; it only stops new items from being created until the count has fallen below the new limit.
	SetItemsLimit(n int) Conveyor

	// ItemsLimit returns the current cap on items in flight across the whole conveyor, or 0 if unlimited (the
	// default).
	ItemsLimit() int

	// StartUnit returns the handle of the implicit start stage that paces item creation, for matching it in Stats.
	// It is not a Stage; there is nothing to move to.
	StartUnit() Unit

	// Stats returns a snapshot of the active run's state and resets the gauge windows. Safe to call at any time,
	// from any goroutine; outside a run it reports the zero Stats.
	Stats() Stats

	// DebugUnitOccupants reports, for every unit, exactly which items occupy its body and its waiting room right
	// now. It exists for debugging and visualization; use Stats for production observability.
	//
	// Safe to call at any time, from any goroutine; outside a run it reports nil.
	DebugUnitOccupants() []UnitOccupants
}

Conveyor moves items through an ordered series of nodes, preserving their relative order. Create one with NewConveyor, build the nodes, then call Run with an ItemProcessor. The conveyor runs one ItemProcessor per item, each on its own goroutine, and handles ordering, capacity, and backpressure between nodes.

A node is either a Stage (AddStage), whose code runs inline in the ItemProcessor, or a FanOut (AddFanOut), where the item schedules work onto branches (Pool or Lane) that run it in parallel. An item advances between nodes with MoveTo.

Background work started with Stage.Retain or FanOut.Retain is represented by a Wave: wait for it with Wave.Wait, or read its Finished and Err.

func NewConveyor

func NewConveyor(options ...Option) Conveyor

NewConveyor creates a conveyor with no nodes: add them with AddStage and AddFanOut, then call Run.

type FanOut

type FanOut interface {
	Unit

	// AddPool adds a Pool branch: a single step where a scheduled task runs to completion. Chain SetLimit to run
	// several tasks at once, and pass OptName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddPool(opts ...AnyUnitOption) Pool

	// AddLane adds a Lane branch: a pipeline of its own interior nodes (AddStage, AddFanOut) through which each
	// piece of scheduled work travels as a child item. Pass OptName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddLane(opts ...AnyUnitOption) Lane

	// MoveTo advances the item into this fan-out. It blocks until it is the item's turn and an item slot (SetLimit) is
	// free. Pool capacity does not gate entry.
	//
	// Work prepared with Schedule before entering becomes the item's initial batch and may start before MoveTo
	// returns. Otherwise the item enters with an empty body, and the first Schedule after entry is the initial batch.
	// SetBackpressure decides whether entering releases the previous node at once or keeps its slot until the initial
	// batch makes progress.
	//
	// The item behind cannot enter this fan-out until this item's initial batch is known (it schedules, leaves, or
	// retains); meanwhile it may wait in the fan-out's waiting room.
	//
	// It returns ErrForeignContext, ErrStaleContext, or the item's cancellation cause, whether the cancellation is
	// visible on ctx or only on the item's own context (see ItemProcessor). It panics on misuse: moving backward,
	// re-entering this node, or a node outside the item's own series.
	MoveTo(ctx context.Context) error

	// TryMoveTo is MoveTo without waiting: it enters the fan-out only if it can do so right now, and reports whether
	// it did. It bypasses the waiting room and never jumps an item already waiting there. Busy pools do not make it
	// decline; entering may keep the previous node's slot instead (see SetBackpressure).
	//
	// When entered is false nothing happened: the item stays where it is, and work prepared with Schedule stays
	// prepared for a later attempt or a MoveTo. A canceled item returns (false, its cancellation cause). It panics on
	// the same misuse as MoveTo.
	TryMoveTo(ctx context.Context) (entered bool, err error)

	// Schedule adds tasks to this item's work at this fan-out. It never blocks. Allowed callers:
	//   - the ItemProcessor, inside this fan-out with an open body (not yet left, not retained), or before entering it
	//     while the fan-out is still ahead of the item;
	//   - a task running on one of this fan-out's pools, before the task returns;
	//   - a child item of one of this fan-out's lanes, before its callback returns.
	//
	// Inside the fan-out, tasks are queued at once. Before entry, tasks are prepared: the Tasks are consumed, but
	// nothing runs and no generator or channel is pulled until the item enters. All prepared calls together form the
	// initial batch (see FanOutBackpressure). Prepared work is discarded without running if the item passes the
	// fan-out without entering or returns, and once a canceled item calls a node method or returns; a declined
	// TryMoveTo keeps it. Prepared work does not appear in Stats.
	//
	// Preparing the work and then entering lets it start as part of the entry:
	//
	// 	if err := f.Schedule(ctx, db1.NewTask(writeDB1), db2.NewTask(writeDB2)); err != nil {
	// 		return err
	// 	}
	// 	if err := f.MoveTo(ctx); err != nil {
	// 		return err
	// 	}
	//
	// Work is queued per branch by item age, then by call order, then by argument order. Running work is never
	// preempted.
	//
	// Calling Schedule with no tasks is legal: it makes the initial batch known (and empty), which lets the item
	// behind enter.
	//
	// It returns ErrForeignContext; ErrStaleContext for the context of a finished item, a finished child, a finished
	// wave, or (before entry) a processor that has returned; or the item's cancellation cause. In these cases nothing is
	// queued or prepared. It panics on misuse: a task of another fan-out, a Task submitted twice, a fan-out the item
	// has already passed, a body closed by leaving (or trying to leave), Retain, or a task calling it for a fan-out it
	// does not run under.
	Schedule(ctx context.Context, tasks ...Task) error

	// Wait blocks until every task scheduled so far in this fan-out's body has finished, including work those tasks
	// scheduled themselves, and returns the first error or the item's cancellation cause. The item keeps its slot and
	// may Schedule again afterwards. Use it for rounds: wait, plan the next tasks from the results, schedule again.
	// Repeated calls return the same error again.
	//
	// Only the ItemProcessor (or a child item at a fan-out of its own lane) may call it, while the body is open. It
	// panics when called with a task's context (a task must never wait for other work), before entering, after leaving,
	// or after Retain.
	Wait(ctx context.Context) error

	// Retain hands the work scheduled here to the returned Wave and lets the item move on without waiting for it. This
	// fan-out's slot stays held until the work is done and the item has moved on. The work may still grow from its own
	// running tasks. It is the fan-out counterpart of Stage.Retain: use it to overlap this fan-out's work with later
	// stages, then call Wave.Wait before the step that needs the result.
	//
	// After Retain the ItemProcessor may not Schedule or Wait here again. Retain does not release the previous node's
	// slot earlier than SetBackpressure allows.
	//
	// It panics on misuse: a fan-out the item does not currently occupy, or nothing to retain (never entered, already
	// retained, or the body was closed by leaving).
	Retain(ctx context.Context) Wave

	// SetLimit sets how many items may be inside this fan-out at once (default 1; a limit <= 0 means 1), and returns
	// the fan-out for chaining. An item counts from entering until it moves into the next node or its waiting room;
	// after Retain it counts until the retained work is done and the item has moved on. The limit does not let the item
	// behind enter before this item's initial batch is known.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor.
	SetLimit(limit int) FanOut

	// SetQueueSize gives this fan-out a waiting room of size items in front of it (a size <= 0 means none), and
	// returns the fan-out for chaining.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor.
	SetQueueSize(size int) FanOut

	// Limit returns how many items may be inside this fan-out at once.
	Limit() int

	// QueueSize returns the size of this fan-out's waiting room, or 0 if it has none.
	QueueSize() int

	// SetBackpressure selects when an item entering this fan-out releases the slot it held before: the previous node's
	// slot, or its place in this fan-out's waiting room. The default is BackpressureBalanced. See FanOutBackpressure.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor. It applies to items entering after
	// the call; items already inside keep the mode they entered with.
	SetBackpressure(mode FanOutBackpressure) FanOut

	// Backpressure returns the backpressure mode.
	Backpressure() FanOutBackpressure

	// Branches returns this fan-out's branches, pools and lanes alike, in creation order.
	Branches() []Branch
}

FanOut is a node whose work runs in parallel on branches. The item enters with MoveTo, adds tasks with Schedule (before or after entering, as often as needed), and the move that takes it out of the node waits for all of them to finish. Use it for concurrent work such as writing to several databases, or, with a Lane, to turn one item into several child journeys.

A branch is either a Pool (AddPool), a single step where a task runs and is done, or a Lane (AddLane), a pipeline whose steps a task travels as a child item. Most fan-outs need only pools.

type FanOutBackpressure

type FanOutBackpressure int

FanOutBackpressure selects when an item entering a fan-out releases the slot it held before: the previous node's slot, or its place in the fan-out's waiting room (see FanOut.SetBackpressure). Entry itself is the same in every mode: the item's turn and a free item slot (SetLimit). Pool capacity never gates entry.

The modes differ in how much of the item's initial batch must start before the previous slot is released. The initial batch is the work prepared with Schedule before entry or, if none, the first Schedule after entry. Later Schedule calls and work scheduled by running tasks never delay the release.

A task counts as started when its callback is dispatched; a lane task counts when its child item is created. A source that ends without producing a task (NewTasks with count 0, a closed channel, a finished generator) counts for its branch as well. Work waiting behind a full pool has not started.

const (
	// BackpressureBalanced is the default. The previous slot is released when the first task of the initial batch starts
	// on any branch, or when every branch of the batch ran out without a task. An item whose work is all waiting for
	// full pools keeps the previous node busy; an item with some work running does not. A good general-purpose choice.
	BackpressureBalanced FanOutBackpressure = iota

	// BackpressureBuffered releases the previous slot on entering, like a plain stage. More items wait inside the
	// fan-out (within its limit) and upstream feels the pressure later. Use it when item N+1 should reach an idle pool
	// while item N waits for a busy one.
	BackpressureBuffered

	// BackpressureStrict releases the previous slot when every branch of the initial batch has started one task or ran
	// out. An item with any initial branch fully waiting keeps the previous slot; a younger item whose work can start
	// may still enter and run ahead of it on the branches. Fewer items in flight, at the cost of throughput when task
	// durations differ. Use it when per-item memory or upstream resources are costly.
	BackpressureStrict
)

func (FanOutBackpressure) String

func (m FanOutBackpressure) String() string

String names the mode for messages and debugging.

type Gauge

type Gauge struct {
	Min  int
	Max  int
	Last int
}

Gauge is a windowed view of an integer quantity: the minimum and maximum it reached since the previous Stats read, plus its value at that read.

type ItemProcessor

type ItemProcessor func(ctx context.Context) error

ItemProcessor processes one item, moving it through the conveyor's nodes with Stage.MoveTo / FanOut.MoveTo using the context it receives. It need not enter every node and may return early.

Returning an error shuts the conveyor down: no new items are created, later items are canceled, and earlier ones are allowed to finish.

Cancellation is judged by the item's own context. Once it is canceled (a failed task or Retain, a shutdown), every node method returns the cause, even when called with a context that hides the cancellation (context.WithoutCancel). A derived context with its own deadline still works. Code that must run after cancellation can still run in plain Go once the node method has returned; it just cannot run inside a node.

type Lane

type Lane interface {
	Branch

	// AddStage adds an interior Stage to this lane. Chain SetLimit to let several children run its code at once.
	AddStage(opts ...AnyUnitOption) Stage

	// AddFanOut adds an interior FanOut to this lane, so a child item may itself scatter work onto its own
	// branches.
	AddFanOut(opts ...AnyUnitOption) FanOut

	// NewTask creates a task of one callback — one child journey through this lane. A nil fn panics.
	NewTask(fn TaskFunc) Task

	// NewTasks creates a task of count independent callbacks fn(ctx, 0)..fn(ctx, count-1), each one child journey
	// through this lane, admitted in index order. A count <= 0 yields a no-op task; a nil fn with count > 0
	// panics.
	//
	// The ctx each callback receives is the child's own — pass it to the lane's stages, not the item's.
	NewTasks(count int, fn func(ctx context.Context, index int) error) Task

	// NewTasksGen creates a task whose callbacks are produced by gen, pulled one by one as the lane's entrance
	// frees up. Use it to stream work without materializing it upfront; gen should respect the ItemProcessor's
	// ctx. State it reads must not be mutated until the wave's Started channel closes. A nil gen is a no-op.
	NewTasksGen(gen iter.Seq[TaskFunc]) Task

	// NewTasksChan creates a task whose callbacks are received from ch, pulled one by one as the lane's entrance
	// frees up. The producer must run in its own goroutine, respect the ItemProcessor's ctx, and close the channel
	// when done. A nil channel is a no-op.
	NewTasksChan(ch <-chan TaskFunc) Task
}

Lane is a branch of a FanOut's work that is a pipeline in its own right: it has interior nodes (AddStage, AddFanOut), and each piece of scheduled work runs as a child item travelling them with the same MoveTo an ItemProcessor uses on the conveyor.

A lane's entrance admits one child at a time (there is no SetLimit), so its parallelism comes from the interior stages' own limits. Children are created in queue order (item age, then Schedule order) and enter the interior nodes in that order. A child may Schedule more work at the fan-out its lane belongs to, before its callback returns.

A lane with no interior nodes behaves as a Pool with limit 1.

type OptName

type OptName string

OptName names a node, so it shows up by that name in Stats and in error messages. An unnamed node falls back to a positional name that shifts when the topology changes around it.

type Option

type Option func(c *conveyor)

Option configures a Conveyor at creation. See NewConveyor and OptShutdownContext.

func OptShutdownContext

func OptShutdownContext(factory ShutdownContextFactory) Option

OptShutdownContext bounds how long items may keep running after a shutdown begins — triggered by the Run context being canceled, or by an ItemProcessor error. Once shutdown starts, the factory is asked for a context; when that context is done, every in-flight item's context is canceled.

return context.WithTimeout(context.Background(), 30*time.Second) // a grace period, then cancel
return alreadyDoneCtx, nil                                       // cancel in-flight items at once
return nil, nil                                                  // no limit: leave them to finish

Without this option, items are left to finish on their own.

type Pool

type Pool interface {
	Branch

	// SetLimit sets how many of this pool's tasks may run at once (default 1; a limit <= 0 means 1), and returns
	// the pool for chaining.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor.
	SetLimit(limit int) Pool

	// Limit returns how many of this pool's tasks may run at once.
	Limit() int

	// NewTask creates a task of one callback — one slot's worth of work on this pool. A nil fn panics.
	NewTask(fn TaskFunc) Task

	// NewTasks creates a task of count independent callbacks fn(ctx, 0)..fn(ctx, count-1), each one slot's worth
	// of work on this pool, built lazily as slots free up. A count <= 0 yields a no-op task; a nil fn with
	// count > 0 panics.
	NewTasks(count int, fn func(ctx context.Context, index int) error) Task

	// NewTasksGen creates a task whose callbacks are produced by gen, pulled one by one as pool slots free up. Use
	// it to stream work without materializing it upfront; gen should respect the ItemProcessor's ctx. State it
	// reads must not be mutated until Wait returns or, after Retain, until the wave's Started channel closes. A nil
	// gen is a no-op.
	NewTasksGen(gen iter.Seq[TaskFunc]) Task

	// NewTasksChan creates a task whose callbacks are received from ch, pulled one by one as pool slots free up.
	// The producer must run in its own goroutine, respect the ItemProcessor's ctx, and close the channel when
	// done. A nil channel is a no-op.
	NewTasksChan(ch <-chan TaskFunc) Task
}

Pool is a single-step branch of a FanOut's work: a task runs on the slot it was given and is done. SetLimit(n) bounds how many of its tasks run at once.

A pool task cannot move or wait: calling MoveTo or Wait with its context panics. Use a Lane for a branch whose work is a multi-step path. The one node method a pool task may call is Schedule of the pool's own fan-out, to add follow-up work before the task returns.

Work is queued per branch by item age: a free slot never goes to a younger item's work while an older item has work queued there. Work that has started is never taken back.

type ShutdownContextFactory

type ShutdownContextFactory func(cause error) (context.Context, context.CancelFunc)

ShutdownContextFactory produces the context that bounds a shutdown. cause is the first item error, or the Run context's cancellation cause. See OptShutdownContext.

type ShutdownError

type ShutdownError interface {
	error
	// Cause returns the shutdown reason: the first item error, or the Run context's cancellation cause.
	Cause() error
	// contains filtered or unexported methods
}

ShutdownError is the cancellation cause of an item's context when the conveyor cancels it during a shutdown. Recover it with errors.As, or reach the wrapped cause with errors.Is (e.g. against context.Canceled).

The interface is sealed: only the conveyor creates ShutdownErrors, so matching one reliably means the conveyor canceled the item.

type Stage

type Stage interface {
	Unit

	// MoveTo advances the item into this stage, releasing the previous node. It blocks until the stage (or its
	// waiting room) has room and it is the item's turn.
	//
	// It returns ErrForeignContext, ErrStaleContext, or the item's cancellation cause — whether the cancellation is
	// visible on ctx or only on the item's own context (see ItemProcessor). It panics on misuse: moving backward,
	// re-entering an already-entered stage, or a handle from another conveyor or series.
	MoveTo(ctx context.Context) error

	// TryMoveTo is MoveTo without waiting: it enters the stage only if a slot is free right now, and reports
	// whether it did. Use it to make a stage optional under load — skip it, or take a different path — instead of
	// queueing.
	//
	// entered == false means nothing happened: the item stays where it is and the stage remains unentered, so it
	// may be tried again or entered later with a blocking MoveTo. It bypasses the stage's waiting room
	// (SetQueueSize) and never jumps an item already waiting there.
	//
	// A canceled item returns (false, its cancellation cause), whether the cancellation is visible on ctx or only on
	// the item's own context. It panics on the same misuse as MoveTo.
	TryMoveTo(ctx context.Context) (entered bool, err error)

	// Retain runs bgOp in the background while keeping this stage's slot held, letting the item move on without
	// releasing the stage. The slot is freed once bgOp returns and the item has moved on. Wait for bgOp with the
	// returned Wave (before or after a later MoveTo); an error from bgOp cancels the item.
	//
	// On a canceled item — canceled on ctx or on its own context — bgOp does not run and the returned wave is
	// already finished, carrying the cancellation cause.
	//
	// It panics on misuse: a handle from another conveyor, or a stage the item does not currently occupy.
	Retain(ctx context.Context, bgOp func() error) Wave

	// SetLimit sets how many items may run this stage's code at once (default 1; a limit <= 0 means 1), and
	// returns the stage for chaining. Item order through the stage is only guaranteed at limit 1.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor; it never evicts items already
	// admitted.
	SetLimit(limit int) Stage

	// SetQueueSize gives this stage a waiting room of size items in front of it (a size <= 0 means none), and
	// returns the stage for chaining. An item that cannot enter the stage directly waits here instead, freeing the
	// previous node. Only MoveTo uses the waiting room; TryMoveTo never does.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor; it never evicts items already
	// waiting.
	SetQueueSize(size int) Stage

	// Limit returns how many items may run this stage's code at once.
	Limit() int

	// QueueSize returns the size of this stage's waiting room, or 0 if it has none.
	QueueSize() int
}

Stage is a node whose work runs inline: the item enters it with MoveTo, runs the stage's code in the ItemProcessor body while holding the slot, then moves on. A stage is entered at most once per item.

type Stats

type Stats struct {
	// InFlight counts the conveyor's own items — one per live ItemProcessor call. Child items of a lane's work are
	// not counted; their work shows up as branch occupancy instead.
	InFlight    Gauge
	LiveWorkers Gauge      // worker goroutines alive for root items
	Units       []UnitStat // one entry per node and branch, in creation order (index 0 is the implicit start stage)
}

Stats is a snapshot of a conveyor's runtime state, for metrics and observability. Its gauge fields report the range observed since the previous Stats call, not just a point sample.

Reading resets the windows, so wire exactly one consumer to Stats.

type Task

type Task struct {
	// contains filtered or unexported fields
}

Task is a bundle of work for one branch, produced by a branch's constructors (NewTask, NewTasks, NewTasksGen, NewTasksChan). Submit tasks with FanOut.Schedule. The tasks of one call may belong to different branches of the same fan-out; several tasks for the same branch start in the order they are listed.

A Task is single-use: submitting the same Task twice panics.

type TaskFunc

type TaskFunc = func(ctx context.Context) error

TaskFunc is one piece of a branch's work. It receives a context — pass it to everything it calls, and to the lane's own nodes if the branch is a Lane — and should return promptly when canceled. A returned error cancels the item.

On a Pool the context can be used with Schedule of the pool's fan-out, to add follow-up work before the task returns, and with nothing else: a pool's work has nowhere to go and must not wait for other work.

type Unit

type Unit interface {
	fmt.Stringer
	// contains filtered or unexported methods
}

Unit is a handle to something with capacity: a Stage, a FanOut, a Branch (Pool or Lane), or Conveyor.StartUnit. It names the node in Stats and in error messages.

type UnitOccupants

type UnitOccupants struct {
	Unit Unit

	// InBody lists the item numbers holding a slot in the node itself, in arrival order; an item holding several slots
	// appears once per slot. A slot may belong to an item that has already moved on: one kept by Stage.Retain or
	// FanOut.Retain for its work, or the previous node's slot kept under Balanced or Strict backpressure until the
	// item's initial batch makes progress (see FanOutBackpressure). Such an item appears in two units at once.
	InBody []int64

	// InQueue lists the item numbers waiting in front of the node, in arrival order. Under Balanced or Strict an entry
	// may belong to an item already admitted to the fan-out from its waiting room. For a branch it is the backlog in
	// queue order (item age, then Schedule order), one entry per submission whose tasks are not all handed out yet: a
	// Schedule call after entry, or all calls before entry together.
	InQueue []int64
}

UnitOccupants is a point-in-time snapshot of exactly which items occupy one unit right now. See Conveyor.DebugUnitOccupants.

type UnitStat

type UnitStat struct {
	Unit     Unit
	Occupied Gauge // slots of the node in use since the previous Stats read
	Limit    int   // the node's capacity — the denominator for Occupied
	Queued   Gauge // work waiting in front of the node since the previous Stats read
}

UnitStat is the per-node portion of Stats. Unit is the handle (a Stage, a FanOut, a Pool, a Lane, or Conveyor.StartUnit) that produced it.

Occupied and Limit describe the node itself. A slot counts while an item runs the node's code or has work outstanding there, while Stage.Retain or FanOut.Retain keeps it for work in flight, and while an item admitted to a Balanced or Strict fan-out still keeps the previous node's slot (see FanOutBackpressure). So a node's occupancy may include items that are already in the next node. Work prepared with Schedule before entering a fan-out is not counted anywhere.

Queued describes what waits in front of the node: items in a waiting room (under Balanced or Strict, possibly an item already admitted to the fan-out from that room), or, for a branch, the submissions whose tasks are not all handed out yet: one per Schedule call after entry, and one for all calls before entry together. Running work is never counted, and a branch's backlog is not bounded by its fan-out's limit.

type Wave

type Wave interface {
	// Started is closed once the item has stopped adding to the wave (it retained or moved on) and every task the
	// ItemProcessor scheduled into it has been handed out: its streaming sources drained, or the item canceled. Work
	// that tasks or lane children scheduled themselves does not delay it. Use it with streaming sources (NewTasksGen,
	// NewTasksChan): state they read must not be mutated until it closes.
	Started() <-chan struct{}

	// Finished is closed once every task of this wave has finished, including work those tasks scheduled themselves,
	// or was skipped because the item was canceled, and the item has stopped adding to the wave. After it is closed, Err
	// reports the outcome.
	Finished() <-chan struct{}

	// Err returns the first error the wave's work produced, or nil. It is only final once Finished is closed.
	// Reading it counts as observing the outcome.
	Err() error

	// Wait blocks until the wave is finished and returns its error, named after the node the work belongs to
	// ("<node> work: ..."), or nil. Only the item that created the wave may call it, with its own context or one
	// derived from it.
	//
	// A failing task cancels the item, so a wait in progress wakes at once. If the wave is finished by then, its error
	// is returned and counts as observed. If other tasks are still winding down, the item's cancellation cause is
	// returned and the outcome stays unobserved: wait for Finished and read Err, or call Wait again after Finished is
	// closed. A canceled item never gets nil. A canceled call context (a deadline the caller added) ends the wait the
	// same way.
	//
	// It returns ErrForeignContext for a context without an item and ErrStaleContext once the item has finished. It
	// panics on misuse: a wave of another item (a lane child may wait only on its own waves), or a task's context (a
	// task must never wait for other work).
	Wait(ctx context.Context) error
}

Wave is a handle to background work an item started: a Stage.Retain operation, or fan-out work handed over with FanOut.Retain. Wait for it with Wait, before or after the next MoveTo (the choice decides which node's slot the item holds meanwhile), or read Finished and Err.

A wave whose error nobody observes still fails the item when it completes.