From e2cd0a326e094ebe47afc5688c9b60d201080a2f Mon Sep 17 00:00:00 2001 From: Garry Hill Date: Thu, 19 Mar 2026 12:17:04 +0000 Subject: [PATCH 1/3] Use full jitter strategy for retry backoff Switch from multiplicative jitter (90-100% of delay) to full jitter (random between 1 and delay), matching the AWS-recommended strategy used by the TypeScript client. Also align max delay from 30s to 32s. --- .changeset/mean-hairs-eat.md | 6 ++++++ .../elixir-client/lib/electric/client/fetch/http.ex | 12 +++++------- 2 files changed, 11 insertions(+), 7 deletions(-) create mode 100644 .changeset/mean-hairs-eat.md diff --git a/.changeset/mean-hairs-eat.md b/.changeset/mean-hairs-eat.md new file mode 100644 index 0000000000..5cb0458a73 --- /dev/null +++ b/.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. diff --git a/packages/elixir-client/lib/electric/client/fetch/http.ex b/packages/elixir-client/lib/electric/client/fetch/http.ex index cc0d6c0d41..1cd8cfe4aa 100644 --- a/packages/elixir-client/lib/electric/client/fetch/http.ex +++ b/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): + # random_between(0, 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) From cd199f630352bcba05273432549037ec7a23d2ac Mon Sep 17 00:00:00 2001 From: Garry Hill Date: Thu, 19 Mar 2026 12:24:13 +0000 Subject: [PATCH 2/3] Add response header validation Validate that required Electric headers are present on responses, raising a clear error mentioning proxy/CDN misconfiguration when they're missing. Previously, missing headers either caused unhelpful errors (electric-handle) or were silently ignored (electric-offset, electric-schema, electric-cursor). --- .changeset/silly-singers-cheer.md | 8 + packages/elixir-client/config/runtime.exs | 2 +- .../lib/electric/client/embedded.ex | 4 +- .../elixir-client/lib/electric/client/mock.ex | 3 +- .../elixir-client/lib/electric/client/poll.ex | 43 +++++ .../test/electric/client/mock_test.exs | 1 + .../test/electric/client_test.exs | 170 +++++++++++++++++- 7 files changed, 227 insertions(+), 4 deletions(-) create mode 100644 .changeset/silly-singers-cheer.md diff --git a/.changeset/silly-singers-cheer.md b/.changeset/silly-singers-cheer.md new file mode 100644 index 0000000000..778d51f612 --- /dev/null +++ b/.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. + diff --git a/packages/elixir-client/config/runtime.exs b/packages/elixir-client/config/runtime.exs index 1f05485aaf..4a2b7867b0 100644 --- a/packages/elixir-client/config/runtime.exs +++ b/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) diff --git a/packages/elixir-client/lib/electric/client/embedded.ex b/packages/elixir-client/lib/electric/client/embedded.ex index aaf5dfd801..9f19362200 100644 --- a/packages/elixir-client/lib/electric/client/embedded.ex +++ b/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 diff --git a/packages/elixir-client/lib/electric/client/mock.ex b/packages/elixir-client/lib/electric/client/mock.ex index 0c9246c62d..453fbdc183 100644 --- a/packages/elixir-client/lib/electric/client/mock.ex +++ b/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 diff --git a/packages/elixir-client/lib/electric/client/poll.ex b/packages/elixir-client/lib/electric/client/poll.ex index 86360d69e8..471010e933 100644 --- a/packages/elixir-client/lib/electric/client/poll.ex +++ b/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} -> @@ -272,6 +273,48 @@ defmodule Electric.Client.Poll do end end + 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!(resp) do shape_handle(resp) || raise Client.Error, message: "Missing electric-handle header", resp: resp diff --git a/packages/elixir-client/test/electric/client/mock_test.exs b/packages/elixir-client/test/electric/client/mock_test.exs index 80d322b4d2..ff41037813 100644 --- a/packages/elixir-client/test/electric/client/mock_test.exs +++ b/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) diff --git a/packages/elixir-client/test/electric/client_test.exs b/packages/elixir-client/test/electric/client_test.exs index 77b5410fce..8ed5831b5a 100644 --- a/packages/elixir-client/test/electric/client_test.exs +++ b/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 From 15c8f045d9a6bb8583b6237b2a6305404b65429c Mon Sep 17 00:00:00 2001 From: Garry Hill Date: Thu, 19 Mar 2026 14:32:11 +0000 Subject: [PATCH 3/3] review feedback --- .github/workflows/elixir_client_tests.yml | 1 - packages/elixir-client/lib/electric/client/fetch/http.ex | 4 ++-- packages/elixir-client/lib/electric/client/poll.ex | 7 +------ packages/elixir-client/lib/electric/client/shape_state.ex | 2 +- 4 files changed, 4 insertions(+), 10 deletions(-) diff --git a/.github/workflows/elixir_client_tests.yml b/.github/workflows/elixir_client_tests.yml index 7ca9e0e061..2dc2afcb0c 100644 --- a/.github/workflows/elixir_client_tests.yml +++ b/.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: diff --git a/packages/elixir-client/lib/electric/client/fetch/http.ex b/packages/elixir-client/lib/electric/client/fetch/http.ex index 1cd8cfe4aa..bb04865180 100644 --- a/packages/elixir-client/lib/electric/client/fetch/http.ex +++ b/packages/elixir-client/lib/electric/client/fetch/http.ex @@ -239,8 +239,8 @@ defmodule Electric.Client.Fetch.HTTP do end defp retry_delay(n) do - # Full jitter strategy (AWS recommended): - # random_between(0, min(cap, base * 2^n)) + # 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) diff --git a/packages/elixir-client/lib/electric/client/poll.ex b/packages/elixir-client/lib/electric/client/poll.ex index 471010e933..20a6752371 100644 --- a/packages/elixir-client/lib/electric/client/poll.ex +++ b/packages/elixir-client/lib/electric/client/poll.ex @@ -136,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 @@ -315,11 +315,6 @@ 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 - end - defp shape_handle(%Fetch.Response{shape_handle: shape_handle}) do shape_handle end diff --git a/packages/elixir-client/lib/electric/client/shape_state.ex b/packages/elixir-client/lib/electric/client/shape_state.ex index 6f1888bac2..e66d6f4c51 100644 --- a/packages/elixir-client/lib/electric/client/shape_state.ex +++ b/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()}},