Obfuscating batch flushes

[xemul]

The output_stream::flush() looks like this

template <typename CharType>
future<> output_stream<CharType>::flush() noexcept {
    if (!_batch_flushes) {
        return do_sync_flush(_fd);
    } else {
        if (_ex) {
            // flush is a good time to deliver outstanding errors
            return make_exception_future<>(std::move(_ex));
        } else {
            _flush = true;
            if (!_in_batch) {
                add_to_flush_poller(this);
                _in_batch = promise<>();
            }
        }
    }
    return make_ready_future<>();
}

so if batch-flushing takes place the stream.flush() resolves immediately without exception. Later, the .poll_flush() may put and exception from _fd.flush()/_fd.put() int stream's .ex field, but nothing more.

Next, consider an RPC-like message exchange:

    while (true) {
        co_await client_stream.write();
        co_await client_stream.flush();
        co_await client_stream.recv();
        co_await handle();
    }

if sending a message in batch-flush fails no place here ever receives this exception:

• .flush() resolves without exception as was shown above
• .recv() hangs waiting for the response that may never arrive, because request sending failed
• .handle() won't get called
• .write() on next loop iteration may notice the exception, but it won't be called either

[xemul]

Found in #1146