Function retry_async 

pub async fn retry_async<T, Op, Fut, IsRetryable, E>(
    policy: &NetworkRetryPolicy,
    scheme: &'static str,
    operation: &'static str,
    op: Op,
    is_retryable: IsRetryable,
    metrics: Option<&dyn MetricsCollector>,
) -> Result<T, E>
where
    Op: FnMut() -> Fut,
    Fut: Future<Output = Result<T, E>>,
    IsRetryable: Fn(&E) -> bool,
    E: Display,

Executes op with reconnect/backoff according to policy.

scheme and operation identify the retrying component in retry log messages (structured scheme/operation fields, e.g. ws/connect: transient error — retrying) and in metrics.

When metrics is Some, every attempt (the first included) is recorded via MetricsCollector::increment_retry_attempt, and a final failure — attempts exhausted or a non-retryable first error — records exactly ONE error via increment_errors(operation, "e:{scheme}:{operation}") before returning it. Call sites must not also count the same exhaustion in their Err arms (one error per exhausted retry sequence).

is_retryable classifies errors: retryable errors are retried, permanent errors are not.

Security note: error Display is logged at WARN level

On every retry, the error’s Display representation is emitted via tracing::warn! for operator visibility during connection retries. Callers MUST sanitize errors before returning them from op if they may contain sensitive content such as connection strings, embedded credentials, or host‑port pairs. Sanitization belongs at the source — in the IO call whose error is wrapped — not here.

This log call is intentional and should not be removed: it provides the only diagnostic signal that a networked component is retrying and why.

Example

let result = retry_async(
    &config.reconnect,
    "ws",
    "connect",
    || async move { connect_to_server().await },
    |err: &CamelError| matches!(err, CamelError::Io(_)),
    metrics,
).await?;