toc

Mar 23, 2026

Changes in this version

+ type Observer struct

+ func NewObserver(pipelineID string) *Observer

+ func (o *Observer) AddStage(s ObserverStage)

+ func (o *Observer) Run(ctx context.Context, interval time.Duration, fn func(PipelineSnapshot))

+ func (o *Observer) RunWithTicker(ctx context.Context, ticks <-chan time.Time, fn func(PipelineSnapshot))

+ func (o *Observer) SetDiagnosis(fn func() *core.Diagnosis)

+ func (o *Observer) Snapshot() PipelineSnapshot

+ type ObserverStage struct

+ Name string

+ Stats func() Stats

+ UnitLabel string

+ type PipelineSnapshot struct

+ At time.Time

+ Diagnosis *core.Diagnosis

+ GoHeap uint64

+ PipelineID string

+ RSS uint64

+ RSSOK bool

+ Stages []StageSnapshotEntry

+ type StageSnapshotEntry struct

+ Name string

+ Order int

+ QueueCapacity int

+ QueueDepth int64

+ State core.StageState

+ Stats Stats

+ UnitLabel string

+ Workers int

Mar 23, 2026

Changes in this version

+ type BudgetAllocatorHandle struct

+ func BudgetAllocator(stages []StageAllocation, drumName string, budgetFraction float64, ...) *BudgetAllocatorHandle

+ func (h *BudgetAllocatorHandle) Callback() func(context.Context, memctl.MemInfo)

type LimitSource

+ const LimitSourceBudgetAllocator

+ type StageAllocation struct

+ Limits *LimitManager

+ Name string

+ Share float64

Mar 23, 2026

Changes in this version

+ type LimitHandle struct

+ func (h *LimitHandle) Close()

+ func (h *LimitHandle) ProposeCount(limit int)

+ func (h *LimitHandle) ProposeWeight(limit int64)

type LimitManager

+ func (m *LimitManager) Register(source LimitSource) *LimitHandle

type Options

+ OversizePolicy OversizePolicy

+ type OversizePolicy int

+ const OversizeReject

+ const OversizeWait

type Stage

+ func (s *Stage[T, R]) Limits() *LimitManager

Mar 23, 2026

Changes in this version

type LimitSnapshot

+ CountProposals map[string]int

+ WeightProposals map[string]int64

+ type LimitSource = string

+ const LimitSourceMemoryRope

+ const LimitSourceProcessingRope

+ const LimitSourceWeightRope

Mar 23, 2026

Changes in this version

+ type LimitManager struct

+ func NewLimitManager(setCount func(int) int, setWeight func(int64) int64, defaultCount int, ...) *LimitManager

+ func (m *LimitManager) Effective() LimitSnapshot

+ func (m *LimitManager) ProposeCount(source string, limit int)

+ func (m *LimitManager) ProposeWeight(source string, limit int64)

+ func (m *LimitManager) WithdrawCount(source string)

+ func (m *LimitManager) WithdrawWeight(source string)

+ type LimitSnapshot struct

+ AppliedCount int

+ AppliedWeight int64

+ CountSource string

+ CountSources int

+ EffectiveCount int

+ EffectiveWeight int64

+ WeightSource string

+ WeightSources int

type Stage

+ func (s *Stage[T, R]) DisableMaxWIPWeight()

type Stats

+ MaxWIPWeightEnabled bool

Mar 23, 2026

Changes in this version

+ type MemoryRopeHandle struct

+ func MemoryRope(pipeline *Pipeline, drum string, setHeadWIPWeight func(int64) int64, ...) *MemoryRopeHandle

+ func (h *MemoryRopeHandle) Callback() func(context.Context, memctl.MemInfo)

+ func (h *MemoryRopeHandle) Stats() MemoryRopeStats

+ type MemoryRopeStats struct

+ Adjustments int64

+ Applied int64

+ Budget int64

+ Headroom int64

+ Weight int64

type RopeController

+ func NewWeightRopeController(pipeline *Pipeline, drum string, setHeadWIPWeight func(int64) int64, ...) *RopeController

+ type ScalingHistory struct

+ func NewScalingHistory() *ScalingHistory

+ func (h *ScalingHistory) DiminishingReturns(workers int, threshold float64) bool

+ func (h *ScalingHistory) Len() int

+ func (h *ScalingHistory) Record(workers int, throughput float64)

+ func (h *ScalingHistory) Reset()

+ func (h *ScalingHistory) ScalingGain(workers int) (float64, bool)

Mar 22, 2026

Changes in this version

+ func Adapt(name string, prev, curr Stats, elapsed time.Duration) core.StageObservation

Mar 22, 2026

Changes in this version

+ type FocusingStep int

+ const StepElevate

+ const StepExploit

+ const StepIdentify

+ const StepPreventInertia

+ const StepSubordinate

+ func ClassifyStep(prevConstraint string, currConstraint string, ropeActive bool, ...) FocusingStep

+ func (s FocusingStep) String() string

Mar 22, 2026

Changes in this version

+ func BufferCapacity(throughput float64, protectionTime time.Duration) int

+ type MemoryFeverZone int

+ const MemoryGreen

+ const MemoryRed

+ const MemoryUnknown

+ const MemoryYellow

+ func MemoryFever(headroom, limit uint64, yellowAt, redAt float64) MemoryFeverZone

+ func (z MemoryFeverZone) String() string

Mar 22, 2026

Changes in this version

+ type RopeController struct

+ func NewRopeController(pipeline *Pipeline, drum string, setHeadWIP func(int) int, ...) *RopeController

+ func (rc *RopeController) Run(ctx context.Context)

+ func (rc *RopeController) RunWithTicker(ctx context.Context, ticks <-chan time.Time)

+ func (rc *RopeController) Stats() RopeStats

+ type RopeOption func(*RopeController)

+ func WithInitialRopeLength(n int) RopeOption

+ func WithRopeLogger(l *log.Logger) RopeOption

+ func WithRopeSafetyFactor(factor float64) RopeOption

+ type RopeStats struct

+ AdjustmentCount int64

+ DrumErrorRate float64

+ DrumGoodput float64

+ HeadAppliedWIP int

+ RopeLength int

+ RopeUtilization float64

+ RopeWIP int

Mar 22, 2026

Changes in this version

type BatcherStats

+ func (s BatcherStats) ToStats() Stats

+ type IncomingEdge struct

+ From string

+ Ratio int

type IntervalStats

+ ArrivalRate float64

+ Goodput float64

+ type Pipeline struct

+ func NewPipeline() *Pipeline

+ func (p *Pipeline) AddEdge(from, to string)

+ func (p *Pipeline) AddEdgeWithRatio(from, to string, ratio int)

+ func (p *Pipeline) AddStage(name string, stats func() Stats)

+ func (p *Pipeline) AncestorsOf(target string) []string

+ func (p *Pipeline) DirectPredecessors(name string) []string

+ func (p *Pipeline) EdgeRatio(from, to string) int

+ func (p *Pipeline) Freeze()

+ func (p *Pipeline) HasPath(from, to string) bool

+ func (p *Pipeline) Heads() []string

+ func (p *Pipeline) HeadsTo(target string) []string

+ func (p *Pipeline) Incoming(name string) []IncomingEdge

+ func (p *Pipeline) StageStats(name string) func() Stats

+ func (p *Pipeline) Stages() []string

type Stats

+ WIPWaitCount int64

+ WIPWaitNs int64

type WeightedBatcherStats

+ func (s WeightedBatcherStats) ToStats() Stats

Mar 22, 2026

Changes in this version

type Options

+ TrackServiceTimeDist bool

+ type ServiceTimeSummary struct

+ Count int64

+ Max time.Duration

+ Mean time.Duration

+ Min time.Duration

+ Overflow int64

+ P50 time.Duration

+ P95 time.Duration

+ P99 time.Duration

+ StdDev time.Duration

+ Underflow int64

type Stats

+ ServiceTimeDist ServiceTimeSummary

Mar 22, 2026

Changes in this version

+ type BufferZone int

+ const BufferGreen

+ const BufferRed

+ const BufferYellow

+ func (z BufferZone) String() string

+ type IntervalStats struct

+ ApproxUtilization float64

+ CurrActiveWorkers int

+ CurrBufferPenetration float64

+ CurrBufferedDepth int64

+ CurrQueueCapacity int

+ CurrTargetWorkers int

+ Duration time.Duration

+ ErrorRate float64

+ IdleTimeDelta time.Duration

+ ItemsCanceled int64

+ ItemsCompleted int64

+ ItemsFailed int64

+ ItemsSubmitted int64

+ MeanServiceTime time.Duration

+ OutputBlockedDelta time.Duration

+ QueueGrowthRate float64

+ ResetDetected bool

+ ServiceTimeDelta time.Duration

+ Throughput float64

+ func Delta(prev, curr Stats, elapsed time.Duration) IntervalStats

+ func (s IntervalStats) BufferZone() BufferZone

Mar 22, 2026

Changes in this version

+ var ErrStopping = errors.New("toc: stage is stopping")

type Stage

+ func (s *Stage[T, R]) ActiveWorkers() int

+ func (s *Stage[T, R]) SetWorkers(n int) (int, error)

+ func (s *Stage[T, R]) TargetWorkers() int

type Stats

+ ActiveWorkers int

+ TargetWorkers int

Mar 22, 2026

Changes in this version

+ var ErrWeightExceedsLimit = errors.New("toc: item weight exceeds MaxWIPWeight")

type Options

+ MaxWIPWeight int64

type Stage

+ func (s *Stage[T, R]) MaxWIPWeight() int64

+ func (s *Stage[T, R]) SetMaxWIPWeight(n int64) int64

type Stats

+ AdmittedWeight int64

+ MaxWIPWeight int64

Mar 22, 2026

Changes in this version

+ type Reporter struct

+ func NewReporter(interval time.Duration, opts ...ReporterOption) *Reporter

+ func (r *Reporter) AddStage(name string, fn func() Stats)

+ func (r *Reporter) Run(ctx context.Context)

+ type ReporterOption func(*Reporter)

+ func WithLogger(l *log.Logger) ReporterOption

Mar 21, 2026

Changes in this version

type Stage

+ func (s *Stage[T, R]) PauseAdmission()

+ func (s *Stage[T, R]) Paused() bool

+ func (s *Stage[T, R]) ResumeAdmission()

type Stats

+ Paused bool

Mar 21, 2026

Changes in this version

type Stats

+ MaxWaiterCount int

+ WaiterCount int

Mar 21, 2026

Changes in this version

type Options

+ MaxWIP int

type Stage

+ func (s *Stage[T, R]) MaxWIP() int

+ func (s *Stage[T, R]) SetMaxWIP(n int) int

type Stats

+ Admitted int64

+ MaxWIP int

+ RopeWaitCount int64

+ RopeWaitNs int64

Mar 19, 2026

Changes in this version

+ func FromChan[T any](ch <-chan T) <-chan rslt.Result[T]

Mar 19, 2026

Changes in this version

+ type Join struct

+ func NewJoin[A, B, R any](ctx context.Context, srcA <-chan rslt.Result[A], srcB <-chan rslt.Result[B], ...) *Join[R]

+ func (j *Join[R]) Out() <-chan rslt.Result[R]

+ func (j *Join[R]) Stats() JoinStats

+ func (j *Join[R]) Wait() error

+ type JoinStats struct

+ Combined int64

+ DiscardedA int64

+ DiscardedB int64

+ Errors int64

+ ExtraA int64

+ ExtraB int64

+ OutputBlockedTime time.Duration

+ ReceivedA int64

+ ReceivedB int64

+ type Merge struct

+ func NewMerge[T any](ctx context.Context, sources ...<-chan rslt.Result[T]) *Merge[T]

+ func (m *Merge[T]) Out() <-chan rslt.Result[T]

+ func (m *Merge[T]) Stats() MergeStats

+ func (m *Merge[T]) Wait() error

+ type MergeStats struct

+ Dropped int64

+ Forwarded int64

+ Received int64

+ SourceDropped []int64

+ SourceForwarded []int64

+ SourceReceived []int64

+ type MissingResultError struct

+ Source string

+ func (e *MissingResultError) Error() string

Mar 18, 2026

Changes in this version

type Options

+ TrackAllocations bool

type Stats

+ AllocTrackingActive bool

+ ObservedAllocBytes uint64

+ ObservedAllocObjects uint64

+ type Tee struct

+ func NewTee[T any](ctx context.Context, src <-chan rslt.Result[T], n int) *Tee[T]

+ func (t *Tee[T]) Branch(i int) <-chan rslt.Result[T]

+ func (t *Tee[T]) Stats() TeeStats

+ func (t *Tee[T]) Wait() error

+ type TeeStats struct

+ BranchBlockedTime []time.Duration

+ BranchDelivered []int64

+ FullyDelivered int64

+ PartiallyDelivered int64

+ Received int64

+ Undelivered int64

Mar 17, 2026

Changes in this version

+ var ErrClosed = errors.New("toc: stage closed")

+ type Batcher struct

+ func NewBatcher[T any](ctx context.Context, src <-chan rslt.Result[T], n int) *Batcher[T]

+ func (b *Batcher[T]) Out() <-chan rslt.Result[[]T]

+ func (b *Batcher[T]) Stats() BatcherStats

+ func (b *Batcher[T]) Wait() error

+ type BatcherStats struct

+ BatchCount int64

+ BufferedDepth int64

+ Dropped int64

+ Emitted int64

+ Forwarded int64

+ OutputBlockedTime time.Duration

+ Received int64

+ type Options struct

+ Capacity int

+ ContinueOnError bool

+ Weight func(T) int64

+ Workers int

+ type Stage struct

+ func Pipe[T, R any](ctx context.Context, src <-chan rslt.Result[T], ...) *Stage[T, R]

+ func Start[T, R any](ctx context.Context, fn func(context.Context, T) (R, error), opts Options[T]) *Stage[T, R]

+ func (s *Stage[T, R]) Cause() error

+ func (s *Stage[T, R]) CloseInput()

+ func (s *Stage[T, R]) DiscardAndCause() error

+ func (s *Stage[T, R]) DiscardAndWait() error

+ func (s *Stage[T, R]) Out() <-chan rslt.Result[R]

+ func (s *Stage[T, R]) Stats() Stats

+ func (s *Stage[T, R]) Submit(ctx context.Context, item T) error

+ func (s *Stage[T, R]) Wait() error

+ type Stats struct

+ BufferedDepth int64

+ Canceled int64

+ Completed int64

+ Dropped int64

+ Failed int64

+ Forwarded int64

+ IdleTime time.Duration

+ InFlightWeight int64

+ OutputBlockedTime time.Duration

+ Panicked int64

+ QueueCapacity int

+ Received int64

+ ServiceTime time.Duration

+ Submitted int64

+ type WeightedBatcher struct

+ func NewWeightedBatcher[T any](ctx context.Context, src <-chan rslt.Result[T], threshold int, ...) *WeightedBatcher[T]

+ func (b *WeightedBatcher[T]) Out() <-chan rslt.Result[[]T]

+ func (b *WeightedBatcher[T]) Stats() WeightedBatcherStats

+ func (b *WeightedBatcher[T]) Wait() error

+ type WeightedBatcherStats struct

+ BatchCount int64

+ BufferedDepth int64

+ BufferedWeight int64

+ Dropped int64

+ Emitted int64

+ Forwarded int64

+ OutputBlockedTime time.Duration

+ Received int64