From 88d5585c07983681493283acc3068f088abe5a30 Mon Sep 17 00:00:00 2001 From: destel Date: Sun, 26 Jul 2026 14:57:27 +0300 Subject: [PATCH 1/2] Validate arguments at the public API boundary --- batch.go | 2 ++ consume.go | 9 +++++++++ iter.go | 4 ++++ reduce.go | 8 ++++++++ transform.go | 30 ++++++++++++++++++++++++++++++ util.go | 26 +++++++++++++++++++++++++- wrap.go | 2 ++ 7 files changed, 80 insertions(+), 1 deletion(-) diff --git a/batch.go b/batch.go index bcf6d20..6e74f97 100644 --- a/batch.go +++ b/batch.go @@ -20,6 +20,8 @@ import ( // // See the package documentation for more information on non-blocking ordered functions and error handling. func Batch[A any](in <-chan Try[A], size int, timeout time.Duration) <-chan Try[[]A] { + validateMinSize(size, 1) + values, errs := ToChans(in) batches := core.Batch(values, size, timeout) return FromChans(batches, errs) diff --git a/consume.go b/consume.go index f60f6e3..0e487aa 100644 --- a/consume.go +++ b/consume.go @@ -15,6 +15,9 @@ import ( // // See the package documentation for more information on blocking unordered functions and error handling. func ForEach[A any](in <-chan Try[A], n int, f func(A) error) error { + validateN(n) + validateNilFunc(f == nil) + // The n = 1 path is an internal contract, not just an optimization. // Other sinks (Any, Reduce) build their n = 1 behavior on it and rely on: // - items processed sequentially, in stream order @@ -96,6 +99,9 @@ var errFound = errors.New("found") // // See the package documentation for more information on blocking unordered functions and error handling. func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error) { + validateN(n) + validateNilFunc(f == nil) + err := ForEach(in, n, func(a A) error { ok, err := f(a) if err != nil { @@ -122,6 +128,9 @@ func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error) // // See the package documentation for more information on blocking unordered functions and error handling. func All[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error) { + validateN(n) + validateNilFunc(f == nil) + err := ForEach(in, n, func(a A) error { ok, err := f(a) if err != nil { diff --git a/iter.go b/iter.go index dd941b8..a3b6d58 100644 --- a/iter.go +++ b/iter.go @@ -19,6 +19,8 @@ func FromSeq[A any](seq iter.Seq[A], err error) <-chan Try[A] { return out } + validateNilFunc(seq == nil) + out := make(chan Try[A]) go func() { for val := range seq { @@ -31,6 +33,8 @@ func FromSeq[A any](seq iter.Seq[A], err error) <-chan Try[A] { // FromSeq2 converts an iterator of value-error pairs into a stream. func FromSeq2[A any](seq iter.Seq2[A, error]) <-chan Try[A] { + validateNilFunc(seq == nil) + out := make(chan Try[A]) go func() { for val, err := range seq { diff --git a/reduce.go b/reduce.go index 2a4009d..35f682d 100644 --- a/reduce.go +++ b/reduce.go @@ -122,6 +122,9 @@ func reduceStage[A any](in Stream[A], n int, f func(A, A) (A, error)) Stream[A] // // See the package documentation for more information on blocking unordered functions and error handling. func Reduce[A any](in <-chan Try[A], n int, f func(A, A) (A, error)) (result A, hasResult bool, err error) { + validateN(n) + validateNilFunc(f == nil) + var zero A if n == 1 { @@ -247,6 +250,11 @@ func mapReduceStage[A any, K comparable, V any](in Stream[A], nm int, mapper fun // // See the package documentation for more information on blocking unordered functions and error handling. func MapReduce[A any, K comparable, V any](in <-chan Try[A], nm int, mapper func(A) (K, V, error), nr int, reducer func(V, V) (V, error)) (map[K]V, error) { + validateN(nm) + validateNilFunc(mapper == nil) + validateN(nr) + validateNilFunc(reducer == nil) + if nm == 1 && nr == 1 { m := make(map[K]V) err := ForEach(in, 1, func(a A) error { diff --git a/transform.go b/transform.go index 68ad8af..68186cc 100644 --- a/transform.go +++ b/transform.go @@ -12,6 +12,9 @@ import ( // // See the package documentation for more information on non-blocking unordered functions and error handling. func Map[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B] { + validateN(n) + validateNilFunc(f == nil) + return core.FilterMap(in, n, func(a Try[A]) (Try[B], bool) { if a.Error != nil { return Try[B]{Error: a.Error}, true @@ -28,6 +31,9 @@ func Map[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B] // OrderedMap is the ordered version of [Map]. func OrderedMap[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B] { + validateN(n) + validateNilFunc(f == nil) + return core.OrderedFilterMap(in, n, func(a Try[A]) (Try[B], bool) { if a.Error != nil { return Try[B]{Error: a.Error}, true @@ -50,6 +56,9 @@ func OrderedMap[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan // // See the package documentation for more information on non-blocking unordered functions and error handling. func Filter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A] { + validateN(n) + validateNilFunc(f == nil) + return core.FilterMap(in, n, func(a Try[A]) (Try[A], bool) { if a.Error != nil { return a, true // never filter out errors @@ -66,6 +75,9 @@ func Filter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[ // OrderedFilter is the ordered version of [Filter]. func OrderedFilter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A] { + validateN(n) + validateNilFunc(f == nil) + return core.OrderedFilterMap(in, n, func(a Try[A]) (Try[A], bool) { if a.Error != nil { return a, true // never filter out errors @@ -89,6 +101,9 @@ func OrderedFilter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-ch // // See the package documentation for more information on non-blocking unordered functions and error handling. func FilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B] { + validateN(n) + validateNilFunc(f == nil) + return core.FilterMap(in, n, func(a Try[A]) (Try[B], bool) { if a.Error != nil { return Try[B]{Error: a.Error}, true @@ -105,6 +120,9 @@ func FilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <- // OrderedFilterMap is the ordered version of [FilterMap]. func OrderedFilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B] { + validateN(n) + validateNilFunc(f == nil) + return core.OrderedFilterMap(in, n, func(a Try[A]) (Try[B], bool) { if a.Error != nil { return Try[B]{Error: a.Error}, true @@ -127,6 +145,9 @@ func OrderedFilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, err // // See the package documentation for more information on non-blocking unordered functions and error handling. func FlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B] { + validateN(n) + validateNilFunc(f == nil) + if in == nil { return nil } @@ -150,6 +171,9 @@ func FlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan // OrderedFlatMap is the ordered version of [FlatMap]. func OrderedFlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B] { + validateN(n) + validateNilFunc(f == nil) + if in == nil { return nil } @@ -185,6 +209,9 @@ func OrderedFlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) // // See the package documentation for more information on non-blocking unordered functions and error handling. func Catch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A] { + validateN(n) + validateNilFunc(f == nil) + return core.FilterMap(in, n, func(a Try[A]) (Try[A], bool) { if a.Error == nil { return a, true @@ -201,6 +228,9 @@ func Catch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A] { // OrderedCatch is the ordered version of [Catch]. func OrderedCatch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A] { + validateN(n) + validateNilFunc(f == nil) + return core.OrderedFilterMap(in, n, func(a Try[A]) (Try[A], bool) { if a.Error == nil { return a, true diff --git a/util.go b/util.go index 3e08074..2b880a1 100644 --- a/util.go +++ b/util.go @@ -1,6 +1,10 @@ package rill -import "github.com/destel/rill/internal/core" +import ( + "fmt" + + "github.com/destel/rill/internal/core" +) // Drain consumes and discards all items from an input channel, blocking until the channel is closed. func Drain[A any](in <-chan A) { @@ -31,5 +35,25 @@ func DrainNB[A any](in <-chan A) { // // Now work with the users channel as usual. // // Up to 100 users can be buffered if subsequent stages of the pipeline are slow. func Buffer[A any](in <-chan A, size int) <-chan A { + validateMinSize(size, 1) + return core.Buffer(in, size) } + +func validateN(n int) { + if n < 1 { + panic(fmt.Sprintf("rill: n must be at least 1, got %d", n)) + } +} + +func validateMinSize(size int, minSize int) { + if size < minSize { + panic(fmt.Sprintf("rill: size must be at least %d, got %d", minSize, size)) + } +} + +func validateNilFunc(fIsNil bool) { + if fIsNil { + panic("rill: function must not be nil") + } +} diff --git a/wrap.go b/wrap.go index c104d3d..0dca438 100644 --- a/wrap.go +++ b/wrap.go @@ -209,6 +209,8 @@ func ToChans[A any](in <-chan Try[A]) (<-chan A, <-chan error) { // stream <- rill.Try[int]{Error: someError} // }() func Generate[A any](f func(send func(A), sendErr func(error))) <-chan Try[A] { + validateNilFunc(f == nil) + out := make(chan Try[A]) go func() { defer close(out) From f8496e8f6772768227703d3851d5e0254b3d9031 Mon Sep 17 00:00:00 2001 From: destel Date: Sun, 26 Jul 2026 15:21:06 +0300 Subject: [PATCH 2/2] Add validation tests: helper boundaries and FromSeq error precedence --- iter_test.go | 11 +++++++++++ util_test.go | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/iter_test.go b/iter_test.go index a351a6b..7b6fa1e 100644 --- a/iter_test.go +++ b/iter_test.go @@ -112,6 +112,17 @@ func TestFromSeq(t *testing.T) { th.ExpectSlice(t, outSlice, expectedSlice) }) + + t.Run("error with nil iterator", func(t *testing.T) { + out := FromSeq[int](nil, errors.New("some error")) + + outSlice := toItemSlice(out) + + var expectedSlice []Item[int] + expectedSlice = appendErr(expectedSlice, errors.New("some error")) + + th.ExpectSlice(t, outSlice, expectedSlice) + }) } func TestFromSeq2(t *testing.T) { diff --git a/util_test.go b/util_test.go index 9717222..3118bc3 100644 --- a/util_test.go +++ b/util_test.go @@ -44,3 +44,39 @@ func TestBuffer(t *testing.T) { Drain(out) }) } + +func TestValidations(t *testing.T) { + t.Run("ok", func(t *testing.T) { + validateN(1) + validateMinSize(5, 5) + validateNilFunc(false) + }) + + t.Run("n too small", func(t *testing.T) { + defer func() { + if r := recover(); r == nil { + t.Errorf("expected panic") + } + }() + + validateN(0) + }) + + t.Run("size too small", func(t *testing.T) { + defer func() { + if r := recover(); r == nil { + t.Errorf("expected panic") + } + }() + validateMinSize(5, 6) + }) + + t.Run("function is nil", func(t *testing.T) { + defer func() { + if r := recover(); r == nil { + t.Errorf("expected panic") + } + }() + validateNilFunc(true) + }) +}