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
   • optional backpressure that follows the pools with SetAdmission: items enter only while a pool has capacity, or, stricter, also stay in the previous stage until their work has started
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.

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, Detach 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

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 FanOutAdmission
  • func (a FanOutAdmission) 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 reach a kind's own methods (Pool.SetLimit, Lane.AddStage).

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, which in effect bounds
	// how many workers run concurrently: one worker drives one root item for its whole journey, from creation to
	// completion (see Run). A limit <= 0 means unlimited, the default. It returns the conveyor for chaining.
	//
	// Unlike a node's SetLimit, this bounds the conveyor globally, on top of whatever capacity the nodes themselves
	// admit — it does not replace per-node limits, and setting it does not change any of them.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor: raising it wakes the standby
	// worker so waiting to create the next item is picked up at once; lowering it never evicts an item already in
	// flight — it only stops new items from being created until the in-flight 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.Detach is represented by a Wave, waited for with Wave.Wait or read through 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, releasing the previous node on entering. Under AdmitByPools it
	// also waits for a branch with free capacity; under AdmitByPoolsStrict it additionally keeps the previous node's
	// slot until the item's first Schedule has started (see SetAdmission). The item enters with an empty body; add
	// work with Schedule.
	// Until the item's first Schedule (or until it leaves or detaches), the item behind it may step into this
	// fan-out's waiting room but cannot enter the node.
	//
	// 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. Under
	// AdmitByPools it also declines while no pool has free capacity.
	//
	// When entered is false nothing happened. 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)

	// Schedule adds tasks to a body of this fan-out on behalf of the calling context and returns once they are
	// queued. It never blocks. Three callers are allowed:
	//   - the ItemProcessor, while its item is inside this fan-out with an open body (not yet left, not detached);
	//   - 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.
	//
	// Work is queued per branch at the owning item's place: after everything that item and older items already
	// have queued there, ahead of younger items' queued work. Running work is never preempted. Several tasks for
	// one branch in one call start in argument order.
	//
	// The item's first Schedule opens this fan-out to the item behind it. Calling Schedule with no tasks is legal
	// and does only that.
	//
	// It returns ErrForeignContext; ErrStaleContext when the context belongs to a finished item, a finished child,
	// or work whose wave has finished; or the item's cancellation cause. In all these cases nothing is queued.
	// It panics on misuse: a task for another fan-out's branch, a Task submitted twice, the ItemProcessor calling
	// it outside this fan-out, after it left or tried to leave (the body is closed), or after Detach, 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 (visible on ctx
	// or only on the item's own context). The item keeps its slot and may Schedule again afterwards. 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), outside this fan-out,
	// after the body was closed by a leave, or after Detach.
	Wait(ctx context.Context) error

	// Detach hands this fan-out's slot to the work already scheduled here, letting the item move on without
	// waiting for it. Wait for the returned Wave (before or after a later MoveTo), or read its channels. The
	// detached work may still
	// grow from its own running tasks; the wave finishes when all of it is done. After Detach the ItemProcessor
	// may not Schedule or Wait here again.
	//
	// It panics on misuse: a node the item does not currently occupy, or nothing to detach (never entered here,
	// already detached, or the body was already closed by a leave).
	Detach(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 the moment it enters until it steps into the next node
	// or that node's waiting room. After Detach it counts until both the detached work is done and the item has
	// moved on, whichever is later. The ordering gate is separate from the limit: the item behind cannot enter
	// before the item ahead has scheduled, left, or detached, whatever the limit.
	//
	// 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

	// SetAdmission selects when MoveTo lets an item enter this fan-out and when the item leaves the previous stage
	// (default AdmitByLimit), and returns the fan-out for chaining. See FanOutAdmission.
	//
	// 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 behaving as they were let in.
	SetAdmission(a FanOutAdmission) FanOut

	// Admission returns the admission policy.
	Admission() FanOutAdmission

	// 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 (as many times as it likes, from the ItemProcessor or from the tasks themselves), 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 FanOutAdmission

type FanOutAdmission int

FanOutAdmission selects when FanOut.MoveTo lets an item enter the fan-out and when the item leaves the previous stage (see FanOut.SetAdmission).

const (
	// AdmitByLimit is the default. MoveTo lets the item in when the fan-out has a free item slot (SetLimit) and it is
	// the item's turn. The item leaves the previous stage at that moment, like at every other node.
	AdmitByLimit FanOutAdmission = iota

	// AdmitByPools makes the backpressure follow pool capacity. MoveTo lets the item in only when, on top of the
	// above, some branch has a free slot and nothing queued; the item leaves the previous stage at that moment. So
	// items keep entering while any pool can take work, and stop as soon as no pool can, whatever the item limit
	// still allows; the item limit bounds how many items have work queued on a full pool. See docs/4_fan-out.md,
	// "SetAdmission".
	AdmitByPools

	// AdmitByPoolsStrict is AdmitByPools with stricter backpressure: the item also stays in the previous stage until
	// its first Schedule has started one task on every branch it touched (or until it leaves, detaches, or its
	// processor returns). So an item whose work waits for a full pool keeps the stage before the fan-out busy, while
	// a younger item whose work can start may enter and run ahead of it on the branches; downstream order is
	// unchanged. Fewer items in flight, at the cost of throughput when task durations vary. See docs/4_fan-out.md,
	// "SetAdmission".
	AdmitByPoolsStrict
)

func (FanOutAdmission) String

func (a FanOutAdmission) String() string

String names the policy 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, not by the context passed to a node method: once the item's own context 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, using the same MoveTo an ItemProcessor uses on the conveyor.

A lane's entrance always admits one child at a time — there is no SetLimit — so its parallelism comes from the interior stages' own limits. Children are created in the lane's queue order (item age, then Schedule order) and are admitted to the interior nodes in creation 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 is legal but behaves as a Pool pinned at concurrency 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 Detach, 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's work has nowhere to travel, so calling MoveTo or Wait with a pool task's context panics — use a Lane for a branch whose work is a multi-step path. The one node method that context 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 is never given to a younger item's work while an older item has work queued there. Work that has started, or a slot already reserved for a streaming pull, 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 a Stage.Retain or a
	// FanOut.Detach keeps for its work, or the previous-node slot an item admitted to an AdmitByPoolsStrict fan-out keeps
	// until its first Schedule has started (see FanOutAdmission). 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 AdmitByPoolsStrict a queued
	// token may belong to an item already admitted to the fan-out from its waiting room, kept the same way as a
	// previous-node slot. For a branch it is the backlog in queue order — item age, then Schedule order — which is
	// not the order the work arrived in; one entry per collection (the tasks one Schedule call queued on that branch)
	// not yet fully handed out, so an item that schedules there several times (rounds, follow-ups from its tasks)
	// appears once per collection.
	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, for matching a stat back to what was built.

Occupied and Limit describe the node itself: items running its code, with work outstanding, or occupying a branch's entrance. A slot also counts while a Stage.Retain or a FanOut.Detach keeps it for work in flight, and while an item admitted to an AdmitByPoolsStrict fan-out keeps the previous node's slot until its first Schedule has started (see FanOutAdmission) — so a node's occupancy may include items that are already in the next one. Queued describes what is waiting in front of it — items in a stage's or fan-out's waiting room (under AdmitByPoolsStrict possibly a token kept by an item already admitted from that room), or, for a branch, the collections (the tasks one Schedule call queued there) not yet fully handed out. An item may have several collections queued on one branch, so a branch's backlog is not bounded by its fan-out's limit; running work is never counted. Held tokens are not reported separately.

type Wave

type Wave interface {
	// Started is closed once the wave is sealed (detached, or its item 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 count and does not delay it. Relevant mainly for streaming
	// sources (NewTasksGen, NewTasksChan): state they read must not be mutated until it closes.
	Started() <-chan struct{}

	// Finished is closed once the wave is sealed and every task of this wave has finished — including work those
	// tasks scheduled themselves — or was skipped because the item was canceled. 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 poisons 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 the wave is still winding down (other tasks running), 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 work at a fan-out taken over with FanOut.Detach. 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.