Packages
electric
0.4.4
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/timeline.ex
defmodule Electric.Timeline do
@moduledoc """
Module exporting functions for handling Postgres timelines.
Verifies the Postgres ID and its timeline.
"""
require Logger
alias Electric.PersistentKV
@type pg_id :: non_neg_integer()
@type timeline_id :: integer()
@type timeline :: {pg_id(), timeline_id()} | nil
@timeline_key "timeline_id"
@doc """
Checks that we're connected to the same Postgres DB as before and on the same timeline.
TO this end, it checks the provided `pg_id` against the persisted PG ID.
If the PG IDs match, it also checks the provided `pg_timeline` against the persisted timeline.
Normally, Postgres and Electric are on the same timeline and nothing must be done.
If the timelines differ, that indicates that a Point In Time Recovery (PITR) has occurred and all shapes must be cleaned.
If we fail to fetch timeline information, we also clean all shapes for safety as we can't be sure that Postgres and Electric are on the same timeline.
"""
@spec check(timeline(), keyword()) :: :ok
def check(pg_timeline, opts) do
electric_timeline = load_timeline(opts)
verify_timeline(pg_timeline, electric_timeline, opts)
end
@spec verify_timeline(timeline(), timeline(), keyword()) :: :ok
defp verify_timeline({pg_id, timeline_id} = timeline, timeline, _) do
Logger.info("Connected to Postgres #{pg_id} and timeline #{timeline_id}")
:ok
end
defp verify_timeline({pg_id, timeline_id} = timeline, nil, opts) do
Logger.info("No previous timeline detected.")
Logger.info("Connected to Postgres #{pg_id} and timeline #{timeline_id}")
store_timeline(timeline, opts)
end
defp verify_timeline({pg_id, _} = timeline, {electric_pg_id, _}, opts)
when pg_id != electric_pg_id do
Logger.warning(
"Detected different Postgres DB, with ID: #{pg_id}. Old Postgres DB had ID #{electric_pg_id}. Cleaning all shapes."
)
clean_all_shapes_and_store_timeline(timeline, opts)
end
defp verify_timeline({_, timeline_id} = timeline, _, opts) do
Logger.warning("Detected PITR to timeline #{timeline_id}; cleaning all shapes.")
clean_all_shapes_and_store_timeline(timeline, opts)
end
defp clean_all_shapes_and_store_timeline(timeline, opts) do
clean_all_shapes(opts)
store_timeline(timeline, opts)
end
# Loads the PG ID and timeline ID from persistent storage
@spec load_timeline(keyword()) :: timeline()
def load_timeline(opts) do
kv = make_serialized_kv(opts)
case PersistentKV.get(kv, @timeline_key) do
{:ok, [pg_id, timeline_id]} ->
{pg_id, timeline_id}
{:error, :not_found} ->
nil
error ->
Logger.warning("Failed to load timeline ID from persistent storage: #{error}")
nil
end
end
def store_timeline({pg_id, timeline_id}, opts) do
kv = make_serialized_kv(opts)
:ok = PersistentKV.set(kv, @timeline_key, [pg_id, timeline_id])
end
defp make_serialized_kv(opts) do
kv_backend = Keyword.fetch!(opts, :persistent_kv)
# defaults to using Jason encoder and decoder
PersistentKV.Serialized.new!(backend: kv_backend)
end
# Clean up all data (meta data and shape log + snapshot) associated with all shapes
@spec clean_all_shapes(keyword()) :: :ok
defp clean_all_shapes(opts) do
{shape_cache, opts} = Access.get(opts, :shape_cache, {ShapeCache, []})
shape_cache.clean_all_shapes(opts)
:ok
end
end