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

## Summary

Adds a deterministic regression test for **bug 4** from the
restore-from-file bug triage in #4648 (bug doc:
`packages/sync-service/bugs.md`).

This is a **test-only** change. Bug 4's production fix — the two
main-log short-circuits in
`Materializer.read_history_up_to_subscribed/2` — already shipped
alongside the bug 1 fix in #4651; it just lacked dedicated regression
coverage.

## Problem (bug 4)

On startup recovery the subquery materializer replays the source shape's
persisted history (snapshot + main log) up to its subscribed offset.
`Storage.get_log_stream/3` returns one chunk per call for the
**snapshot**, but returns the **entire** requested range for the **main
log** in a single call. When the source shape's main log spans more than
one chunk, a replay loop that keeps advancing through chunk boundaries
past the first main-log chunk re-reads main-log entries it has already
applied. Re-applying a `NewRecord` for an existing key raises `Key ...
already exists`, crashing the materializer (and taking the dependent
shape's consumer down with it → 409 must-refetch on the next poll).

## Solution / coverage

The fix (already merged in #4651) stops iterating as soon as the read
steps into the main log. This PR adds the missing regression test:

`test/electric/shapes/consumer/materializer_test.exs` →
`"does not re-read main-log entries when the main log spans multiple
chunks"`

- Sets `chunk_bytes_threshold: 10` to force the source shape's main log
across ≥2 chunks (asserts the first main-log chunk ends strictly before
the subscribed offset).
- Persists two `NewRecord` inserts in separate main-log chunks,
subscribes past both, and asserts startup replay yields the correct
`value_counts` exactly once.

## Test Plan

- [x] Passes with the fix in place (`59 tests, 0 failures` in the file).
- [x] Fails without the fix: reverting the short-circuits makes the test
crash with `** (RuntimeError) Key "public"."test_table"/"3" already
exists` from `read_history_up_to_subscribed/2` — confirming it's a
genuine failing-first regression test.
- [x] `mix format --check-formatted` clean.

## Notes

An end-to-end variant in `oracle_restore_test.exs` was prototyped but
not included: it does trigger the crash, but the failure self-heals via
a client refetch that re-syncs to the oracle, so it passes with and
without the fix — false-confidence coverage. The deterministic unit test
is the reliable regression guard.

---
Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: robacourt

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