Function execute_with_policy 

pub async fn execute_with_policy<F, Fut, T>(
    policy: &RetryPolicy,
    cancel: Option<&CancellationToken>,
    op: F,
) -> Result<T, FaucetError>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, FaucetError>>,

Execute op under policy. Retries retriable errors (per RetryPolicy::is_retriable) up to max_attempts total attempts, sleeping the backoff delay between attempts. If cancel is provided, a cancellation during a backoff sleep stops waiting and returns the last error immediately (the caller observes the token and flushes). Ok returns at once.

This is the connector-facing, label-free runner used by source connectors; the pipeline loop uses execute_with_policy_metered so the resilience metrics get their {pipeline, row, op} labels.