go-kitsune API Reference

Documents every exported operator and which StageOption features each one actually uses in its implementation.

Feature Key

Cell values

1 · Core Transforms

Notes - Filter, Tap, Reject support Supervise but not OnError; errors from their fn/pred propagate directly. - TapError fires its callback only for non-context errors; context cancellation does not trigger the callback. It does not accept StageOption (implemented via Generate, like Catch). - Finally fires for all exits (success, error, cancellation, early consumer stop). On early stop (e.g. downstream Take), fn receives nil. Does not accept StageOption. - ExpandMap performs BFS expansion: items at depth N are all emitted before any item at depth N+1. fn may return nil for leaf nodes. Accepts WithName and Buffer. Use MaxDepth(n) to cap BFS depth (0 = roots only) and MaxItems(n) to cap total emissions; the stage closes normally when either cap is reached. Use VisitedBy(keyFn) to enable cycle detection (items whose key was already seen are skipped, along with their subtrees); combine with WithDedupSet to override the default MemoryDedupSet backend. WARNING: without MaxDepth, MaxItems, or a downstream Take, ExpandMap traverses the entire reachable graph and can exhaust memory on high-fanout inputs. - ForEach returns a typed ForEachRunner[T]; call .Run(ctx) or .RunAsync(ctx). Supports Concurrency, Ordered, OnError, and Supervise. - Drain returns a DrainRunner[T] (deprecated; use ForEach with a no-op). It satisfies Runnable, so it can be passed to MergeRunners directly; RunAsync is also available on it. - Map → FlatMap → ForEach chains fuse into a single goroutine when the chain is serial, hook-free, and uses default overflow (FP column). See doc/tuning.md for exact eligibility conditions. - Pipeline[T].IsOptimized(opts ...RunOption) []OptimizationReport reports fast-path and fusion eligibility for each stage. Pipeline[T].IsFastPath(opts ...RunOption) bool is a convenience wrapper. Both are non-destructive. - WithContextMapper[T](fn func(T) context.Context) is accepted by Map, FlatMap, and ForEach. It extracts a per-item context from each item using fn, enabling per-item tracing or baggage propagation without requiring the item type to implement ContextCarrier. When set, it takes precedence over ContextCarrier. Setting WithContextMapper disqualifies the stage from the fast path (FP = – when set). See doc/options.md for full details.

2 · Higher-Order Maps

Notes - SwitchMap cancels the active inner pipeline when a new upstream item arrives. ExhaustMap ignores new items while an inner pipeline is active. - Timeout on SwitchMap/ExhaustMap threads a per-item deadline into the inner fn context. - MapBatch (compat) delegates to Batch + FlatMap; it only threads Buffer, Name, Err, and BatchTimeout to the internal Batch stage. - MapPooled: fn receives *Pooled[O] (pointer). ReleaseAll takes []*Pooled[T]. Pooled[T].Release() has a pointer receiver. - MapResult both branches must be consumed before calling Run.

3 · State Transforms

Notes - !: Timeout and CacheBy panic at construction; they are not meaningful for stateful loops. - Concurrency semantics differ by operator: - MapWith / FlatMapWith: each of n workers gets its own independent Ref[S] (worker-local state). - MapWithKey / FlatMapWithKey: the key space is sharded across n workers via hash(key) % n. Same-key items always reach the same worker; lock-free in the hot path. - Supervise wraps the stage loop; the Ref (or keyed ref map) is initialised outside the inner fn and is preserved across restarts. - State TTL: NewKey("name", initial, StateTTL(d)). Ref.Get returns the zero value and resets the slot when the TTL has elapsed. - Per-key eviction (MapWithKey / FlatMapWithKey only): WithKeyTTL(d) evicts map entries that have been inactive for longer than d; the next item for that key starts from the initial value. Eviction is lazy (checked on next access; no background goroutine). Independent of StateTTL. Use WithDefaultKeyTTL(d) as a run-level default; per-stage WithKeyTTL(0) overrides it to disable eviction for that stage.

4 · Batching & Windowing

Notes - ✱ Batch requires at least one flush trigger (BatchCount, BatchMeasure, or BatchTimeout); panics at construction if none are provided. - Batch supports WithClock only when BatchTimeout is also set (the clock powers the flush ticker). - DropPartial discards the final partial batch when the source closes; only full batches are emitted. Use BufferWith(p, Ticker(d)) for fixed-duration time bucketing. - BufferWith takes a second pipeline (closingSelector) as its flush trigger; each signal from that pipeline emits the current buffer. When the selector closes, any remaining items are flushed. Named BufferWith to avoid collision with the Buffer(n) stage option. - ChunkBy emits a group when the key changes. ChunkWhile emits a group when the predicate between adjacent items is false.

5 · Aggregation

Notes - GroupBy, TakeRandom, ReduceWhile, ToMap, Frequencies, FrequenciesBy are all buffering pipeline operators: they buffer the full source and emit exactly one result when the source closes. Combine with Single to extract the value. - Dedupe is identity-based (T comparable). DedupeBy is key-based. Default behaviour is global (all-seen) deduplication. Use DedupeWindow(1) for consecutive-only deduplication. - RandomSample passes each item downstream with independent probability rate; it is stateless and streaming (no buffering). - RunningFrequencies / RunningFrequenciesBy emit a fresh count-map snapshot after each item. - RunningCountBy / RunningSumBy emit a fresh full snapshot after each item; the snapshot map is a copy so it is safe to retain.

6 · Fan-Out / Fan-In

Notes - Merge, Zip, CombineLatest, LatestFrom create no buffered output channel of their own, so Buffer does not apply. - The *With variants run a user fn and do produce a buffered output channel; Buffer and Name apply. - Broadcast requires n ≥ 2. - Share returns a factory; call the factory once per desired branch before building the runner. At least one subscribe call is required. Buffer and WithName can be set on each individual subscribe call (per-subscribe opts override factory opts). Calling the factory after Run() has started panics. Unlike Broadcast, Share allows a single subscriber. - LookupBy and Enrich also accept BatchTimeout as a first-class field on LookupConfig / EnrichConfig; the field is equivalent to passing the BatchTimeout(d) StageOption and is forwarded to the internal Batch stage.

7 · Sources

Source operators produce items from external input. They accept no StageOption.

Time-based sources: accept Buffer, Name, and WithClock as StageOption:

8 · Terminal Functions

Terminal functions run the pipeline and return a materialised result. They accept ...RunOption (not StageOption).

Notes - Iter exposes a pipeline as iter.Seq[T] for range-over-func (Go 1.23+). Breaking out of the loop early cancels the pipeline. - Single returns a plain error if the pipeline emits zero items (unless OrDefault(v) or OrZero[T]() is supplied) or more than one item.

9 · Timing & Observation

Notes - Sample emits the most-recently-seen item on each tick and resets the latch; ticks with no new item produce no output. Unlike Debounce, it does not flush on source close.

10 · Sequence Operators

Notes - Take, Drop, TakeWhile, DropWhile, TakeEvery, DropEvery, DropLast use hardcoded buffer sizes and accept no options. - TakeUntil / DropUntil accept any *Pipeline[U] as boundary; only its first emission matters. - StartWith / EndWith accept no options; they delegate to Concat + FromSlice. - Indexed[T] is struct{ Index int; Value T }. - (p).Skip is an alias for Drop.

11 · Middleware

RateLimitOpt (not StageOption):

CircuitBreakerOpt (not StageOption):

Notes - WithName, Buffer, and WithClock are the only StageOptions that apply to both operators. - CircuitBreaker emits ErrCircuitOpen when the circuit is open.

12 · Stage Composition

13 · Error Routing

14 · Helper / Lift Functions

15 · Pool

16 · Run Options

RunOption values are passed to Runner.Run(ctx, opts...), Runner.RunAsync(ctx, opts...), or any terminal function.

17 · Observability

All hooks are wired into every stage runner automatically when provided via WithHook.

GraphNode exposes: Kind, Name, Concurrency, Buffer, Overflow, BatchSize, Timeout, HasRetry, HasSupervision.

Pipeline[T].Describe() []GraphNode: returns the same []GraphNode snapshot synchronously, without executing the pipeline. Callable on any *Pipeline[T], including intermediate (non-terminal) stages. Useful for static validation and unit-testing graph structure without a full Run.

ContextCarrier: interface implemented by item types that carry a context.Context with an attached trace span. When an item implements ContextCarrier, the engine merges its context into the stage function call: cancellation still comes from the pipeline stage context, but context values (e.g. the active trace span) come from the item. Stage functions can call tracer.Start(ctx, "my-work") to create per-item child spans with no signature changes. Zero cost for items that don't implement the interface. See tails/kotel for OTel integration.

18 · DedupSet Backends

WithDedupSet(s) accepts any value implementing DedupSet:

Notes - MemoryDedupSet is the default for Dedupe, DedupeBy, and ExpandMap. - BloomDedupSet panics if expectedItems <= 0 or falsePositiveRate is not in (0, 1). - TTLDedupSet panics if ttl <= 0. Re-adding an existing key refreshes its expiry. Eviction is lazy (on next Contains or Add); no background goroutine is started. - External backends (Redis, etc.) are available in tails/kredis via kredis.NewDedupSet.

19 · Testing Infrastructure

20 · Tails (External Adapters)

Tails are separate Go modules under tails/ that adapt external systems to kitsune pipelines. Each follows the "user owns the client" principle: the caller creates, configures, and closes connections; kitsune never opens or closes them. See doc/tails.md for detailed usage examples.