Summary
Functions
Returns a specification to start this module under a supervisor.
Removes the cached link values for shape_handle from the shared ETS table.
Safe to call even if the table does not exist (e.g. after a stack shutdown).
Returns the current set of materialized link values for a shape. Checks the shared ETS cache first (written after each committed transaction); falls back to a synchronous GenServer call if the cache has no entry yet.
Callback implementation for GenServer.handle_continue/2.
Callback implementation for GenServer.init/1.
Creates the per-stack ETS table that caches link values for all materializers
in a stack. Called by ConsumerRegistry during stack initialization. Idempotent —
safe to call when the table already exists.
Replay all of the source shape's persisted history (snapshot + log) up to
state.subscribed_offset so the materializer's value_counts reflect the
on-disk state on startup.
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec delete_link_values(Electric.stack_id(), Electric.shape_handle()) :: :ok
Removes the cached link values for shape_handle from the shared ETS table.
Safe to call even if the table does not exist (e.g. after a stack shutdown).
Returns the current set of materialized link values for a shape. Checks the shared ETS cache first (written after each committed transaction); falls back to a synchronous GenServer call if the cache has no entry yet.
Callback implementation for GenServer.handle_continue/2.
Callback implementation for GenServer.init/1.
@spec init_link_values_table(stack_id :: term()) :: :ets.table() | :undefined
Creates the per-stack ETS table that caches link values for all materializers
in a stack. Called by ConsumerRegistry during stack initialization. Idempotent —
safe to call when the table already exists.
@spec link_values_table_name(Electric.stack_id()) :: atom()
@spec new_changes( map(), [Electric.Replication.Changes.change()] | {Electric.Replication.LogOffset.t(), Electric.Replication.LogOffset.t()}, keyword() ) :: :ok
Replay all of the source shape's persisted history (snapshot + log) up to
state.subscribed_offset so the materializer's value_counts reflect the
on-disk state on startup.
Storage.get_log_stream/3 returns at most one chunk per call for
snapshot chunks — but for the main log it returns the entire requested
range [min_offset, subscribed_offset] in one call. So we iterate
through snapshot chunks using Storage.get_chunk_end_log_offset/2,
and as soon as the iteration would step into the main log we stop:
the previous call already streamed everything up to the subscribed
offset. Iterating into the main log would re-read entries already
applied, producing duplicate inserts that crash the materializer.
The subscribed_offset is the Consumer's latest_offset at the time of subscription. We only read up to this offset to avoid duplicates — any changes after this offset will be delivered via new_changes messages from the Consumer.