Add regression test for subquery materializer multi-chunk log replay (bug 4)

packages/sync-service/test/electric/shapes/consumer/materializer_test.exs

@@ -1437,6 +1437,93 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
       # iteration fix guarantees the UPDATE is replayed.
       assert Materializer.get_link_values(mat_ctx) == MapSet.new([99])
     end
+
+    # When the source shape's persisted main log spans MORE than one chunk,
+    # `Storage.get_log_stream/3` returns the *entire* main-log range in a
+    # single call (chunking only applies to the snapshot). Startup replay must
+    # therefore stop iterating as soon as it steps into the main log:
+    # continuing to advance through chunk boundaries would re-read main-log
+    # entries it has already applied, and re-applying a `NewRecord` for a key
+    # that already exists raises "Key already exists", crashing the
+    # materializer and the dependent shape's consumer. This test guards that
+    # each persisted entry is applied exactly once.
+    @tag with_pure_file_storage_opts: [chunk_bytes_threshold: 10]
+    test "does not re-read main-log entries when the main log spans multiple chunks", ctx do
+      shape_handle = "multichunk-test-#{System.unique_integer()}"
+
+      storage = Storage.for_shape(shape_handle, ctx.storage)
+      Storage.start_link(storage)
+      writer = Storage.init_writer!(storage, @shape)
+      Storage.mark_snapshot_as_started(storage)
+
+      # Snapshot with one row at value=10.
+      Storage.make_new_snapshot!(
+        make_snapshot_data([%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}}]),
+        storage
+      )
+
+      # Two main-log INSERTs persisted before the materializer subscribes.
+      # With a tiny `chunk_bytes_threshold` each lands in its own main-log
+      # chunk, so the main log spans more than one chunk — the condition under
+      # which a second read of the same range would re-apply the insert for
+      # key "3".
+      offset_2 = LogOffset.new(100, 0)
+      offset_3 = LogOffset.new(200, 0)
+
+      writer =
+        Storage.append_to_log!(
+          [
+            {offset_2, ~s|"public"."test_table"/"2"|, :insert,
+             ~s|{"key":"\\"public\\".\\"test_table\\"/\\"2\\"","value":{"id":"2","value":"20"},"headers":{"operation":"insert"}}|}
+          ],
+          writer
+        )
+
+      writer =
+        Storage.append_to_log!(
+          [
+            {offset_3, ~s|"public"."test_table"/"3"|, :insert,
+             ~s|{"key":"\\"public\\".\\"test_table\\"/\\"3\\"","value":{"id":"3","value":"30"},"headers":{"operation":"insert"}}|}
+          ],
+          writer
+        )
+
+      Storage.hibernate(writer)
+
+      # Sanity check: the main log really does span more than one chunk. The
+      # first main-log chunk (the one reached from the end of the snapshot)
+      # must end strictly before the last persisted offset. Without this, a
+      # single read would cover the whole main log and the multi-chunk case
+      # under test wouldn't be exercised.
+      first_main_chunk_end =
+        Storage.get_chunk_end_log_offset(LogOffset.last_before_real_offsets(), storage)
+
+      assert not is_nil(first_main_chunk_end)
+      assert LogOffset.compare(first_main_chunk_end, offset_3) == :lt
+
+      ConsumerRegistry.register_consumer(self(), shape_handle, ctx.stack_id)
+
+      {:ok, _pid} =
+        Materializer.start_link(%{
+          stack_id: ctx.stack_id,
+          shape_handle: shape_handle,
+          storage: ctx.storage,
+          columns: ["value"],
+          materialized_type: {:array, :int8}
+        })
+
+      respond_to_call(:await_snapshot_start, :started)
+      # Subscribe past both persisted INSERTs so startup replay walks the
+      # snapshot and the whole multi-chunk main log.
+      respond_to_call(:subscribe_materializer, {:ok, offset_3})
+
+      mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle}
+      assert Materializer.wait_until_ready(mat_ctx) == :ok
+
+      # The snapshot value and both persisted INSERTs must each be applied
+      # exactly once.
+      assert Materializer.get_link_values(mat_ctx) == MapSet.new([10, 20, 30])
+    end
   end
 
   describe "startup race condition handling" do