Latu.Retry (latu v0.4.0)

Copy Markdown View Source

When a failed RPC is tried again, and how long Latu waits in between.

One of these sits on every %Latu.Session{}, so the session is where the policy changes:

Latu.connect!("sc://localhost:15002", retry: [max_retries: 3, max_backoff: 5_000])

The defaults are PySpark's own DefaultPolicy, so Latu retries exactly as PySpark does out of the box: 15 attempts, 50ms growing fourfold to a 60s ceiling, jittered once the wait is long enough for jitter to matter. The budget covers consecutive failures while trying to advance a single execution; any response from the server starts a fresh one.

Times are in milliseconds. max_retries: 0 turns retrying off, which is what a test that wants a failure to surface immediately wants.

What is retried is not configurable, and deliberately — retryable?/1 is PySpark's list: an UNAVAILABLE, a disconnected cursor, anything the server attached a RetryInfo to, and, for an execution, a lost handle with nothing received yet. A RetryInfo's delay is a floor under the backoff, capped at max_server_retry_delay.

Summary

Types

t()

Functions

new(opts \\ [])

Build a policy, validating it.

retryable?(error)

Whether a failed call is worth trying again — PySpark's DefaultPolicy.can_retry.

wait(retry, attempt, floor \\ nil)

How long to wait before attempt attempt, counting from zero.

Types

t()

@type t() :: %Latu.Retry{
  backoff_multiplier: number(),
  initial_backoff: non_neg_integer(),
  jitter: non_neg_integer(),
  max_backoff: non_neg_integer(),
  max_retries: non_neg_integer(),
  max_server_retry_delay: non_neg_integer(),
  min_jitter_threshold: non_neg_integer()
}

Functions

new(opts \\ [])

@spec new(keyword() | t()) :: t()

Build a policy, validating it.

iex> Latu.Retry.new(max_retries: 3).max_retries
3

A %Latu.Retry{} passes straight through, so Latu.connect/2 takes either form.

retryable?(error)

@spec retryable?(Latu.Error.t()) :: boolean()

Whether a failed call is worth trying again — PySpark's DefaultPolicy.can_retry.

UNAVAILABLE; INTERNAL naming a disconnected cursor; or any status at all when the server attached a RetryInfo, which is how a gateway says "later" rather than "no".

wait(retry, attempt, floor \\ nil)

@spec wait(t(), non_neg_integer(), non_neg_integer() | nil) :: non_neg_integer()

How long to wait before attempt attempt, counting from zero.

iex> Latu.Retry.wait(Latu.Retry.new(), 0)
50

A floor — the server's RetryInfo delay, when it sent one — lifts the wait to at least that, capped at max_server_retry_delay; jitter goes on after. Jitter above min_jitter_threshold is the only nondeterminism in the transport, and it is bounded by jitter.