Electric.ShapeCache.ShapeStatus (electric v1.7.10)

Copy Markdown View Source

Keeps track of shape state.

Can recover basic persisted shape metadata from shape storage to repopulate the in-memory cache.

The shape cache then loads this and starts processes (storage and consumer) for each {shape_handle, %Shape{}} pair. These then use their attached storage to recover the status information for the shape (snapshot xmin and latest offset).

The ETS metadata table name is part of the config because we need to be able to access the data in the ETS from anywhere, so there's an internal api, using the full state and an external api using just the table name.

Summary

Functions

Where as fetch_handle_by_shape/2 may under high-write load return stale data -- not finding a shape that has been written -- due to SQLite's cross-connection durability when in WAL mode, where a connection reads against a snapshot of the data, this version does the lookup via the write connection, which is guaranteed to see all writes (SQLite connections can always see their own writes).

Runs a validation step on the existing Shape data.

Remove every shape involved in a subquery (the outer shape plus its dependency materializers) from shape metadata and on-disk storage.

Refresh the shape meta table from SQLite without recreating it.

Given a list of {handle, shape} pairs (e.g. from list_shapes/1), return the set of handles for every shape involved in a subquery: each shape with a non-empty shape_dependencies plus all of its dependency handles.

Sets the last read time for the given shape to the provided time.

Updates the last read time for the given shape to the current time.

Cheaply validate that a shape handle matches the shape definition by matching the shape's saved hash against the provided shape's hash.

Types

shape_counts()

@type shape_counts() :: %{
  total: non_neg_integer(),
  indexed: non_neg_integer(),
  unindexed: non_neg_integer()
}

shape_handle()

@type shape_handle() :: Electric.shape_handle()

stack_id()

@type stack_id() :: Electric.stack_id()

Functions

add_shape(stack_id, shape)

@spec add_shape(stack_id(), Electric.Shapes.Shape.t()) ::
  {:ok, shape_handle()} | {:error, term()}

count_shapes(stack_id)

@spec count_shapes(stack_id()) :: non_neg_integer()

fetch_handle_by_shape(stack_id, shape)

@spec fetch_handle_by_shape(stack_id(), Electric.Shapes.Shape.t()) ::
  {:ok, shape_handle()} | :error

fetch_handle_by_shape_critical(stack_id, shape)

@spec fetch_handle_by_shape_critical(stack_id(), Electric.Shapes.Shape.t()) ::
  {:ok, shape_handle()} | :error

Where as fetch_handle_by_shape/2 may under high-write load return stale data -- not finding a shape that has been written -- due to SQLite's cross-connection durability when in WAL mode, where a connection reads against a snapshot of the data, this version does the lookup via the write connection, which is guaranteed to see all writes (SQLite connections can always see their own writes).

This guarantees that will will return consistent restults at the cost of slower lookups as we're contending with access to the single write connection.

fetch_shape_by_handle(stack_id, shape_handle)

@spec fetch_shape_by_handle(stack_id(), shape_handle()) ::
  {:ok, Electric.Shapes.Shape.t()} | :error

has_shape_handle?(stack_id, shape_handle)

@spec has_shape_handle?(stack_id(), shape_handle()) :: boolean()

initialize(stack_id)

@spec initialize(stack_id()) :: :ok | {:error, term()}

Runs a validation step on the existing Shape data.

The database path is dependent on both @version above and ShapeDb.Connection's @schema_version.

A change to either of those, or to the OTP release, will result in an empty database.

least_recently_used(stack_id, shape_count)

list_shape_handles_for_relations(stack_id, relations)

@spec list_shape_handles_for_relations(stack_id(), [Electric.oid_relation()]) :: [
  shape_handle()
]

list_shapes(stack_id)

@spec list_shapes(stack_id()) :: [{shape_handle(), Electric.Shapes.Shape.t()}]

mark_snapshot_complete(stack_id, shape_handle)

@spec mark_snapshot_complete(stack_id(), shape_handle()) :: :ok | :error

mark_snapshot_started(stack_id, shape_handle)

@spec mark_snapshot_started(stack_id(), shape_handle()) :: :ok | :error

prune_subquery_shapes(stack_id)

@spec prune_subquery_shapes(stack_id()) :: :ok

Remove every shape involved in a subquery (the outer shape plus its dependency materializers) from shape metadata and on-disk storage.

Correctly restoring a subquery shape's on-disk view together with its dependency materializer across a restart is not yet implemented, so for now we drop every subquery shape on restart and let clients re-request them from scratch. This is called once at the start of the shape subsystem's restore — from ShapeLogCollector's restore_shapes, before it (or the shape consumers) rebuild any state from list_shapes/1 — so no restore path ever reinstates a subquery shape. At that point no consumer or routing entry exists for these shapes yet, so a direct metadata + storage delete is sufficient; there is no consumer to stop or routing entry to clear.

reduce_shapes(stack_id, acc, reducer_fun)

refresh(stack_id)

@spec refresh(stack_id()) :: :ok | {:error, term()}

Refresh the shape meta table from SQLite without recreating it.

Used on lock acquisition to pick up any shapes that were created/deleted by a previous instance while we were in read-only mode.

remove_shape(stack_id, shape_handle)

@spec remove_shape(stack_id(), shape_handle()) :: :ok | {:error, term()}

reset(stack_id)

@spec reset(stack_id()) :: :ok

shape_counts(stack_id)

@spec shape_counts(stack_id()) :: shape_counts() | :error

shape_has_been_activated?(stack_id, shape_handle)

@spec shape_has_been_activated?(stack_id(), shape_handle()) :: boolean()

snapshot_started?(stack_id, shape_handle)

subquery_shape_handles(handles_and_shapes)

@spec subquery_shape_handles([{shape_handle(), Electric.Shapes.Shape.t()}]) ::
  MapSet.t(shape_handle())

Given a list of {handle, shape} pairs (e.g. from list_shapes/1), return the set of handles for every shape involved in a subquery: each shape with a non-empty shape_dependencies plus all of its dependency handles.

This is the transitive closure of the subquery hierarchy — a nested dependency that itself has a subquery matches the same filter and contributes its own dependencies — so it covers outer shapes, intermediate dependencies and leaf materializers. Used by prune_subquery_shapes/1.

update_last_read_time(stack_id, shape_handle, time)

Sets the last read time for the given shape to the provided time.

Used for tests, otherwise prefer update_last_read_time_to_now/2.

update_last_read_time_to_now(stack_id, shape_handle)

Updates the last read time for the given shape to the current time.

validate_shape_handle(stack_id, shape_handle, shape)

@spec validate_shape_handle(stack_id(), shape_handle(), Electric.Shapes.Shape.t()) ::
  :ok | :error

Cheaply validate that a shape handle matches the shape definition by matching the shape's saved hash against the provided shape's hash.

version()

@spec version() :: pos_integer()