Request resources may be released at the wrong time

[travisdowns]

Version & Environment

Redpanda version: 148f82aa3

What went wrong and what should have happened instead?

Request resources such as the memory semaphore units should be released when the response has been written out to the socket, and the response object and any other usespace buffers have been destroyed. However, this may not be the case because of the way we process responses in order.

Additional information

Find below the processing for the "second stage" of a request, which involves generating the response. See below for analysis.

                    /**
                     * second stage processed in background.
                     */
                    ssx::background
                      = ssx::spawn_with_gate_then(
                          _rs.conn_gate(),
                          [this, f = std::move(f), seq, correlation]() mutable {
                              return f.then([this, seq, correlation](
                                              response_ptr r) mutable {
                                  r->set_correlation(correlation);
                                  _responses.insert({seq, std::move(r)});
                                  return process_next_response();
                              });
                          })
                          .handle_exception([self](std::exception_ptr e) {
                              // snip
                          })
                          .finally([s = std::move(s), self] {});

After the response is ready, we insert it to _responses queue and call process_next_response. At the very bottom we move s into a finally clause. It is s (a session_resources object) that holds the resources associated with the request, so it is destroyed after the future from process_next_response resolves. However this future does not necessarily resolve after the response has finished (response destroyed, etc): if may in fact resolve immediately if the response isn't the next response in sequence.

For example: requests A and B are sent on the connection in that order, and request B finishes its second stage first: it will not be sent by the process_next_response call when it finishes, but remain enqueued, but its resources are destroyed at this point. Only when A completes both A and B (in that order) will be sent.

[piyushredpanda]

I saw a draft PR already, hence assigning to you, @travisdowns