kv/bulk: ensure in-flight requests end on batcher reset

[dt]

streamingest: ensure batcher is always reset
kv/bulk: ensure in-flight requests end on batcher reset

Previously if streamIngestionProcessor.flushBuffer() had started adding keys to a batcher, it was possible that that batcher had filled and triggered an async flush. Typically this async flush would be awaited when flushBuffer explicitly calls sip.batcher.Flush() immediately after the loop in which it adds keys. However, if one of the key additions returned an error, one of these async flushes could still be in-flight when that error causes flushBuffer to return early, without calling Flush() and waiting for it. Thus, flushBuffer could return and in doing so finish its local tracing span, that it handed to the batcher and is being used to send its still ongoing async flush, resulting in that request then finding itself using a finished span, which is illegal.

The changes here add waiting to inflight requests to batcher.Reset(), and then defer the reset of the batcher in flushBuffer so that it is always run, including before early returns, ensuring that flushBuffer never returns while its batcher has in-flight requests pending.

While I'm here: removed the unused error return from Reset().

[blathers-crl]

It looks like your PR touches production code but doesn't add or edit any test code. Did you consider adding tests to your PR?

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

[cockroach-teamcity]

This change is 

[msbutler]

Looks great!

[dt]

TFTR!

bors r+

[craig]

Build failed (retrying...):

• Bazel Essential CI (Cockroach)

[dt]

bors r+

[craig]

Already running a review

[craig]

Build succeeded:

• Bazel Essential CI (Cockroach)