Skip to content

Add pipeline settlement signal - #108

Merged
destel merged 23 commits into
mainfrom
f/settlement
Aug 28, 2026
Merged

destel merged 23 commits into
mainfrom
f/settlement

Conversation

@destel

@destel destel commented Aug 27, 2026

Copy link
Copy Markdown
Owner

This feature was replaced by the Scope API (#109)

Blocking sinks in rill return as soon as their outcome is known. That's deliberate – the sooner we get
control back, the sooner we can cancel the source that might still be producing work, and whatever is
still in flight across the pipeline stages.

What that model doesn't give us is an "everything is done" event. Sometimes we need one: to free a
shared resource, or to safely look at side effects the callbacks produced. Basically
sync.WaitGroup.Wait(), but for a pipeline. That's what we're adding.

ctx, cancel := context.WithCancel(ctx)
defer cancel()

// other pipeline stages go here

settled, opt := rill.Settlement()

err := rill.ForEach(input, 5, func(x int) error {
	return process(ctx, x)
}, opt)

if err != nil {
	// handle
}

// stop the source and anything still in flight
cancel()  

// nothing from the pipeline runs after this
<-settled 

// now it's safe to free shared resources and observe side effects

One catch, and it's the whole reason cancel() is in that snippet: settlement doesn't stop anything, it
only tells us when the pipeline ran out of work. If the source keeps producing, we keep waiting,
so stopping the source has to come first.

How it works

We say a pipeline stage is settled once it won't do any more work: it has fully consumed its input, and
every callback it started has returned.

Most stages already expose this, they just don't call it that. Map, Filter, FromSlice and friends
close their output channel only after they've settled, so a closed output is the settlement signal.
Sinks (e.g. ForEach, ToSlice, Reduce) are the hole – no output channel, nothing to close, no signal. We give them an equivalent one.

The useful part is that it composes backwards. A settled sink means the stage before it closed its
output, which means that stage settled, which means the stage before it closed its output, and so on up
to the source. One signal at the end of a linear pipeline covers the whole thing. And it's a real
happens-before edge, not a timing coincidence – the tests pin that.

Two things to know:

  • Branching pipelines (anything with Tee) need one settlement per terminal sink. A settled sink says
    nothing about callbacks still running in a sibling branch.
  • Custom stages are covered as long as they play by the same rule: don't close the output until the
    input is consumed and the work is done.

Compatibility

Existing call sites keep working, since the options are variadic. But every sink's signature changed, so
code that assigns a sink to an exact function type has to be updated. Technically, this is a breaking
API change.

@destel destel added the new label Aug 27, 2026
@codecov

codecov Bot commented Aug 27, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 98.30508% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 99.87%. Comparing base (0526448) to head (5a9b950).

Files with missing lines Patch % Lines
util.go 91.66% 1 Missing ⚠️
Additional details and impacted files
@@            Coverage Diff             @@
##             main     #108      +/-   ##
==========================================
+ Coverage   99.76%   99.87%   +0.11%     
==========================================
  Files          15       16       +1     
  Lines         837      788      -49     
==========================================
- Hits          835      787      -48     
+ Misses          2        1       -1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@destel
destel merged commit 8f859f2 into main Aug 28, 2026
5 of 6 checks passed
@destel
destel deleted the f/settlement branch August 28, 2026 07:48
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant