Fix forwardingStreamHandler errgroup race

[moroten]

The commit Add generic gRPC stream forwarding (#306), introduced a flakiness in the grpc_test due to calling errgroup.Go(func() error { return err }) when an error had occurred in a go routine that was asynchrounous to the errgroup.Wait() call. This could lead to errgroup.Go() being called just when errgroup.Wait() was about to return, which is not allowed.

[aspect-workflows]

Test

2 test targets passed

Targets

//pkg/blobstore/sharding/integration:integration_test [k8-fastbuild]                 85ms
//pkg/grpc:grpc_test [k8-fastbuild]                                                  100ms

Total test execution time was 185ms. 27 tests (93.1%) were fully cached saving 6s.

[moroten]

This PR now adds program.Group.GoAsync to generalize the support needed to remove the errgroup.Go(func() { return err }) lines inside asynchronous go routines. Does that look alright?

[EdSchouten]

I don't know. I'd rather not make program.Group fat. Stuff should just clean up their own goroutines. What are your thoughts on doing something like this instead?

var propagatedError atomic.Pointer[error]
ctxWithCancel, cancel := context.WithCancel(ctx)
defer cancel()

x, err := NewStream()
...

If we want to propagate errors from the background goroutine, we do this:

propagatedError.Store(err)
cancel()

[moroten]

Okay, I resolved it with context.WithCancelCause.

[EdSchouten]

Completely forgot about CancelCauseFunc. Nice!