fix (sync service): validate offset is not bigger than last offset fo…

…r the shape (#2091)

This PR fixes issue
#2063.

.changeset/famous-bananas-report.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Validate that user provided offset is not bigger than the shape's latest offset.

packages/sync-service/lib/electric/plug/serve_shape_plug.ex

@@ -4,6 +4,7 @@ defmodule Electric.Plug.ServeShapePlug do
 
   # The halt/1 function is redefined further down below
   import Plug.Conn, except: [halt: 1]
+  import Electric.Replication.LogOffset, only: [is_log_offset_lt: 2]
 
   alias Electric.Plug.Utils
   import Electric.Plug.Utils, only: [hold_conn_until_stack_ready: 2]
@@ -26,6 +27,9 @@ defmodule Electric.Plug.ServeShapePlug do
                                  "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."
                              })
+  @offset_out_of_bounds Jason.encode!(%{
+                          offset: ["out of bounds for this shape"]
+                        })
 
   defmodule Params do
     use Ecto.Schema
@@ -234,6 +238,20 @@ defmodule Electric.Plug.ServeShapePlug do
     end
   end
 
+  defp handle_shape_info(
+         %Conn{assigns: %{handle: shape_handle, offset: offset}} = conn,
+         {active_shape_handle, last_offset}
+       )
+       when (is_nil(shape_handle) or shape_handle == active_shape_handle) and
+              is_log_offset_lt(last_offset, offset) do
+    # We found a shape that matches the shape definition
+    # and the shape has the same ID as the shape handle provided by the user
+    # but the provided offset is wrong as it is greater than the last offset for this shape
+    conn
+    |> send_resp(400, @offset_out_of_bounds)
+    |> halt()
+  end
+
   defp handle_shape_info(
          %Conn{assigns: %{handle: shape_handle}} = conn,
          {active_shape_handle, last_offset}

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

@@ -152,6 +152,30 @@ defmodule Electric.Plug.ServeShapePlugTest do
              }
     end
 
+    test "returns 400 when offset is out of bounds", ctx do
+      Mock.ShapeCache
+      |> expect(:get_shape, fn @test_shape, _opts ->
+        {@test_shape_handle, @test_offset}
+      end)
+
+      invalid_offset = LogOffset.increment(@test_offset)
+
+      conn =
+        ctx
+        |> conn(
+          :get,
+          %{"table" => "public.users"},
+          "?handle=#{@test_shape_handle}&offset=#{invalid_offset}"
+        )
+        |> call_serve_shape_plug(ctx)
+
+      assert conn.status == 400
+
+      assert Jason.decode!(conn.resp_body) == %{
+               "offset" => ["out of bounds for this shape"]
+             }
+    end
+
     test "returns 400 for live request when offset == -1", ctx do
       conn =
         ctx