Packages
electric
1.0.19
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/shape_cache/compaction_runner.ex
defmodule Electric.ShapeCache.CompactionRunner do
use GenServer
require Logger
alias Electric.ShapeCache.Storage
@schema NimbleOptions.new!(
stack_id: [type: :string, required: true],
shape_handle: [type: :string, required: true],
storage: [type: :mod_arg, required: true],
compaction_period: [type: :non_neg_integer, default: :timer.minutes(10)]
)
def start_link(opts) do
with {:ok, opts} <- NimbleOptions.validate(opts, @schema) do
GenServer.start_link(__MODULE__, opts, name: name(opts))
end
end
def name(opts) do
Electric.ProcessRegistry.name(opts[:stack_id], __MODULE__, opts[:shape_handle])
end
@impl GenServer
def init(opts) do
clean_after_period(opts)
Process.set_label({:compaction_runner, opts[:stack_id], opts[:shape_handle]})
Logger.metadata(stack_id: opts[:stack_id], shape_handle: opts[:shape_handle])
{:ok, opts}
end
@impl GenServer
def handle_info(:clean, opts) do
Logger.info("Triggering compaction for shape #{opts[:shape_handle]}")
clean_after_period(opts)
Storage.compact(opts[:storage])
Logger.info("Compaction complete for shape #{opts[:shape_handle]}")
{:noreply, opts}
end
defp clean_after_period(opts) do
# add a large random jitter to avoid all compactions happening at the same time
half_period = div(opts[:compaction_period], 2)
next_msg = opts[:compaction_period] + Enum.random(-half_period..half_period)
Process.send_after(self(), :clean, next_msg)
end
end