httpx error deserialization in prefect_ray.RayTaskRunner

[ka7eh]

What happened + What you expected to happen

I'm using prefect_ray.RayTaskRunner with my Prefect flows. I'm using httpx in Prefect tasks for some API calls. I noticed when httpx raises an HTTPStatusError, ray fails to deserialize the error and throws the following error:

ray.exceptions.RaySystemError: System error: HTTPStatusError.__init__() missing 2 required keyword-only arguments: 'request' and 'response'
traceback: Traceback (most recent call last):
  File "/path/to/project/venv/lib/python3.11/site-packages/ray/_private/serialization.py", line 423, in deserialize_objects
    obj = self._deserialize_object(data, metadata, object_ref)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/path/to/project/venv/lib/python3.11/site-packages/ray/_private/serialization.py", line 280, in _deserialize_object
    return self._deserialize_msgpack_data(data, metadata_fields)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/path/to/project/venv/lib/python3.11/site-packages/ray/_private/serialization.py", line 235, in _deserialize_msgpack_data
    python_objects = self._deserialize_pickle5_data(pickle5_data)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/path/to/project/venv/lib/python3.11/site-packages/ray/_private/serialization.py", line 225, in _deserialize_pickle5_data
    obj = pickle.loads(in_band)
          ^^^^^^^^^^^^^^^^^^^^^
TypeError: HTTPStatusError.__init__() missing 2 required keyword-only arguments: 'request' and 'response'

Running without RayTaskRunner throws the right error, like this:

httpx.HTTPStatusError: Server error '500 INTERNAL SERVER ERROR' for url 'https://httpbin.org/status/500'
For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500

Is this an issue in ray? Or prefect_ray? Or am I doing something wrong?

Versions / Dependencies

OS: Ubuntu 22.04
Python: 3.11

ray==2.31.0
prefect==2.19.7
prefect-ray==0.3.6

Reproduction script

import httpx
from prefect import flow, task
from prefect_ray import RayTaskRunner

@task(name="Get status")
def get_status() -> str:
    resp = httpx.get("https://httpbin.org/status/500")
    resp.raise_for_status()
    return resp.text


@flow(name="HTTPX Error Handling", task_runner=RayTaskRunner, log_prints=True)
def httpx_error_handling() -> None:
    status_result = get_status.submit()
    print(status_result.result())

httpx_error_handling()

Issue Severity

High: It blocks me from completing my task.

[rynewang]

Looks like httpx status errors are not serializable? We need classes to follow https://docs.ray.io/en/latest/ray-core/objects/serialization.html#customized-serialization

[rynewang]

As a workaround you can wrap the raised exception in your own serializable exception class. If you still have issues for this, feel free to reopen this issue.