fix: Return 409 must-refetch instead of 400 on shape definition and handle mismatch (#2476)

Fixes #2474

Currently, Electric returns a 400 error code when the shape definition
and shape handle do not match.
However, we discovered a pattern for handling dependent shapes on the
proxy where shape B is constructed based on the results for shape A. For
example, if we want to load all teams for a user we would first create a
shape to get all team_ids a user belongs to and then another shape that
selects all teams where `team_id IN (team_id_1, ..., team_id_n)`. The
definition of the second shape changes whenever the results of the first
shape change. For this to work properly, we need Electric to return 409
must-refetch with the handle of the new shape when the shape definition
changes.

Author: kevin-dp

.changeset/shy-eagles-search.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": minor
+---
+
+Electric now also returns 409 instead of 400 when the shape handle and the shape definition do not match.

packages/sync-service/lib/electric/shapes/api.ex

@@ -259,23 +259,14 @@ defmodule Electric.Shapes.Api do
   end
 
   defp handle_shape_info(nil, %Request{} = request) do
-    %{params: %{shape_definition: shape, handle: shape_handle}, api: api} = request
-    # There is no shape that matches the shape definition (because shape info is `nil`)
-    if shape_handle != nil && Shapes.has_shape?(api, shape_handle) do
-      # but there is a shape that matches the shape handle
-      # thus the shape handle does not match the shape definition
-      # and we return a 400 bad request status code
-      {:error, Response.shape_definition_mismatch(request)}
-    else
-      # The shape handle does not exist or no longer exists
-      # e.g. it may have been deleted.
-      # Hence, create a new shape for this shape definition
-      # and return a 409 with a redirect to the newly created shape.
-      # (will be done by the recursive `handle_shape_info` call)
-      api
-      |> Shapes.get_or_create_shape_handle(shape)
-      |> handle_shape_info(request)
-    end
+    %{params: %{shape_definition: shape}, api: api} = request
+    # There is no shape that matches the shape definition (because shape info is `nil`).
+    # Hence, create a new shape for this shape definition
+    # and return a 409 with a redirect to the newly created shape.
+    # (will be done by the recursive `handle_shape_info` call)
+    api
+    |> Shapes.get_or_create_shape_handle(shape)
+    |> handle_shape_info(request)
   end
 
   defp handle_shape_info(
@@ -303,26 +294,28 @@ defmodule Electric.Shapes.Api do
 
   defp handle_shape_info(
          {active_shape_handle, _},
-         %Request{params: %{handle: shape_handle}} = request
+         %Request{} = request
        ) do
-    if Shapes.has_shape?(request.api, shape_handle) do
-      # The shape with the provided ID exists but does not match the shape definition
-      # otherwise we would have found it and it would have matched the previous function clause
-      {:error, Response.shape_definition_mismatch(request)}
-    else
-      # The requested shape_handle is not found, returns 409 along with a location redirect for clients to
-      # re-request the shape from scratch with the new shape id which acts as a consistent cache buster
-      # e.g. GET /v1/shape?table={root_table}&handle={new_shape_handle}&offset=-1
-
-      # TODO: discuss returning a 307 redirect rather than a 409, the client
-      # will have to detect this and throw out old data
-
-      {:error,
-       Response.error(request, @must_refetch,
-         handle: active_shape_handle,
-         status: 409
-       )}
-    end
+    # Either the requested shape handle exists or does not exist.
+    # If it exists there is a mismatch between the shape definition and the shape handle
+    # (otherwise we would have matched the previous function clause).
+    # The mismatch may occur because the shape definition has changed,
+    # which happens frequently when working with dependent shapes
+    # where a shape's WHERE clause is constructed based on the values of another shape
+    # (e.g. to load all children pointed at by a FK in a parent table).
+
+    # If the shape handle does not exist, it may have never existed or it may have been deleted.
+    # In either case we return a 409 with a location redirect for clients to
+    # re-request the shape from scratch with the new shape id which acts as a consistent cache buster
+    # e.g. GET /v1/shape?table={root_table}&handle={new_shape_handle}&offset=-1
+
+    # TODO: discuss returning a 307 redirect rather than a 409, the client
+    # will have to detect this and throw out old data
+    {:error,
+     Response.error(request, @must_refetch,
+       handle: active_shape_handle,
+       status: 409
+     )}
   end
 
   defp hold_until_stack_ready(%Api{} = api) do

packages/sync-service/test/electric/plug/router_test.exs

@@ -1199,7 +1199,7 @@ defmodule Electric.Plug.RouterTest do
              ] = Jason.decode!(conn.resp_body)
     end
 
-    test "GET receives 400 when shape handle does not match shape definition", %{
+    test "GET receives 409 when shape handle does not match shape definition", %{
       opts: opts
     } do
       where = "value ILIKE 'yes%'"
@@ -1221,13 +1221,17 @@ defmodule Electric.Plug.RouterTest do
         conn("GET", "/v1/shape?table=items", %{offset: next_offset, handle: shape_handle})
         |> Router.call(opts)
 
-      assert %{status: 400} = conn
+      assert %{status: 409} = conn
 
-      assert conn.resp_body ==
-               Jason.encode!(%{
-                 message:
-                   "The specified shape definition and handle do not match. Please ensure the shape definition is correct or omit the shape handle from the request to obtain a new one."
-               })
+      assert Jason.decode!(conn.resp_body) == [
+               %{"headers" => %{"control" => "must-refetch"}}
+             ]
+
+      new_shape_handle = get_resp_header(conn, "electric-handle")
+      assert new_shape_handle != shape_handle
+
+      assert get_resp_header(conn, "location") ==
+               "/v1/shape?handle=#{new_shape_handle}&offset=-1&table=items"
     end
 
     test "GET receives 409 to a newly created shape when shape handle is not found and no shape matches the shape definition",

packages/sync-service/test/electric/plug/serve_shape_plug_test.exs

@@ -677,15 +677,14 @@ defmodule Electric.Plug.ServeShapePlugTest do
              ]
     end
 
-    test "sends 400 when shape handle does not match shape definition",
+    test "sends 409 when shape handle does not match shape definition",
          ctx do
+      new_shape_handle = "new-shape-handle"
+
       Mock.ShapeCache
       |> expect(:get_shape, fn @test_shape, _opts -> nil end)
-      |> stub(:has_shape?, fn @test_shape_handle, _opts -> true end)
-
-      Mock.Storage
-      |> stub(:for_shape, fn @test_shape_handle, opts ->
-        {@test_shape_handle, opts}
+      |> expect(:get_or_create_shape_handle, fn @test_shape, _opts ->
+        {new_shape_handle, @test_offset}
       end)
 
       conn =
@@ -697,13 +696,14 @@ defmodule Electric.Plug.ServeShapePlugTest do
         )
         |> call_serve_shape_plug(ctx)
 
-      assert conn.status == 400
+      assert conn.status == 409
 
-      assert Jason.decode!(conn.resp_body) == %{
-               "message" =>
-                 "The specified shape definition and handle do not match." <>
-                   " Please ensure the shape definition is correct or omit the shape handle from the request to obtain a new one."
-             }
+      assert Jason.decode!(conn.resp_body) == [%{"headers" => %{"control" => "must-refetch"}}]
+      assert get_resp_header(conn, "electric-handle") == [new_shape_handle]
+
+      assert get_resp_header(conn, "location") == [
+               "/?handle=#{new_shape_handle}&offset=-1&table=public.users"
+             ]
     end
 
     test "sends 400 when omitting primary key columns in selection", ctx do

packages/sync-service/test/electric/shapes/api_test.exs

@@ -209,11 +209,11 @@ defmodule Electric.Shapes.ApiTest do
       |> expect(:get_shape, fn @test_shape, _opts ->
         nil
       end)
-      |> expect(:has_shape?, fn ^request_handle, _opts ->
-        true
+      |> expect(:get_or_create_shape_handle, fn @test_shape, _opts ->
+        {@test_shape_handle, @test_offset}
       end)
 
-      assert {:error, %{status: 400} = response} =
+      assert {:error, %{status: 409} = response} =
                Api.validate(
                  ctx.api,
                  %{
@@ -223,9 +223,8 @@ defmodule Electric.Shapes.ApiTest do
                  }
                )
 
-      assert %{
-               message: "The specified shape definition and handle do not match" <> _
-             } = response_body(response)
+      assert response.handle == @test_shape_handle
+      assert [%{headers: %{control: "must-refetch"}}] = response_body(response)
     end
 
     test "shape for handle does not match the shape definition", ctx do
@@ -235,11 +234,8 @@ defmodule Electric.Shapes.ApiTest do
       |> expect(:get_shape, fn @test_shape, _opts ->
         {@test_shape_handle, @before_all_offset}
       end)
-      |> expect(:has_shape?, fn ^request_handle, _opts ->
-        true
-      end)
 
-      assert {:error, %{status: 400} = response} =
+      assert {:error, %{status: 409} = response} =
                Api.validate(
                  ctx.api,
                  %{
@@ -249,9 +245,8 @@ defmodule Electric.Shapes.ApiTest do
                  }
                )
 
-      assert %{
-               message: "The specified shape definition and handle do not match" <> _
-             } = response_body(response)
+      assert response.handle == @test_shape_handle
+      assert [%{headers: %{control: "must-refetch"}}] = response_body(response)
     end
 
     test "returns a 409 error when requested shape handle does not exist", ctx do
@@ -261,9 +256,6 @@ defmodule Electric.Shapes.ApiTest do
       |> expect(:get_shape, fn @test_shape, _opts ->
         {@test_shape_handle, @before_all_offset}
       end)
-      |> expect(:has_shape?, fn ^request_handle, _opts ->
-        false
-      end)
 
       assert {:error, %{status: 409} = response} =
                Api.validate(

packages/typescript-client/test/integration.test.ts

@@ -635,8 +635,8 @@ describe(`HTTP Sync`, () => {
     const shapeOffset = res.headers.get(`electric-offset`)!
     const fakeMidOffset = shapeOffset
       .split(`_`)
-      .map(BigInt)
-      .map((x, i) => (i === 0 ? x - BigInt(1) : x))
+      .map(Number)
+      .map((x, i) => (i === 0 ? x - 1 : x))
       .join(`_`)
     const etag = res.headers.get(`etag`)
     expect(etag, `Response should have etag header`).not.toBe(null)
@@ -818,9 +818,8 @@ describe(`HTTP Sync`, () => {
         const responseSize = responseSizes.shift()
         const isLastResponse = responseSizes.length === 0
         if (!isLastResponse) {
-          // expect chunks to be close to 10 kB +- some kB
-          const expectedSize = 10 * 1e3
-          expect(responseSize).closeTo(expectedSize, expectedSize * 0.2)
+          // expect chunks to be close to 10 kB +- 20%
+          expect(responseSize).closeTo(10 * 1e3, 2e3)
         } else {
           // expect last response to be ~ 10 kB or less
           expect(responseSize).toBeLessThan(11 * 1e3)
@@ -842,7 +841,7 @@ describe(`HTTP Sync`, () => {
       url: `${BASE_URL}/v1/shape`,
       params: {
         table: issuesTableUrl,
-        where: `1=1`,
+        where: `1 x 1`, // invalid SQL
       },
       signal: aborter.signal,
       handle: streamState.handle,
@@ -859,16 +858,18 @@ describe(`HTTP Sync`, () => {
     expect(invalidIssueStream.error).instanceOf(FetchError)
     expect((invalidIssueStream.error! as FetchError).status).toBe(400)
     expect(invalidIssueStream.isConnected()).false
-    expect(error!.message).contains(
-      `The specified shape definition and handle do not match. Please ensure the shape definition is correct or omit the shape handle from the request to obtain a new one.`
-    )
+    expect((error! as FetchError).json).toStrictEqual({
+      message: `Invalid request`,
+      errors: {
+        where: [`At location 17: syntax error at or near "x"`],
+      },
+    })
   })
 
   it(`should detect shape deprecation and restart syncing`, async ({
     expect,
     insertIssues,
     issuesTableUrl,
-    waitForIssues,
     aborter,
     clearIssuesShape,
   }) => {
@@ -878,12 +879,23 @@ describe(`HTTP Sync`, () => {
     await insertIssues({ id: rowId, title: `foo1` })
 
     const statusCodesReceived: number[] = []
+    let numRequests = 0
 
-    let fetchPausePromise = Promise.resolve()
     const fetchWrapper = async (...args: Parameters<typeof fetch>) => {
-      await fetchPausePromise
+      // before any subsequent requests after the initial one, ensure
+      // that the existing shape is deleted and some more data is inserted
+      if (numRequests === 2) {
+        await insertIssues({ id: rowId2, title: `foo2` })
+        await clearIssuesShape(issueStream.shapeHandle)
+      }
+
+      numRequests++
       const response = await fetch(...args)
-      if (response.status < 500) statusCodesReceived.push(response.status)
+
+      if (response.status < 500) {
+        statusCodesReceived.push(response.status)
+      }
+
       return response
     }
 
@@ -911,19 +923,11 @@ describe(`HTTP Sync`, () => {
           expect(statusCodesReceived).toHaveLength(2)
           expect(statusCodesReceived[0]).toBe(200)
           expect(statusCodesReceived[1]).toBe(200)
-
-          // before any subsequent requests after the initial one, ensure
-          // that the existing shape is deleted and some more data is inserted
-          fetchPausePromise = Promise.resolve().then(async () => {
-            await insertIssues({ id: rowId2, title: `foo2` })
-            await waitForIssues({ numChangesExpected: 2 })
-            await clearIssuesShape(issueStream.shapeHandle)
-          })
         } else if (upToDateReachedCount === 2) {
           // the next up to date message should have had
           // a 409 interleaved before it that instructed the
           // client to go and fetch data from scratch
-          expect(statusCodesReceived.length).toBeGreaterThanOrEqual(5)
+          expect(statusCodesReceived).toHaveLength(5)
           expect(statusCodesReceived[2]).toBe(409)
           expect(statusCodesReceived[3]).toBe(200)
           return res()