Rill: bounded concurrency as a chain of channel functions

Every concurrent function has an n argument

The API is a set of functions over plain Go channels, and the shape of every signature is the design. A stream goes in, a stream comes out, and between them sits an integer that says how many of the callbacks may run at once.

In the quick start, a slice of ids becomes a channel, that channel is mapped with a concurrency of 3 into a channel of users, and the users are consumed with a concurrency of 2:

ctx, scope := rill.WithContext(ctx)

ids := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

users := rill.Map(ids, 3, func(id int) (*api.User, error) {
	return api.GetUser(ctx, id)
})

err := rill.ForEach(users, 2, func(u *api.User) error {
	if u.IsActive {
		return nil
	}
	u.IsActive = true
	return api.SaveUser(ctx, u)
}, scope)

Three things fall out of that. Concurrency is a property of a stage rather than of the program, so two stages running at 3 and 2 is normal rather than surprising. The consumer is just another stage, so swapping `ForEach` for `ToSlice` turns the same pipeline into one that collects results instead of acting on them. And the errors from both the fetch and the save end up in one place.

The concurrency bound is what makes backpressure real. Because there are no per-item allocations or goroutines and items are processed as they arrive, the same code handles a small slice, a stream larger than memory, or an endless one. The library describes itself as not a framework for exactly this reason: there is nothing to register, no base type to embed, and nothing that owns your goroutines.

Errors flow downstream and the scope drains before it returns

The error and cancellation model is the other half of the design and the part most worth understanding before you build on it.

Errors are values travelling on the stream rather than something collected out of band. The README describes this as centralized error handling: errors travel downstream alongside the values and are handled at the end of the pipeline, and they can also be intercepted mid-pipeline where a stage needs to react to them. That last part is what makes things like a retry or a fallback stage expressible without a side channel.

Cancellation is structured and fail-fast. `rill.WithContext(ctx)` returns a derived context and a scope, and terminal stages such as `ForEach` and `First` treat the scope as a functional option. On the first error, the scope cancels the context, waits until nothing is running any more, and only then returns that error. The waiting is the point: a function that returned as soon as it saw an error would leave goroutines writing to channels nobody is reading, and this library does not do that.

What that model does not give you is per-item partial success. If you need to know which of a thousand items failed and keep the other 999, you are outside what these helpers express, and you are back to handling errors per item in your own callback.

The same drain-then-return behaviour shows up in `First`, which the README describes as cancelling the context and waiting for the pipeline to finish on the first match or error. Consistency across terminal stages is what makes these functions composable without surprising leaks.

For structured concurrency the comparison to draw is `errgroup`: cancel on first error, wait for everything to stop. That is exactly the semantics here, made available without importing anything.

Batch and Unbatch turn N calls into one

The batching example is the most immediately useful part of the library, because it is where performance and cost both come from. Batching reduces the number of queries and API calls, raises throughput, and typically lowers cost, and the cases that benefit most are external services and databases.

The mechanism is two functions. `Batch` takes a stream of individual items and produces a stream of slices, and `Unbatch` flattens a stream of slices back into a stream of items. `Batch` takes a batch size and a timeout, and the quick start passes -1 for the timeout, which is how you express no timeout at all:

idBatches := rill.Batch(ids, 5, -1)

userBatches := rill.Map(idBatches, 3, func(ids []int) ([]*api.User, error) {
	return api.GetUsers(ctx, ids)
})

users := rill.Unbatch(userBatches)

The stages before and after the batch are unchanged. Only the middle stage's callback signature differs, which is the composition property doing the work: a stage that operates on batches sits between two stages that operate on items without either knowing.

The timeout argument is the part to think about. With a fixed size and no timeout, a slow producer can leave a partial batch waiting indefinitely, which is fine for a finite slice and wrong for an unbounded stream. The README also describes a real-time use for the same function, where an `UpdateUserTimestamp` call fires on every user action and batching decouples that hot path from the write, grouping independent operations as they arrive rather than waiting for a set size.

Beyond batching, the library ships order preservation, non-commutative streaming reduction, map-reduce, splitting and merging. Pipelines are usually linear but the README notes they can form any cycle-free topology.

Ordered functions trade latency for output order

Ordinary concurrent code writes results in completion order, which depends on how the Go runtime schedules goroutines and how long each result takes. That is usually fine and occasionally wrong.

The ordered variants exist for the case where it is wrong. `OrderedMap` and `OrderedFilter` stay concurrent, but each worker holds its result until every earlier result has been sent, so the output order matches the input. The README names the cost plainly: some latency.

The example makes the value concrete. Checking a thousand hosted files for the first one containing a given string is slow sequentially, and a conventional concurrent filter finds the fastest match rather than the first. Combining `OrderedFilter` with `First` finds the first, while keeping at most five files in memory at a time, and it generates its work with `Generate`, a producer that stops when the context is cancelled:

urls := rill.Generate(func(send func(string), sendError func(error)) {
	for i := 0; i < 1000 && ctx.Err() == nil; i++ {
		send(fmt.Sprintf("https://example.com/file-%d.txt", i))
	}
})

matchedUrls := rill.OrderedFilter(urls, 5, func(url string) (bool, error) {
	content, err := api.DownloadFile(ctx, url)
	if err != nil {
		return false, err
	}
	return bytes.Contains(content, needle), nil
})

The guarantee covers errors as well as values, which is the part that makes ordering trustworthy rather than approximate. If worker seven fails while worker nine is already done, the error surfaces in position seven.

The memory bound is the other benefit worth noting. Bounding concurrency at 5 bounds how much is downloaded at once, so a thousand large files do not have to fit in memory. The same property is what makes the streaming claims hold for inputs larger than memory.

Small, typed, dependency-free, and at version 0.9

Rill is MIT licensed, has zero dependencies, and declares Go 1.25 in its `go.mod`. The last push was on 2026-09-29 and the newest release is v0.9.0, cut the same day, after v0.8.1 in February and v0.8.0 in August 2025.

Two things about that version line are worth weighing.

The Go 1.25 requirement is a real constraint for adoption. It is a recent toolchain, and pinning to it means your CI images and developer machines have to move with it.

The repository layout says something about the scope. Flat files at the root map one to one onto the API surface: `batch.go`, `consume.go`, `iter.go`, `merge.go`, `reduce.go`, `transform.go`, `wrap.go`, `options.go` and `util.go`, each with its own test file and several with benchmark tests. Under `internal/` and `mockapi/` sit the implementation details and the fake API the examples and tests run against. There is no `cmd/`, no plugin directory and no server component, which is what a library that stays inside your binary should look like.

There is a `.golangci.yaml` and a `.pre-commit-config.yaml`, so the project lints itself on the same footing it asks of its users.

Editorial conclusion

Rill fits Go services that call an external API or database many times per request and currently manage goroutines, `sync.WaitGroup` and cancellation by hand, since it removes that code without replacing Go's channel model. Do not reach for it when your concurrency is a handful of goroutines you can already read at a glance, or when you need per-item cancellation beyond the first error, because the cancellation model here is fail-fast and whole-pipeline. Verify first that your module can move to Go 1.25, that `go get github.com/destel/rill` resolves the version you expect against v0.9.0, and that the ordered functions meet your latency budget, since `OrderedMap` and `OrderedFilter` deliberately trade latency for output order.

Frequently asked questions

Official sources

1. destel/rill on GitHub
2. Issues
3. License: MIT
4. README
5. Releases