Packages
electric
1.4.7
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.10
1.6.9
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.16
1.4.16-beta-1
1.4.15
1.4.14
1.4.13
1.4.12
1.4.11
1.4.10
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.4
1.3.3
1.3.2
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.14
1.1.13
1.1.12
1.1.11
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
retired
1.1.5
retired
1.1.4
retired
1.1.3
retired
1.1.2
1.1.1
1.1.0
1.0.24
1.0.23
1.0.22
1.0.21
1.0.20
1.0.19
1.0.18
1.0.17
1.0.15
1.0.13
1.0.12
1.0.11
1.0.10
1.0.9
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-beta.23
1.0.0-beta.22
1.0.0-beta.20
1.0.0-beta.19
1.0.0-beta.18
1.0.0-beta.17
1.0.0-beta.16
1.0.0-beta.15
1.0.0-beta.14
1.0.0-beta.13
1.0.0-beta.12
1.0.0-beta.11
1.0.0-beta.10
1.0.0-beta.9
1.0.0-beta.8
1.0.0-beta.7
1.0.0-beta.6
1.0.0-beta.5
1.0.0-beta.4
1.0.0-beta.3
1.0.0-beta.2
1.0.0-beta.1
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.5.2
0.4.4
Postgres sync engine. Sync little subsets of your Postgres data into local apps and services.
Current section
Files
Jump to
Current section
Files
lib/electric/replication/schema_reconciler.ex
defmodule Electric.Replication.SchemaReconciler do
@moduledoc """
Takes care of periodically reconciling the schema of the database with
the inspector caches and active shapes.
Covers cases where either the table was recreated and thus isn't in the
publication anymore, or where some alterations were made to the schema,
but we don't see them because there were no writes to the affected tables.
"""
use GenServer
require Logger
alias Electric.Postgres.Inspector
alias Electric.ShapeCache.ShapeCleaner
@name_schema_tuple {:tuple, [:atom, :atom, :any]}
@genserver_name_schema {:or, [:atom, @name_schema_tuple]}
@schema NimbleOptions.new!(
name: [type: @genserver_name_schema, required: false],
period: [type: :pos_integer, required: false, default: 60_000],
inspector: [type: :any, required: true],
stack_id: [type: :string, required: true]
)
def start_link(opts) do
with {:ok, opts} <- NimbleOptions.validate(Map.new(opts), @schema) do
name = Map.get(opts, :name, name(opts.stack_id))
GenServer.start_link(__MODULE__, opts, name: name)
end
end
def name(stack_id) do
Electric.ProcessRegistry.name(stack_id, __MODULE__)
end
@doc """
Triggers an immediate schema reconciliation check.
"""
def reconcile_now(name_or_pid) do
GenServer.call(name_or_pid, :reconcile_now)
end
def init(opts) do
Process.set_label({:schema_reconciler, opts.stack_id})
Logger.metadata(stack_id: opts.stack_id)
Electric.Telemetry.Sentry.set_tags_context(stack_id: opts.stack_id)
Logger.debug("SchemaReconciler started with state #{inspect(opts)}")
{:ok, opts, {:continue, :reconcile}}
end
def handle_continue(:schedule_next_check, state) do
Process.send_after(self(), :reconcile, state.period)
{:noreply, state}
end
def handle_continue(:reconcile, state) do
_ = handle_reconcile(state)
{:noreply, state, {:continue, :schedule_next_check}}
end
def handle_call(:reconcile_now, _from, state) do
_ = handle_reconcile(state)
{:reply, :ok, state}
end
def handle_info(:reconcile, state) do
{:noreply, state, {:continue, :reconcile}}
end
defp handle_reconcile(state) do
case Electric.StatusMonitor.status(state.stack_id) do
%{conn: :up} ->
state.inspector
|> Inspector.list_relations_with_stale_cache()
|> handle_diverged_relations(state)
status ->
Logger.debug("Schema reconciliation skipped due to inactive stack: #{inspect(status)}")
end
catch
# We essentially never want to fail here, as this is a periodic task.
# If it fails, we'll just try again next time, so no additional retries are implemented
type, reason ->
st = __STACKTRACE__
Logger.error("Schema reconciliation failed: #{Exception.format(type, reason, st)}")
:error
end
defp handle_diverged_relations({:ok, []}, _state), do: :ok
defp handle_diverged_relations({:ok, diverged_relations}, state) do
with :ok <- ShapeCleaner.remove_shapes_for_relations(state.stack_id, diverged_relations),
:ok <-
Enum.each(diverged_relations, fn {oid, _} -> Inspector.clean(oid, state.inspector) end) do
:ok
else
{:error, reason} ->
Logger.warning("Schema reconciliation failed: #{reason}")
:error
end
end
defp handle_diverged_relations(:error, _state),
do: Logger.warning("Schema reconciliation failed while fetching fresh relations")
end