Rewrite package and function docs

batch.go

@@ -4,23 +4,22 @@ import (
 	"time"
 )
 
-// Batch takes a stream of items and returns a stream of batches based on a maximum size and a timeout.
+// Batch groups consecutive values of the stream into batches. With
+// timeout = -1, it accumulates values until the batch reaches size,
+// then emits it. When the input is exhausted, any pending values are
+// emitted as a final batch.
 //
-// A batch is emitted when one of the following conditions is met:
-//   - The batch reaches the maximum size
-//   - The time since the first item was added to the batch exceeds the timeout
-//   - An error is encountered in the input stream
-//   - The input stream is closed
+// Input errors create batch boundaries: any pending batch is emitted
+// first, followed by the error as a separate item. This function never
+// emits empty batches.
 //
-// Errors are never included in batches. Each error is forwarded to the output as a separate item,
-// preserving the relative order of values and errors.
+// A positive timeout adds another trigger: each batch has that much
+// time to fill, starting from its first value. When the timeout expires,
+// the pending batch is emitted even if it is not full. This trades batch
+// size for latency: sparse input produces smaller batches, but no value
+// is ever held longer than timeout, assuming there's no backpressure.
 //
-// This function never emits empty batches. To disable the timeout and emit batches only based on the size,
-// set the timeout to -1. Setting the timeout to zero is not supported and will result in a panic
-//
-// This is a non-blocking ordered function that processes items sequentially.
-//
-// See the package documentation for more information on non-blocking ordered functions and error handling.
+// A zero timeout panics. Use a small positive timeout instead.
 func Batch[A any](in <-chan Try[A], size int, timeout time.Duration) <-chan Try[[]A] {
 	validateMinSize(size, 1)
 	if timeout == 0 {
@@ -113,10 +112,8 @@ func Batch[A any](in <-chan Try[A], size int, timeout time.Duration) <-chan Try[
 	return out
 }
 
-// Unbatch is the inverse of [Batch]. It takes a stream of batches and returns a stream of individual items.
-//
-// This is a non-blocking ordered function that processes items sequentially.
-// See the package documentation for more information on non-blocking ordered functions and error handling.
+// Unbatch flattens a stream of slices into a stream of their values.
+// This function is the inverse of [Batch].
 func Unbatch[A any](in <-chan Try[[]A]) <-chan Try[A] {
 	if in == nil {
 		return nil

benchmark_test.go

@@ -10,9 +10,9 @@ import (
 )
 
 // The stock B/op and allocs/op columns are the per-item metrics reported by
-// benchmarkThroughput, integer divided, so anything under one allocation per
-// item reads as zero. We use custom floating point metrics and suppress the
-// stock ones - even when -benchmem is passed explicitly.
+// benchmarkThroughput, calculated using integer division, so anything under
+// one allocation per item reads as zero. We use custom floating-point metrics
+// and suppress the stock ones - even when -benchmem is passed explicitly.
 // This can only be done by mutating the flag.
 func TestMain(m *testing.M) {
 	flag.Parse()
@@ -54,7 +54,7 @@ func benchmarkThroughput(b *testing.B, definePipeline func(in <-chan Try[int]))
 }
 
 // benchmarkThroughputForLevels runs benchmarkThroughput once per concurrency
-// level, as a subtest named name/n.
+// level as a subtest named name/n.
 func benchmarkThroughputForLevels(b *testing.B, name string, levels []int, definePipeline func(in <-chan Try[int], n int)) {
 	for _, n := range levels {
 		b.Run(fmt.Sprintf("%s/%d", name, n), func(b *testing.B) {
@@ -65,8 +65,8 @@ func benchmarkThroughputForLevels(b *testing.B, name string, levels []int, defin
 	}
 }
 
-// This benchmark acts as baseline. A single drainer is
-// as simple as pipeline can get.
+// This benchmark acts as a baseline. A single drainer is
+// as simple as a pipeline can get.
 func BenchmarkDrain(b *testing.B) {
 	benchmarkThroughput(b, func(in <-chan Try[int]) {
 		Drain(in)

consume.go

@@ -5,15 +5,18 @@ import (
 	"sync/atomic"
 )
 
-// ForEach applies a function f to each item in an input stream and returns the first error encountered.
+// ForEach calls f on the stream's values, like a concurrent for-range
+// loop. It immediately returns the first observed error. Otherwise, it
+// returns nil after the input is fully consumed and every call to f has
+// returned.
 //
-// This is a blocking unordered function that processes items concurrently using n goroutines.
+// The argument n bounds the number of concurrent calls to f. When n = 1,
+// ForEach processes items sequentially in stream order, the same as in a
+// regular for-range loop: f can safely read and modify shared state without
+// synchronization, and all its effects are visible to the caller after
+// ForEach returns.
 //
-// When n = 1, ForEach processes items sequentially in stream order, similar to a regular
-// for-range loop: f can safely read and modify shared state without synchronization,
-// and all its effects are visible to the caller after ForEach returns.
-//
-// See the package documentation for more information on blocking unordered functions and error handling.
+// See the [rill] package documentation for the full contract shared by all sinks.
 func ForEach[A any](in <-chan Try[A], n int, f func(A) error, options ...SinkOption) error {
 	validateN(n)
 	validateNilFunc(f == nil)
@@ -52,10 +55,10 @@ func ForEach[A any](in <-chan Try[A], n int, f func(A) error, options ...SinkOpt
 	return Err(out, options...)
 }
 
-// Err returns the first error encountered in the input stream or nil if there were no errors.
+// Err immediately returns the first error of the stream. Otherwise, it
+// returns nil after the input is fully consumed.
 //
-// This is a blocking ordered function that processes items sequentially.
-// See the package documentation for more information on blocking ordered functions and error handling.
+// See the [rill] package documentation for the full contract shared by all sinks.
 func Err[A any](in <-chan Try[A], options ...SinkOption) error {
 	defer Discard(in, options...)
 
@@ -68,12 +71,11 @@ func Err[A any](in <-chan Try[A], options ...SinkOption) error {
 	return nil
 }
 
-// First returns the first value or error encountered in the input stream.
-// If the stream is empty or its first item is an error, found is false and
-// value is the zero value of A.
+// First returns the first item of the stream: (value, true, nil) if
+// the item is a value, (zero, false, err) if it is an error, or
+// (zero, false, nil) if the stream is empty.
 //
-// This is a blocking ordered function that processes items sequentially.
-// See the package documentation for more information on blocking ordered functions and error handling.
+// See the [rill] package documentation for the full contract shared by all sinks.
 func First[A any](in <-chan Try[A], options ...SinkOption) (value A, found bool, err error) {
 	defer Discard(in, options...)
 
@@ -91,13 +93,15 @@ func First[A any](in <-chan Try[A], options ...SinkOption) (value A, found bool,
 // sharing cannot contaminate across calls.
 var errFound = errors.New("found")
 
-// Any reports whether the input stream contains an item that satisfies the condition f.
-// This function returns true as soon as it finds such an item. Otherwise, it returns false.
+// Any reports whether the stream contains a value that matches the
+// condition f. It immediately returns (true, nil) or (false, err) on
+// the first observed match or error, respectively. Otherwise, it
+// returns (false, nil) after the input is fully consumed and every call
+// to f has returned.
 //
-// Any is a blocking unordered function that processes items concurrently using n goroutines.
-// When n = 1, items are processed sequentially in stream order.
+// The argument n bounds the number of concurrent calls to f.
 //
-// See the package documentation for more information on blocking unordered functions and error handling.
+// See the [rill] package documentation for the full contract shared by all sinks.
 func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error), options ...SinkOption) (bool, error) {
 	validateN(n)
 	validateNilFunc(f == nil)
@@ -119,14 +123,15 @@ func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error), options ...Sin
 	return false, err
 }
 
-// All reports whether all items in the input stream satisfy the condition f.
-// This function returns false as soon as it finds an item that does not satisfy the condition or encounters an error.
-// Otherwise, it returns true.
+// All reports whether every value in the stream matches the condition
+// f. It immediately returns (false, nil) or (false, err) on the first
+// observed mismatch or error, respectively. Otherwise, it returns
+// (true, nil) after the input is fully consumed and every call to f has
+// returned.
 //
-// All is a blocking unordered function that processes items concurrently using n goroutines.
-// When n = 1, items are processed sequentially in stream order.
+// The argument n bounds the number of concurrent calls to f.
 //
-// See the package documentation for more information on blocking unordered functions and error handling.
+// See the [rill] package documentation for the full contract shared by all sinks.
 func All[A any](in <-chan Try[A], n int, f func(A) (bool, error), options ...SinkOption) (bool, error) {
 	validateN(n)
 	validateNilFunc(f == nil)