fix(elixir-client): Improve elixir client parity

.changeset/mean-hairs-eat.md

@@ -0,0 +1,6 @@
+---
+'@core/elixir-client': patch
+---
+
+Switch retry backoff to full jitter strategy, matching the TypeScript client
+and providing better spread to avoid thundering herd.

.changeset/silly-singers-cheer.md

@@ -0,0 +1,8 @@
+---
+'@core/elixir-client': patch
+---
+
+Add response header validation matching the TypeScript client. Missing Electric
+headers now raise a clear error pointing to proxy/CDN misconfiguration instead
+of silently producing incorrect behaviour.
+

.github/workflows/elixir_client_tests.yml

@@ -22,7 +22,6 @@ jobs:
     env:
       MIX_ENV: test
       ELECTRIC_PORT: 3100
-      ELECTRIC_URL: 'http://127.0.0.1:3333'
       DATABASE_URL: 'postgresql://postgres:password@127.0.0.1:54323/postgres?sslmode=disable'
     services:
       postgres:

packages/elixir-client/config/runtime.exs

@@ -3,7 +3,7 @@ import Config
 config :logger, level: :warning
 
 if config_env() == :test do
-  port = 3333
+  port = System.get_env("ELECTRIC_PORT", "3333") |> String.to_integer()
   default_database_url = "postgresql://postgres:password@localhost:54321/electric?sslmode=disable"
   database_url = System.get_env("DATABASE_URL", default_database_url)
 

packages/elixir-client/lib/electric/client/embedded.ex

@@ -49,7 +49,9 @@ if Code.ensure_loaded?(Electric.Shapes.Api) do
         last_offset: convert_offset(response.offset),
         shape_handle: response.handle,
         schema: Api.schema(response),
-        next_cursor: nil,
+        # Embedded mode has no CDN, so cursor isn't needed for cache busting,
+        # but we set a dummy value to satisfy header validation.
+        next_cursor: 0,
         request_timestamp: timestamp,
         # Ensure we enumerate the response body in the same process that made the
         # original api request so the cleanup shape process cleanup works

packages/elixir-client/lib/electric/client/fetch/http.ex

@@ -239,13 +239,11 @@ defmodule Electric.Client.Fetch.HTTP do
   end
 
   defp retry_delay(n) do
-    (Integer.pow(2, n) * 1000 * jitter())
-    |> min(30_000 * jitter())
-    |> trunc()
-  end
-
-  defp jitter() do
-    1 - 0.1 * :rand.uniform()
+    # Full jitter strategy (AWS recommended), minimum 1ms:
+    #   random_between(1, min(cap, base * 2^n))
+    # See: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
+    delay = min(32_000, Integer.pow(2, n) * 1000)
+    :rand.uniform(delay)
   end
 
   defp now, do: System.monotonic_time(:second)

packages/elixir-client/lib/electric/client/mock.ex

@@ -204,7 +204,8 @@ defmodule Electric.Client.Mock do
       body: jsonify(opts[:body] || []),
       schema: Keyword.get(opts, :schema, nil),
       shape_handle: Keyword.get(opts, :shape_handle, nil),
-      last_offset: Keyword.get(opts, :last_offset, nil)
+      last_offset: Keyword.get(opts, :last_offset, nil),
+      next_cursor: Keyword.get(opts, :next_cursor, nil)
     }
   end
 

packages/elixir-client/lib/electric/client/poll.ex

@@ -84,6 +84,7 @@ defmodule Electric.Client.Poll do
 
     case Fetch.request(client, request) do
       %Fetch.Response{status: status} = resp when status in 200..299 ->
+        validate_headers!(resp, state)
         handle_success(resp, client, state, shape_key)
 
       {:error, %Fetch.Response{status: 409} = resp} ->
@@ -135,7 +136,7 @@ defmodule Electric.Client.Poll do
   defp maybe_add_cache_buster(params, buster), do: Map.put(params, "cache-buster", buster)
 
   defp handle_success(resp, client, state, shape_key) do
-    response_handle = shape_handle!(resp)
+    response_handle = resp.shape_handle
     expired_handle = ExpiredShapesCache.get_expired_handle(shape_key)
 
     # Check for stale CDN response — always enter stale-retry to add a cache
@@ -272,9 +273,46 @@ defmodule Electric.Client.Poll do
     end
   end
 
-  defp shape_handle!(resp) do
-    shape_handle(resp) ||
-      raise Client.Error, message: "Missing electric-handle header", resp: resp
+  defp validate_headers!(%Fetch.Response{} = resp, %ShapeState{} = state) do
+    # Validate required Electric response fields, matching the TypeScript
+    # client's createFetchWithResponseHeadersCheck middleware.
+    #
+    # We check parsed struct fields (not raw HTTP headers) so this works
+    # for both real HTTP responses and the Mock fetch implementation.
+    #
+    # Rules (based on TypeScript's createFetchWithResponseHeadersCheck):
+    # - All responses: electric-handle, electric-offset
+    # - Non-live responses: electric-schema (server always sends it, but we
+    #   only validate when we don't have one yet since it's first-write-wins)
+    # - Live responses: electric-cursor (CDN cache buster)
+    is_live? = state.up_to_date?
+
+    missing = []
+
+    missing =
+      if is_nil(resp.shape_handle), do: ["electric-handle" | missing], else: missing
+
+    missing =
+      if is_nil(resp.last_offset), do: ["electric-offset" | missing], else: missing
+
+    missing =
+      if not is_live? and is_nil(state.schema) and is_nil(resp.schema),
+        do: ["electric-schema" | missing],
+        else: missing
+
+    missing =
+      if is_live? and is_nil(resp.next_cursor),
+        do: ["electric-cursor" | missing],
+        else: missing
+
+    if missing != [] do
+      raise Client.Error,
+        message:
+          "Response is missing required Electric header(s): #{Enum.join(missing, ", ")}. " <>
+            "This usually indicates a proxy or CDN misconfiguration — " <>
+            "check that your proxy forwards all Electric headers " <>
+            "(electric-handle, electric-offset, electric-schema, electric-cursor) to the client."
+    end
   end
 
   defp shape_handle(%Fetch.Response{shape_handle: shape_handle}) do

packages/elixir-client/lib/electric/client/shape_state.ex

@@ -42,7 +42,7 @@ defmodule Electric.Client.ShapeState do
           offset: Offset.t(),
           schema: Client.schema() | nil,
           value_mapper_fun: Client.ValueMapper.mapper_fun() | nil,
-          next_cursor: binary() | nil,
+          next_cursor: Client.cursor() | nil,
           up_to_date?: boolean(),
           tag_to_keys: %{optional(term()) => MapSet.t()},
           key_data: %{optional(term()) => %{tags: MapSet.t(), msg: term()}},

packages/elixir-client/test/electric/client/mock_test.exs

@@ -44,6 +44,7 @@ defmodule Electric.Client.MockTest do
         schema: %{id: %{type: "int8"}},
         last_offset: Offset.new(0, 1),
         shape_handle: "my-shape",
+        next_cursor: 1,
         body: [
           Client.Mock.change(value: %{id: "4444"}),
           Client.Mock.up_to_date(lsn: 1234)

packages/elixir-client/test/electric/client_test.exs

@@ -608,6 +608,9 @@ defmodule Electric.ClientTest do
 
   defp bypass_resp(conn, body, opts) do
     status = Keyword.get(opts, :status, 200)
+    # Default cursor to a value matching Electric server behaviour — every
+    # successful response includes a cursor for CDN cache busting.
+    cursor = Keyword.get(opts, :cursor, if(opts[:last_offset], do: "1"))
 
     # the quick-responding tests are less flaky with a small delay --
     # think this is an issue with bypass. for very fast responses
@@ -620,7 +623,7 @@ defmodule Electric.ClientTest do
     |> put_optional_header("electric-handle", opts[:shape_handle])
     |> put_optional_header("electric-offset", opts[:last_offset])
     |> put_optional_header("electric-schema", opts[:schema])
-    |> put_optional_header("electric-cursor", opts[:cursor])
+    |> put_optional_header("electric-cursor", cursor)
     |> Plug.Conn.resp(status, body)
   end
 
@@ -1013,6 +1016,171 @@ defmodule Electric.ClientTest do
         stream(ctx) |> Stream.run()
       end)
     end
+
+    test "raises a clear error when required Electric headers are missing from response", ctx do
+      # When a proxy or CDN strips Electric headers from responses, the client
+      # should raise a clear error listing ALL missing headers and suggesting
+      # proxy/CORS misconfiguration as the likely cause.
+      #
+      # Currently:
+      # - electric-handle: raises with unhelpful "Missing electric-handle header"
+      # - electric-offset: silently falls back to previous offset (no error)
+      # - electric-schema: silently skipped, values returned as raw strings
+
+      body =
+        Jason.encode!([
+          %{
+            "headers" => %{"operation" => "insert"},
+            "offset" => "1_0",
+            "value" => %{"id" => "1"}
+          },
+          %{"headers" => %{"control" => "up-to-date", "global_last_seen_lsn" => 9999}}
+        ])
+
+      # Return a 200 response with valid body but NO Electric headers at all —
+      # simulating a proxy that strips custom headers.
+      Bypass.expect_once(ctx.bypass, fn conn ->
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.resp(200, body)
+      end)
+
+      error =
+        assert_raise Client.Error, fn ->
+          stream(ctx) |> Enum.take(1)
+        end
+
+      # The error message should mention proxy/CDN misconfiguration and list
+      # the missing headers, not just say "Missing electric-handle header".
+      assert error.message =~ ~r/proxy|CDN/i,
+             "Expected error message to mention proxy/CDN misconfiguration for easier " <>
+               "debugging, but got: #{inspect(error.message)}"
+
+      assert error.message =~ "electric-handle",
+             "Error should list electric-handle as missing"
+
+      assert error.message =~ "electric-offset",
+             "Error should list electric-offset as missing"
+
+      assert error.message =~ "electric-schema",
+             "Error should list electric-schema as missing"
+    end
+
+    test "missing electric-offset header is detected, not silently ignored", ctx do
+      # When electric-offset is missing, the client silently falls back to
+      # the previous offset. This can cause the client to re-request the
+      # same data or get stuck without any indication of the problem.
+
+      body =
+        Jason.encode!([
+          %{
+            "headers" => %{"operation" => "insert"},
+            "offset" => "1_0",
+            "value" => %{"id" => "1"}
+          },
+          %{"headers" => %{"control" => "up-to-date", "global_last_seen_lsn" => 9999}}
+        ])
+
+      # Response has handle and schema but no offset header
+      Bypass.expect_once(ctx.bypass, fn conn ->
+        schema = Jason.encode!(%{"id" => %{type: "text"}})
+
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.put_resp_header("electric-handle", "my-shape")
+        |> Plug.Conn.put_resp_header("electric-schema", schema)
+        |> Plug.Conn.resp(200, body)
+      end)
+
+      assert_raise Client.Error, ~r/electric-offset/i, fn ->
+        stream(ctx) |> Enum.take(1)
+      end
+    end
+
+    test "missing electric-schema header on non-live request is detected, not silently ignored",
+         ctx do
+      # The TypeScript client requires electric-schema on non-live responses.
+      # When it's missing, the value mapper is never built and values are
+      # returned as raw strings with no type conversion — silently producing
+      # wrong types.
+
+      body =
+        Jason.encode!([
+          %{
+            "headers" => %{"operation" => "insert"},
+            "offset" => "1_0",
+            "value" => %{"id" => "1"}
+          },
+          %{"headers" => %{"control" => "up-to-date", "global_last_seen_lsn" => 9999}}
+        ])
+
+      # Response has handle and offset but no schema header
+      Bypass.expect_once(ctx.bypass, fn conn ->
+        conn
+        |> Plug.Conn.put_resp_content_type("application/json")
+        |> Plug.Conn.put_resp_header("electric-handle", "my-shape")
+        |> Plug.Conn.put_resp_header("electric-offset", "1_0")
+        |> Plug.Conn.resp(200, body)
+      end)
+
+      assert_raise Client.Error, ~r/electric-schema/i, fn ->
+        stream(ctx) |> Enum.take(1)
+      end
+    end
+
+    test "missing electric-cursor header on live response is detected, not silently ignored",
+         ctx do
+      # electric-cursor is the cache buster for the next live request. Without
+      # it, CDNs can serve stale data on live polls.
+
+      schema = Jason.encode!(%{"id" => %{type: "text"}})
+
+      body1 =
+        Jason.encode!([
+          %{
+            "headers" => %{"operation" => "insert"},
+            "offset" => "1_0",
+            "value" => %{"id" => "1"}
+          },
+          %{"headers" => %{"control" => "up-to-date", "global_last_seen_lsn" => 9998}}
+        ])
+
+      body_live =
+        Jason.encode!([
+          %{
+            "headers" => %{"operation" => "insert"},
+            "offset" => "2_0",
+            "value" => %{"id" => "2"}
+          },
+          %{"headers" => %{"control" => "up-to-date", "global_last_seen_lsn" => 9999}}
+        ])
+
+      {:ok, request_count} = Agent.start_link(fn -> 0 end)
+
+      Bypass.stub(ctx.bypass, "GET", "/v1/shape", fn conn ->
+        count = Agent.get_and_update(request_count, fn c -> {c + 1, c + 1} end)
+
+        if count == 1 do
+          # Initial non-live response with all required headers
+          bypass_resp(conn, body1,
+            shape_handle: "my-shape",
+            last_offset: "1_0",
+            schema: schema
+          )
+        else
+          # Live response with handle and offset but missing cursor
+          conn
+          |> Plug.Conn.put_resp_content_type("application/json")
+          |> Plug.Conn.put_resp_header("electric-handle", "my-shape")
+          |> Plug.Conn.put_resp_header("electric-offset", "2_0")
+          |> Plug.Conn.resp(200, body_live)
+        end
+      end)
+
+      assert_raise Client.Error, ~r/electric-cursor/i, fn ->
+        stream(ctx) |> Enum.take(4)
+      end
+    end
   end
 
   defp bypass_response(ctx, responses, opts \\ []) do