pubsub: Make batch request results independent

[mitsos1os]

Allow batch requests to return results independently without one blocking the other.

This will in turn allow Subscription.Receive to return a message from a request (since the Receive function signature returns only one message), without waiting for all the requests from the batch to complete.

Except for this, package behavior should remain as is...

Fixes: #3456

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[vangent]

Can't this lose errors?

I.e., if getNextBatch gets an error, and writes it to resultChannel.err, and then immediately closes resultChannel.msgs, I think this select could end up here, and we'd never receive the error.

Instead of returning a struct that contains two channels, I think it might be better to return a single channel, which delivers structs (where the struct can either have messages or an error).

[mitsos1os]

I think there might be a bit more to it...

So, given that g.Wait in getNextBatch will return only after each goroutine has exited, it means that any messages sent, will certainly happen before Wait returns.

So when Wait returns, if there was an error it will be written to the error channel which is not buffered . So, this will block the execution until it is consumed. So it won't lose any errors since it won't move to close the channel unless the error gets consumed.

But you may be right about the flow being simpler returning a single channel. I will try this out

[mitsos1os]

@vangent Thank you for your suggestions! I have refactored the execution flow to address them.

This is what you get when coding tired after midnight 😄 ..

Can you please take an other look?

[vangent]

I haven't taken another look, but the tests are still failing with a data race.

You should be able to reproduce this locally with -race....

WARNING: DATA RACE
Read at 0x00c0000bc6b8 by goroutine 2559:
gocloud.dev/pubsub.(*secondReceiveBlockedDriverSub).ReceiveBatch()
/Users/runner/work/go-cloud/go-cloud/pubsub/pubsub_test.go:291 +0x38
gocloud.dev/pubsub.(*Subscription).getNextBatch.func1.1()
/Users/runner/work/go-cloud/go-cloud/pubsub/pubsub.go:674 +0x128
gocloud.dev/internal/retry.call()
/Users/runner/work/go-cloud/go-cloud/internal/retry/retry.go:52 +0x10c
gocloud.dev/internal/retry.Call()
/Users/runner/work/go-cloud/go-cloud/internal/retry/retry.go:40 +0x140
gocloud.dev/pubsub.(*Subscription).getNextBatch.func1()
/Users/runner/work/go-cloud/go-cloud/pubsub/pubsub.go:670 +0x58
golang.org/x/sync/errgroup.(*Group).Go.func1()
/Users/runner/go/pkg/mod/golang.org/x/sync@v0.7.0/errgroup/errgroup.go:78 +0x7c

Previous write at 0x00c0000bc6b8 by goroutine 2560:
gocloud.dev/pubsub.(*secondReceiveBlockedDriverSub).ReceiveBatch()
/Users/runner/work/go-cloud/go-cloud/pubsub/pubsub_test.go:291 +0x4c
gocloud.dev/pubsub.(*Subscription).getNextBatch.func1.1()
/Users/runner/work/go-cloud/go-cloud/pubsub/pubsub.go:674 +0x128
gocloud.dev/internal/retry.call()
/Users/runner/work/go-cloud/go-cloud/internal/retry/retry.go:52 +0x10c
gocloud.dev/internal/retry.Call()
/Users/runner/work/go-cloud/go-cloud/internal/retry/retry.go:40 +0x140
gocloud.dev/pubsub.(*Subscription).getNextBatch.func1()
/Users/runner/work/go-cloud/go-cloud/pubsub/pubsub.go:670 +0x58
golang.org/x/sync/errgroup.(*Group).Go.func1()
/Users/runner/go/pkg/mod/golang.org/x/sync@v0.7.0/errgroup/errgroup.go:78 +0x7c

[mitsos1os]

@vangent the race condition was on test code now... I had forgotten to save the -race flag when running tests locally
It is now fixed as well

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 73.42%. Comparing base (25c3def) to head (e55069e).

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #3457      +/-   ##
==========================================
+ Coverage   73.38%   73.42%   +0.04%     
==========================================
  Files         113      113              
  Lines       14959    14964       +5     
==========================================
+ Hits        10977    10987      +10     
+ Misses       3208     3204       -4     
+ Partials      774      773       -1     

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

[vangent]

pubsub.go looks good now, just a nit on the test. Thanks!

[vangent]

The hardcoded timings might be racy.

Could secondReceiveBlockedDriverSub just wait forever (or maybe until the passed-in ctx is Done) on the second call, and s.Receive wait forever (no context.WithTimeout)? Without your change, the test would hang and eventually timeout; after it, I think the receive should work ~right away (getting the first message) and then exit successfully.

[mitsos1os]

Yeah, I opted in for context timeout because in the case of failure, I didn't want the test to hang for a long amount of time and get a faster failure.

Anyway, I changed it to how you proposed with the wait-forever approach and made sure that on master the test hangs.