Packages
surgex
4.14.0
6.0.1
6.0.0
5.1.1
5.1.0
5.0.0
5.0.0-git-015aa39a
4.15.2
4.15.1
4.15.1-git-98b3
4.15.0
4.15.0-git-943e
4.15.0-git-5806
4.14.1
4.14.0
4.13.1
4.13.0
4.12.0
4.11.0
4.11.0-git-97d4
4.11.0-git-14c8
4.10.0
4.10.0-git-37f2
4.9.0
4.9.0-git-9d5f
4.8.1-git-80a7
4.8.0
4.8.0-git-fbda
4.8.0-git-e058
4.7.0
4.7.0-git-5414
4.6.1
4.6.0
4.5.0
4.4.0
4.3.0
4.2.1
4.2.0
4.1.1
4.1.1-git-9f6b
4.1.1-git-6f0cd
4.1.0
4.1.0-git-e934
4.1.0-git-8c28
4.1.0-git-56a9
4.1.0-git-55fd
4.0.0
3.2.8
3.2.7
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
retired
3.1.0
3.0.0
2.24.1
2.24.0
retired
2.23.0
2.22.0
2.21.0
2.20.1
2.20.0
2.19.0
2.18.1
2.18.0
2.17.0
2.16.0
2.15.0
2.14.0
2.13.0
2.12.1
2.12.0
2.11.0
2.10.0
2.9.0
2.8.0
2.7.0
2.6.0
2.5.1
2.5.0
2.4.0
retired
2.3.0
2.2.1
2.2.0
2.1.1
2.1.0
2.0.0
1.6.0
1.5.2
1.5.1
1.5.0
1.4.0
1.3.1
retired
1.3.0
retired
1.2.1
1.2.0
1.1.0
1.0.0
0.5.2
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.0
0.1.0
All Things Elixir @ Surge Ventures Inc, the creators of Fresha
Current section
Files
Jump to
Current section
Files
lib/surgex/data_pipe/follower_sync.ex
defmodule Surgex.DataPipe.FollowerSync do
@moduledoc """
Acquires a PostgreSQL slave synchronization with a remote master.
## Usage
It can be configured globally or per repo as follows:
config :surgex,
follower_sync_enabled: true,
follower_sync_timeout: 15_000,
follower_sync_interval: 1_000
config :my_project, MyProject.MyRepo,
# ...
follower_sync_enabled: true,
follower_sync_timeout: 15_000,
follower_sync_interval: 1_000
As a convenience versus calling `Surgex.DataPipe.FollowerSync.call/2` all the time, it can be
`use`d in a repo module as follows:
defmodule MyProject.MyRepo do
use Surgex.DataPipe.FollowerSync
end
MyProject.MyRepo.ensure_follower_sync(lsn)
Refer to `Surgex.DataPipe` for a complete data pipe example.
"""
require Logger
alias Surgex.DataPipe.{FollowerSync, PostgresSystemUtils}
defmacro __using__(_) do
quote do
@doc """
Waits for slave repo to catch up with master's changes up to specified log location (lsn).
"""
def ensure_follower_sync(lsn) do
FollowerSync.call(__MODULE__, lsn)
end
end
end
@doc """
Waits for a given slave repo's sync up to specific remote master's lsn.
"""
def call(repo, lsn) do
cond do
!enabled?(repo) ->
:ok
!PostgresSystemUtils.lsn_valid?(lsn) ->
Logger.warn("Invalid LSN: #{inspect(lsn)}")
{:error, :invalid_lsn}
true ->
wait_for_sync(repo, lsn)
end
end
defp wait_for_sync(repo, lsn, start_time \\ get_current_time()) do
case PostgresSystemUtils.get_last_wal_replay_lsn(repo) do
{:ok, last_lsn} ->
handle_lsn_update(repo, lsn, last_lsn, start_time)
:error ->
Logger.warn(fn -> "No replay LSN (consider setting follower_sync_enabled: false)" end)
{:error, :no_replay_lsn}
end
end
defp handle_lsn_update(repo, lsn, last_lsn, start_time) do
current_time = get_current_time()
elapsed_time = current_time - start_time
timeout = get_timeout(repo)
interval = get_interval(repo)
cond do
normalize_lsn(last_lsn) >= normalize_lsn(lsn) ->
Logger.info(fn -> "Follower sync acquired after #{elapsed_time}ms" end)
:ok
elapsed_time >= timeout ->
Logger.warn(fn -> "Follower sync timeout after #{timeout}ms: #{last_lsn} < #{lsn}" end)
{:error, :timeout}
true ->
:timer.sleep(interval)
wait_for_sync(repo, lsn, start_time)
end
end
defp normalize_lsn(lsn), do: String.pad_leading(lsn, 16, "0")
defp get_current_time, do: :os.system_time(:milli_seconds)
defp enabled?(repo), do: get_config(repo, :follower_sync_enabled, true)
defp get_timeout(repo), do: get_config(repo, :follower_sync_timeout, 15_000)
defp get_interval(repo), do: get_config(repo, :follower_sync_interval, 1_000)
defp get_config(repo, key, default) do
case Confix.get_in([repo, key]) do
nil ->
Application.get_env(:surgex, key, default)
repo_value ->
repo_value
end
end
end