[serve] Prevent ASGI receive() from hanging on disconnects - #44647

Why are these changes needed?

See #44644 for the user-reported issue.

In #43713, I changed the behavior of the ASGI receive interface to not raise an exception but rather silently hang if the request was not present in the proxy. This was done under the assumption that the code calling receive would be subsequently cancelled anyways. However, this assumption was faulty for a few reasons:

• The user may ignore the CancelledError but still want to fetch ASGI messages.
• Cancellation is best-effort. Specifically, because we are currently using the run_coroutine_threadsafe / concurrent.futures.Executor interface, there is no way to guarantee that either the scheduling task is cancelled or the resulting object ref is cancelled. In rare cases, neither may happen (see asyncio.run_coroutine_threadsafe leaves underlying cancelled asyncio task running python/cpython#105836).

This PR mitigates the issue by returning an appropriate disconnect ASGI message when the proxy no longer has the request in scope.

Note that this is not a perfect solution: we may still drop the messages prior to the disconnect (e.g., the body of the HTTP request) if the replica does not fetch them prior to the client disconnecting.

• Note that this is not a regression from the current behavior and should only happen in practice if the client disconnects very quickly after initiating the request.
• As a follow-up, we should rework this codepath to something more robust. For example we could use a push-based model where the proxy instead pushes messages eagerly to the replica (this was originally very difficult to implement but should be possible now).

TODO before merging:

• Manually verify the fix using the reproduction script in the issue.

Related issue number

Closes #44644

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(