On master @ 58cee73, source/channel.go:48-57: after an item has been consumed from the user's Input channel, the forwarding select races cancellation against delivery:
case item, ok := <-s.Input:
...
select {
case <-ctx.Done():
return
case out <- item:When ctx is already canceled, Go's pseudo-random select choice can take the ctx.Done() arm even though out has buffer space — the item was irreversibly dequeued from the caller's channel but never enters the pipeline, with no error emitted. That contradicts the library's documented stance that items already read are still processed. The same pattern can drop one in-flight error in source/error.go (~:60).
Trigger: cancel ctx while the source is forwarding; lose up to one item (or one error) per cancellation.
Options: (a) attempt a non-blocking send to out before honoring ctx.Done(); (b) restructure so the dequeue and delivery are a single committed step (only select on ctx.Done() before consuming from Input, then deliver unconditionally — safe for the engine, which drains until close, but should be documented for non-engine consumers); or (c) document the at-most-one-loss semantics explicitly. (a) or (b) preferred — silent loss is the worst of the three.
Relations: complements #76 (CancelStop mode); found in a deep review of the source package.
On master @ 58cee73,
source/channel.go:48-57: after an item has been consumed from the user'sInputchannel, the forwarding select races cancellation against delivery:When ctx is already canceled, Go's pseudo-random select choice can take the
ctx.Done()arm even thoughouthas buffer space — the item was irreversibly dequeued from the caller's channel but never enters the pipeline, with no error emitted. That contradicts the library's documented stance that items already read are still processed. The same pattern can drop one in-flight error insource/error.go(~:60).Trigger: cancel ctx while the source is forwarding; lose up to one item (or one error) per cancellation.
Options: (a) attempt a non-blocking send to
outbefore honoringctx.Done(); (b) restructure so the dequeue and delivery are a single committed step (only select onctx.Done()before consuming from Input, then deliver unconditionally — safe for the engine, which drains until close, but should be documented for non-engine consumers); or (c) document the at-most-one-loss semantics explicitly. (a) or (b) preferred — silent loss is the worst of the three.Relations: complements #76 (CancelStop mode); found in a deep review of the source package.