electric-sql / electric-sql/electric

Materializer reads only one snapshot chunk from PureFileStorage, missing remaining data

Open
#3,891 1 comment 0 reactions 0 assignees View on GitHub
shape-with-subqueries
Dominant language
TypeScript
Stars
10.4k
Forks
375
Avg merge
3d 1h
Merged PRs (30d)
18

Description

## Summary

The materializer calls `Storage.get_log_stream(before_all(), subscribed_offset)` once during startup and expects to receive **all** data in that range — snapshot chunks and main log entries. However, `PureFileStorage.get_log_stream/3` is designed to return **one snapshot chunk per call**, requiring the caller to make repeated calls to read subsequent chunks and eventually the main log. The materializer doesn't do this, so it only ever reads snapshot chunk 0.

This means the materializer silently misses:
1. Snapshot chunks 1+ (for snapshots that span multiple chunks)
2. All main log entries between the snapshot boundary and `subscribed_offset`

When a subsequent DELETE arrives for a key the materializer never saw, it crashes with `KeyError` on `Map.pop!/2`.

## How PureFileStorage.get_log_stream works

The function returns a stream for a **single** chunk or log segment, based on the `min_offset`:

```elixir
# pure_file_storage.ex
case {last_snapshot_chunk, min_offset} do
{_, x} when is_min_offset(x) ->
Snapshot.stream_chunk_lines(opts, 0) # returns only chunk 0

{%LogOffset{} = latest, min_offset} when is_log_offset_lt(min_offset, latest) ->
Snapshot.stream_chunk_lines(opts, op_offset + 1) # returns next chunk

{nil, _offset} ->
wait_for_chunk_file_or_snapshot_end(opts, op_offset + 1)

{%LogOffset{}, offset} ->
stream_main_log(offset, max_offset, opts) # returns main log segment
end
```

The API layer handles this correctly by calling `get_log_stream` repeatedly — each call returns one chunk, the caller tracks the last offset, then makes another call to get the next chunk, eventually reaching the main log.

## How the materializer uses it

The materializer calls it exactly **once**:

```elixir
# materializer.ex
def get_stream_up_to_offset(min_offset, subscribed_offset, storage) do
if is_nil(subscribed_offset) or is_log_offset_lte(subscribed_offset, min_offset) do
{:ok, min_offset, []}
else
stream = Storage.get_log_stream(min_offset, subscribed_offset, storage)
{:ok, subscribed_offset, stream}
end
end
```

It gets back only snapshot chunk 0, then sets its offset to `subscribed_offset` — skipping everything in between.

## Why tests don't catch this

Existing materializer tests use `PureFileStorage` but work because:
- Snapshots are small enough to fit in one chunk
- The test helper responds to `:subscribe_materializer` with `LogOffset.last_before_real_offsets()`, meaning there are no main log entries to read

`InMemoryStorage.get_log_stream/3` has fundamentally different behavior — it returns a `ConcurrentStream` that streams **all** data from snapshot through the main log in a single call. So any test using `InMemoryStorage` (including integration tests via `with_complete_stack` prior to recent changes) would not exhibit this bug.

## Crash sequence in production

```
** (KeyError) key "\"public\".\"issues\"/\"4e8547e7-...\"" not found in: %{}
(electric 1.4.3) lib/electric/shapes/consumer/materializer.ex:366: anonymous fn/3 in Electric.Shapes.Consumer.Materializer.apply_changes/2
```

1. Inner shape created, snapshot taken (fits in one chunk)
2. Inner consumer processes a replication INSERT → writes to main log, advances `latest_offset`
3. Inner consumer does NOT forward to materializer (`materializer_subscribed? == false` — correct)
4. Materializer starts, calls `subscribe_materializer` → gets `subscribed_offset` = real offset
5. Materializer reads `get_log_stream(before_all(), real_offset)` → gets only snapshot chunk 0, misses the INSERT in the main log
6. Replication DELETE arrives → forwarded to materializer → `Map.pop!(%{}, key)` → **KeyError**

## Suggested fix

Either:
- **Fix the materializer** to call `get_log_stream` in a loop (like the API layer does), advancing `min_offset` after each chunk until it reaches `subscribed_offset`
- **Fix PureFileStorage** to return a continuous stream that chains snapshot chunks into the main log when called with `before_all()` as `min_offset`

## Reproducing in tests

The existing materializer unit tests need to be adapted to use `PureFileStorage` with scenarios where:
1. The snapshot has multiple chunks (to verify chunk chaining)
2. There are main log entries between the snapshot boundary and `subscribed_offset` (to verify snapshot-to-log chaining)

A unit test that demonstrates the gap exists in `materializer_test.exs` under "get_log_stream gap: snapshot-only read misses main log entries".

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.