fix: retries and address parsing

CHANGELOG.md

@@ -19,6 +19,17 @@ its own schema and says exactly what it is. `Latu.Result.Arrow` is the reader 
 streaming format, no dependency — and `Latu.Result.Nx` the mapping, behind the now-optional
 `:nx`. Adding `{:nx, "~> 0.13"}` is what turns them on; without it `to_nx/2` says so.
 
+**Every RPC retries on the session's `Latu.Retry`**, not only the result stream — PySpark's
+own behaviour — except the best-effort releases. A `RetryInfo` on an error makes it retryable
+whatever its status, with its delay as a floor under the backoff, capped by the new
+`max_server_retry_delay` (10 min); `%Latu.Error{}` gains `retry_delay`. A unary call's
+`[:latu, :retry, :attempt]` event carries `rpc` where an execution's carries `operation_id`.
+Smaller: a cleartext token is allowed to every loopback address, not four spellings of it, and
+an IPv6 literal host connects at all (`sc://[::1]:15002` crashed inside elixir-grpc);
+`SPARK_USER` precedes the OS user as the default `user_id`; `lit/1` refuses a non-UTF-8 binary
+and names Spark's `X'…'` literal; `true`/`false` are refused where a column name is taken
+instead of naming a column `"true"`. No migration.
+
 **`Latu.error_details/2` restores the whole message.** The server abbreviates a gRPC status
 message to 2048 characters, so a long analysis error arrived cut short with `...`; the detail
 carries it whole, and the call now puts it on the error beside the causes. No migration.

docs/decisions.md

@@ -1339,3 +1339,41 @@ gRPC status cut to 2048 characters ending in `...`, and that is what `%Latu.Erro
 the same on the explicit call: `error_details/2` puts the thrown error's message on the struct
 beside the causes. Nothing changes without the call, by the same reasoning that makes the fetch
 explicit; a short message is identical on both paths.
+\n
+## 2026-09-07 — Every RPC retries, `RetryInfo` counts, and four smaller alignments
+
+**Unary RPCs retry on the session's policy.** PySpark wraps every call in `Retrying`; Latu
+retried only the execution stream, so a transient `UNAVAILABLE` on `AnalyzePlan` or `Config`
+failed outright. `Client.retrying/3` now wraps every call that asks the server for something.
+Not `ExecutePlan`/`ReattachExecute` — `Latu.Client.Execution` owns those — and not the releases
+that are best effort: `ReleaseExecute` on the stream, and `ReleaseSession` inside
+`disconnect(release: true)`, which has a five-second budget to keep. Backing off blocks the
+caller's process, as the reattach path already does. The retry event carries `rpc` where an
+execution's carries `operation_id`.
+
+**A `RetryInfo` makes any status retryable, and its delay is a floor.** PySpark's
+`DefaultPolicy` third case; `Latu.Retry.retryable?/1` is now the one predicate, with
+`max_server_retry_delay` (10 min) capping the floor before jitter. Spark's own server never
+attaches one — `ErrorUtils`/`SparkConnectService` on 4.2 have no `RetryInfo` — so this is for
+a gateway in front of it. The old comment blaming elixir-grpc for not decoding the detail was
+wrong: `details` is the same `Any` list `ErrorInfo` is read from.
+
+**Loopback is what the kernel keeps on the machine**: `localhost`, 127/8, `::1` in any
+spelling, and 127/8 as an IPv4-mapped address — parsed with `:inet.parse_address/1` rather than
+listed. PySpark checks the string `"localhost"` alone. Testing it found that an IPv6 literal
+never connected: `"host:port"` compatibility mode splits on colons, so the target takes
+elixir-grpc's `ipv6:[addr]:port` form when the host parses as one.
+
+**`SPARK_USER` comes before the OS user** for the default `user_id`, PySpark's own order. The
+`user-agent` header stays unset: PySpark sends its agent only as `client_type`, so there was no
+parity to gain.
+
+**A binary literal must be UTF-8**: the proto field is a string and the server refuses invalid
+bytes with an opaque error; `lit/1` refuses first and names Spark's `X'…'` literal. **Booleans
+are refused wherever a name is taken**: they are atoms, and `to_name(true)` was a column called
+`"true"`.
+
+Not done: caching the `create_dataframe/3` config probe on the returned session. A caller
+rarely threads `df.session` back, so the saving would seldom land, and a cached threshold goes
+stale the moment `set_conf/3` moves it — SparkEx shipped exactly that bug. One round trip per
+call stays.

lib/latu.ex

@@ -80,7 +80,7 @@ defmodule Latu do
     * `:keepalive` — HTTP/2 ping interval in milliseconds. Defaults to `60_000`.
     * `:keepalive_tolerance` — pings missed before the channel is considered dead. Defaults
       to `2`.
-    * `:retry` — a `Latu.Retry` policy every execution retries under. Defaults to `%Latu.Retry{}`.
+    * `:retry` — a `Latu.Retry` policy every RPC retries under. Defaults to `%Latu.Retry{}`.
 
   Why these numbers, and why the knobs live on the session rather than in application config,
   is in `docs/decisions.md`.
@@ -779,7 +779,9 @@ defmodule Latu do
   name. When no data name is in the schema, the schema renames by position —
   `create_dataframe(session, [n: [1]], schema: "id INT")` — and a schema naming some columns but
   not others is refused, naming them. The server parses the schema as the frame is built (one
-  round trip), so a malformed one fails here rather than at the first action.
+  round trip), so a malformed one fails here rather than at the first action. Every column
+  ships nullable — Explorer writes Arrow fields so — and the server refuses to cast one to a
+  `NOT NULL` field (`NULLABLE_COLUMN_OR_FIELD`); leave nullability to the default.
 
   ## Examples
 

lib/latu/client.ex

@@ -12,11 +12,10 @@ defmodule Latu.Client do
   alias Latu.Progress
   alias Latu.Protocol.Spark.Connect, as: Proto
   alias Latu.Protocol.Spark.Connect.SparkConnectService.Stub
+  alias Latu.Retry
   alias Latu.Session
   alias Latu.Telemetry
 
-  @loopback ~w(localhost 127.0.0.1 ::1 0:0:0:0:0:0:0:1)
-
   # The one field each AnalyzePlan response arm carries. Read out of PySpark's
   # `AnalyzeResult.fromProto`; `persist` and `unpersist` carry none.
   @analyze_fields %{
@@ -94,7 +93,7 @@ defmodule Latu.Client do
   end
 
   defp released(%Session{} = session) do
-    case release_session(session, timeout: @release_on_disconnect_timeout) do
+    case release_session(session, [timeout: @release_on_disconnect_timeout], &rpc/3) do
       {:ok, session} ->
         session
 
@@ -163,7 +162,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.analyze_plan(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("AnalyzePlan", session, call),
+    with {:ok, response} <- retrying("AnalyzePlan", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       {:ok, response, session}
@@ -305,7 +304,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.config(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("Config", session, call),
+    with {:ok, response} <- retrying("Config", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       warn(op, response.warnings)
@@ -400,7 +399,7 @@ defmodule Latu.Client do
     request = struct!(request, interrupt_scope(opts))
     call = fn -> Stub.interrupt(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("Interrupt", session, call),
+    with {:ok, response} <- retrying("Interrupt", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       {:ok, response.interrupted_ids, session}
@@ -444,7 +443,8 @@ defmodule Latu.Client do
 
     call = fn -> Stub.get_status(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("GetStatus", session, call) |> explain_missing_session(session),
+    with {:ok, response} <-
+           retrying("GetStatus", session, call) |> explain_missing_session(session),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       {:ok, Enum.map(response.operation_statuses, &operation_status/1), session}
@@ -512,7 +512,7 @@ defmodule Latu.Client do
     call = fn -> Stub.clone_session(session.channel, request, timeout: session.timeout) end
 
     with {:ok, response} <-
-           rpc("CloneSession", session, call) |> explain_missing_session(session),
+           retrying("CloneSession", session, call) |> explain_missing_session(session),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id),
          {:ok, clone} <- cloned(session, response, wanted) do
@@ -553,6 +553,12 @@ defmodule Latu.Client do
   end
 
   def release_session(%Session{} = session, opts) when is_list(opts) do
+    release_session(session, opts, &retrying/3)
+  end
+
+  # `through` is `retrying/3` for the call a caller makes, `rpc/3` for the courtesy on the way
+  # out of `disconnect/2`, which has a budget of its own to keep.
+  defp release_session(%Session{} = session, opts, through) do
     opts = Keyword.validate!(opts, allow_reconnect: false, timeout: session.timeout)
 
     request = %Proto.ReleaseSessionRequest{
@@ -564,7 +570,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.release_session(session.channel, request, timeout: opts[:timeout]) end
 
-    with {:ok, response} <- rpc("ReleaseSession", session, call) do
+    with {:ok, response} <- through.("ReleaseSession", session, call) do
       Session.confirm(session, response.session_id, response.server_side_session_id)
     end
   end
@@ -614,7 +620,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.artifact_status(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("ArtifactStatus", session, call),
+    with {:ok, response} <- retrying("ArtifactStatus", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       cached =
@@ -642,7 +648,7 @@ defmodule Latu.Client do
       |> GRPC.Stub.recv(timeout: session.timeout)
     end
 
-    with {:ok, response} <- rpc("AddArtifacts", session, call),
+    with {:ok, response} <- retrying("AddArtifacts", session, call),
          {:ok, session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       case Enum.reject(response.artifacts, & &1.is_crc_successful) do
@@ -1040,6 +1046,35 @@ defmodule Latu.Client do
     %Proto.UserContext{user_id: session.user_id, user_name: session.user_name}
   end
 
+  @doc false
+  # Every RPC that asks the server for something is retried on the session's policy, as
+  # PySpark's `Retrying` wraps every call. Not the two that open a result stream —
+  # `Latu.Client.Execution` decides those — and not a best-effort release, which must never
+  # hold a caller up. Backing off blocks the caller's process, the one place Latu has to do it.
+  # Public for the offline test; `call` is any zero-arity function returning what a `Stub` call
+  # does.
+  def retrying(name, %Session{retry: %Retry{} = retry} = session, call) do
+    retrying(name, session, call, retry, 0)
+  end
+
+  defp retrying(name, session, call, retry, attempt) do
+    case rpc(name, session, call) do
+      {:error, %Error{} = error} = failure ->
+        if attempt < retry.max_retries and Retry.retryable?(error) do
+          backoff = Retry.wait(retry, attempt, error.retry_delay)
+          ids = %{session_id: session.session_id, rpc: name}
+          Telemetry.attempt(:retry, backoff, attempt + 1, ids)
+          Process.sleep(backoff)
+          retrying(name, session, call, retry, attempt + 1)
+        else
+          failure
+        end
+
+      result ->
+        result
+    end
+  end
+
   # Every unary RPC funnels through here, which is why the telemetry span is here and not at
   # ten call sites. Only the session *id* is handed over — see `Latu.Telemetry`.
   defp rpc(name, %Session{} = session, call) do
@@ -1062,16 +1097,24 @@ defmodule Latu.Client do
   # `deps/googleapis`, which `grpc_core` depends on — so this needs no dependency, no proto
   # generation and no round trip.
   @error_info_url "type.googleapis.com/google.rpc.ErrorInfo"
+  @retry_info_url "type.googleapis.com/google.rpc.RetryInfo"
 
   defp rpc_error(%GRPC.RPCError{} = error) do
     Error.new(:rpc, error.message, [status: error.status, details: error.details] ++ info(error))
   end
 
   defp info(%GRPC.RPCError{details: details}) when is_list(details) do
-    case Enum.find(details, &(&1.type_url == @error_info_url)) do
-      nil -> []
-      any -> from_error_info(Google.Rpc.ErrorInfo.decode(any.value))
-    end
+    Enum.flat_map(details, fn
+      %{type_url: @error_info_url, value: value} ->
+        from_error_info(Google.Rpc.ErrorInfo.decode(value))
+
+      # Spark itself never attaches one; a gateway in front of it may, to say "later".
+      %{type_url: @retry_info_url, value: value} ->
+        [retry_delay: delay_ms(Google.Rpc.RetryInfo.decode(value))]
+
+      _other ->
+        []
+    end)
   rescue
     # Inspecting an error must never be the thing that fails. A trailer Latu cannot read leaves
     # the fields nil, which is exactly what a server that sent none does.
@@ -1080,6 +1123,13 @@ defmodule Latu.Client do
 
   defp info(%GRPC.RPCError{}), do: []
 
+  # An absent duration is still a RetryInfo — "retry, no particular wait" — so 0, not nil.
+  defp delay_ms(%Google.Rpc.RetryInfo{retry_delay: nil}), do: 0
+
+  defp delay_ms(%Google.Rpc.RetryInfo{retry_delay: %{seconds: seconds, nanos: nanos}}) do
+    seconds * 1_000 + div(nanos, 1_000_000)
+  end
+
   # The metadata keys are PySpark's — `connect.py`'s `convert_exception` reads the same six.
   defp from_error_info(%Google.Rpc.ErrorInfo{metadata: metadata}) do
     [
@@ -1133,7 +1183,7 @@ defmodule Latu.Client do
 
     call = fn -> Stub.fetch_error_details(session.channel, request, timeout: session.timeout) end
 
-    with {:ok, response} <- rpc("FetchErrorDetails", session, call),
+    with {:ok, response} <- retrying("FetchErrorDetails", session, call),
          {:ok, _session} <-
            Session.confirm(session, response.session_id, response.server_side_session_id) do
       {:ok, filled(error, Enum.map(causes(response), &cause/1))}
@@ -1207,8 +1257,14 @@ defmodule Latu.Client do
   # "host:port" puts elixir-grpc in compatibility mode, which rewrites to `ipv4:host:port`
   # and resolves through Gun rather than the DNS resolver. That is what we want: a Spark
   # Connect session is server-side state pinned to one server, so re-resolving or balancing
-  # across A records would silently move us to a server that has never heard of it.
-  defp target(%Session{host: host, port: port}), do: "#{host}:#{port}"
+  # across A records would silently move us to a server that has never heard of it. An IPv6
+  # literal has colons of its own, so it takes the one form elixir-grpc reads brackets in.
+  defp target(%Session{host: host, port: port}) do
+    case :inet.parse_ipv6strict_address(String.to_charlist(host)) do
+      {:ok, _address} -> "ipv6:[#{host}]:#{port}"
+      {:error, _not_ipv6} -> "#{host}:#{port}"
+    end
+  end
 
   defp open(target, opts) do
     case GRPC.Stub.connect(target, opts) do
@@ -1287,13 +1343,28 @@ defmodule Latu.Client do
   defp check_token_transport(%Session{token: nil}), do: :ok
   defp check_token_transport(%Session{use_ssl: true}), do: :ok
 
-  defp check_token_transport(%Session{host: host}) when host in @loopback, do: :ok
-
   defp check_token_transport(%Session{host: host}) do
-    {:error,
-     Error.new(
-       :connect,
-       "refusing to send a bearer token in cleartext to #{host}; add ;use_ssl=true to the URL"
-     )}
+    if loopback?(host) do
+      :ok
+    else
+      {:error,
+       Error.new(
+         :connect,
+         "refusing to send a bearer token in cleartext to #{host}; add ;use_ssl=true to the URL"
+       )}
+    end
+  end
+
+  # What never leaves the machine: `localhost`, all of 127/8, `::1` in any spelling, and 127/8
+  # written as an IPv4-mapped IPv6 address. PySpark checks the string "localhost" alone.
+  defp loopback?("localhost"), do: true
+
+  defp loopback?(host) do
+    case :inet.parse_address(String.to_charlist(host)) do
+      {:ok, {127, _, _, _}} -> true
+      {:ok, {0, 0, 0, 0, 0, 0, 0, 1}} -> true
+      {:ok, {0, 0, 0, 0, 0, 0xFFFF, high, _low}} -> div(high, 256) == 127
+      _ -> false
+    end
   end
 end

lib/latu/client/execution.ex

@@ -18,11 +18,6 @@ defmodule Latu.Client.Execution do
   # The retry policy proper is a separate concern. See dev/README.md.
   @max_empty_reattaches 100
 
-  # gRPC status codes, spelled out because this module must not reference GRPC — see the
-  # layering test.
-  @unavailable 14
-  @internal 13
-
   # Spark reports a lost execution or session in the message, with no status code of its own to
   # match on. PySpark detects it the same way.
   @lost_handle ["INVALID_HANDLE.OPERATION_NOT_FOUND", "INVALID_HANDLE.SESSION_NOT_FOUND"]
@@ -143,15 +138,17 @@ defmodule Latu.Client.Execution do
       # The server has no record of this execution. If nothing had arrived yet, the original
       # ExecutePlan never landed and re-sending it under the same operation_id is safe.
       lost_handle?(error) and is_nil(execution.last_response_id) and budget ->
-        {{:restart, Retry.wait(retry, retries)}, %{execution | retries: retries + 1}}
+        wait = Retry.wait(retry, retries, error.retry_delay)
+        {{:restart, wait}, %{execution | retries: retries + 1}}
 
       # If responses had already arrived, they are gone and re-sending would duplicate the ones
       # that did. PySpark raises RESPONSE_ALREADY_RECEIVED here for the same reason.
       lost_handle?(error) ->
         {{:fail, unrecoverable(execution, error)}, execution}
 
-      budget and retryable?(error) ->
-        {{:reattach, Retry.wait(retry, retries)}, %{execution | retries: retries + 1}}
+      budget and Retry.retryable?(error) ->
+        wait = Retry.wait(retry, retries, error.retry_delay)
+        {{:reattach, wait}, %{execution | retries: retries + 1}}
 
       true ->
         {{:fail, give_up(execution, error)}, execution}
@@ -162,23 +159,14 @@ defmodule Latu.Client.Execution do
   # Retries
   # =============================================
 
-  # Which errors are retryable is PySpark's DefaultPolicy; how often and how long is the
-  # session's `Latu.Retry`. PySpark's third case — any error carrying RetryInfo metadata — is
-  # not covered: elixir-grpc does not decode that detail.
+  # Which errors are retryable, how often and how long is the session's `Latu.Retry`; the one
+  # case of its own here is a lost handle, which only an execution can have.
   defp policy(%__MODULE__{session: %Session{retry: %Retry{} = retry}}), do: retry
 
   defp lost_handle?(%Error{message: message}) do
     Enum.any?(@lost_handle, &String.contains?(message, &1))
   end
 
-  defp retryable?(%Error{status: @unavailable}), do: true
-
-  defp retryable?(%Error{status: @internal, message: message}) do
-    message =~ "INVALID_CURSOR.DISCONNECTED"
-  end
-
-  defp retryable?(%Error{}), do: false
-
   # Nothing had arrived, so the only way here is a spent budget.
   defp unrecoverable(%__MODULE__{last_response_id: nil} = execution, %Error{} = error) do
     give_up(execution, error)

lib/latu/data_frame.ex

@@ -353,7 +353,10 @@ defmodule Latu.DataFrame do
   end
 
   defp column_name(name) when is_binary(name), do: name
-  defp column_name(name) when is_atom(name) and not is_nil(name), do: Atom.to_string(name)
+
+  defp column_name(name) when is_atom(name) and not is_nil(name) and not is_boolean(name) do
+    Atom.to_string(name)
+  end
 
   defp column_name(name) do
     raise ArgumentError, "a column name is a string or an atom, not #{inspect(name)}"