From dce5d173efb74c9dd8681daba53c2756fb4a9f35 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 1 Jul 2026 14:25:35 +0100 Subject: [PATCH 1/2] Add regression test for subquery materializer multi-chunk log replay (bug 4) The bug-4 fix (main-log short-circuits in `Materializer.read_history_up_to_subscribed/2`) shipped with #4651 but had no dedicated regression test. This adds a deterministic unit test: with a small `chunk_bytes_threshold` the source shape's persisted main log spans more than one chunk, and startup replay must not re-read (and re-apply) main-log entries it already applied. On the pre-fix naive iteration the test crashes the materializer with `Key ... already exists`; with the fix it replays the full history exactly once. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../shapes/consumer/materializer_test.exs | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) diff --git a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs index ed23cd9429..00a4fc8aa0 100644 --- a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs +++ b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs @@ -1437,6 +1437,92 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do # iteration fix guarantees the UPDATE is replayed. assert Materializer.get_link_values(mat_ctx) == MapSet.new([99]) end + + # Bug 4: 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). A startup-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 a key that already exists raises + # "Key already exists" inside the materializer, crashing it (and the + # dependent shape's consumer). The fix stops iterating as soon as the + # read steps into the main log. + @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 buggy iteration + # re-reads the second chunk and re-applies the insert for key "3", + # raising "Key 3 already exists". + 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, the + # scenario wouldn't exercise the bug. + 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 + + # All three values must be present exactly once. On the buggy iteration + # the materializer crashes before this point. + assert Materializer.get_link_values(mat_ctx) == MapSet.new([10, 20, 30]) + end end describe "startup race condition handling" do From 0cfe76a4f6fb4afd43bab963c444922f8722901c Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 2 Jul 2026 15:57:53 +0100 Subject: [PATCH 2/2] Reframe bug-4 test comments to describe the invariant, not past behavior Address review: drop references to "the buggy iteration" (reverted code state) and describe what the test guards in present tense. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../shapes/consumer/materializer_test.exs | 33 ++++++++++--------- 1 file changed, 17 insertions(+), 16 deletions(-) diff --git a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs index 00a4fc8aa0..5cee0c2428 100644 --- a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs +++ b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs @@ -1438,15 +1438,15 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do assert Materializer.get_link_values(mat_ctx) == MapSet.new([99]) end - # Bug 4: 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). A startup-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 a key that already exists raises - # "Key already exists" inside the materializer, crashing it (and the - # dependent shape's consumer). The fix stops iterating as soon as the - # read steps into the main log. + # 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()}" @@ -1464,9 +1464,9 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do # 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 buggy iteration - # re-reads the second chunk and re-applies the insert for key "3", - # raising "Key 3 already exists". + # 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) @@ -1492,8 +1492,9 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do # 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, the - # scenario wouldn't exercise the bug. + # 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) @@ -1519,8 +1520,8 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle} assert Materializer.wait_until_ready(mat_ctx) == :ok - # All three values must be present exactly once. On the buggy iteration - # the materializer crashes before this point. + # 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