[Core] Add retry exception allowlist for user-defined filtering of retryable application-level errors.

[clarkzinzow]

Both Ray libraries (Workflows, Datasets, etc.) and end-users need to be able to retry application-level errors; this is currently supported via the retry_exceptions argument. However, not all application-level errors are transient, and therefore not all application-level errors should be retried.

This PR adds supported for specifying an exception allowlist (List[Exception]) as the retry_exceptions argument, such that an application-level exception will only be retried if it is in the allowlist.

Exception Allowlist

@ray.remote(
    retry_exceptions=ConnectionError,
    max_retries=3,
)
def f(a):
    if a < 0:
        raise ValueError("invalid input")
    else:
        # connect_and_add might transiently raise a ConnectionError
        return connect_and_add(a)

# Fails without retrying the non-transient error.
ray.get(f.remote(-1))

# Retries up to 3 times.
ray.get(f.remote(5))

See the RFC for more motivation and API design discussion: #25743

Related issue number

Closes #25743

Checks

• 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 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 :(

[fishbone]

With regard to the API, I feel if it's just a list of types (not that flexible as this one) which seems more friendly to the user?

Will this be better:

    retry_exceptions: Union[Exception, List[Exception], Callable[[Exception], bool]]

From the implementation point of view, we can always convert the first two to the third one.

[clarkzinzow]

@iycheng We discussed both in the RFC and internally, we settled on the exception predicate instead of an allowlist/denylist since it's more powerful, obviates the need for two separate list arguments to support both allowlists and denylists, and allows the user to multiplex on error metadata attached to the exception such as error/status codes.

This also appears to be the more popular API choice for application-level retry libraries (not transport-level retry libraries, which standardize on the status codes defined in the underlying transport protocol).

[clarkzinzow]

As for accepting both a predicate function and exceptions, implicitly turning the latter it into an allowlist predicate, I like that more than just having the allowlist! But won't that overly complicate the retry_exceptions type? It's already Union[bool, Callabe[[Exception], bool]], this would turn it into Union[bool, Exception, List[Exception], Callable[[Exception], bool]].

Another option would be providing predicate utilities similar to the Google Cloud Python SDK retry library. Concrete comparison below:

# 1. Status quo: predicate function.
@ray.remote(retry_exceptions=lambda e: isinstance(e, (BrokenPipeError, ConnectionResetError)))
def foo:
    pass

# 2. Predicate function + exception list
@ray.remote(retry_exceptions=[BrokenPipeError, ConnectionResetError])
def foo:
    pass

# 3. Predicate function + helpers
@ray.remote(retry_exceptions=ray.util.retry.if_exception_type([BrokenPipeError, ConnectionResetError]))
def foo:
    pass

I feel like the predicate helpers are too verbose to be worth it... Overall I think that I'm leaning towards 2.

[clarkzinzow]

@iycheng I added the Union[Exception, List[Exception]] convenience option.

[clarkzinzow]

@ericl @rkooo567 @pcmoritz API and implementation ready for review.

[clarkzinzow]

We should never see an empty string coming from the Python frontend, but this can still happen in tests and may be allowed in cross-language scenarios, so we add this to be defensive.

[clarkzinzow]

@jjyao @ericl Feedback has been implemented! I'm planning on merging once the build completes (assuming no new failures), so lmk if there are any tweaks that should be made before merging.

[clarkzinzow]

Failures appear to be unrelated, merging.

[SongGuyang]

@clarkzinzow windows://python/ray/tests:test_output failed after this PR.

[clarng]

want to call out the backwards-incompatibility of this change. Perhaps not that big of a deal today given we don't support rolling upgrades to the cluster, but it is still good to stay in the habit of making only backwards compatible change when possible