litefs

defmodule Litefs do
  @moduledoc """

  Litefs sets a primary file in the directory of the mount it is managing to indicate
  the location of the primary when on a replica. There is no file on the primary.

  Elixir communicates in a mesh with names that may be different than the hostname
  that is contained in the file managed by Litefs. As Litefs can change the primary
  without notice, we need to periodically update the state of the primary in memory.

  There are several cases when we try to update the primary:
  1) when a node goes connects / up / down
  2) after the primary_check_time has passed which is default 30 seconds

  NOTE 2022-08-24

  On a litefs crash with a stale filesystem, there are some errors to handle.
  1) Elixir will have a stale filehandle and will return (Exqlite.Error) disk I/O error
  2) Litefs will not be able to unmount or restart as Elixir has the file handle open.

  Based on the fly_rpc and fly_postgres packages.

  """

  use GenServer
  require Logger

  @tab :litefs
  @primary_check_time 30_000
  @verbose_log false

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(opts) do
    _tab = :ets.new(@tab, [:named_table, :public, read_concurrency: true])

    primary_file_path =
      opts
      |> Keyword.get(:database)
      |> Path.dirname()

    primary_file_path = primary_file_path <> "/.primary"
    set_primary_path(primary_file_path)

    # monitor new node up/down activity
    :global_group.monitor_nodes(true)
    {:ok, %{}, {:continue, :update_primary}}
  end

  def set(key, value) do
    :ets.insert(@tab, {key, value})
  end

  def get(key) do
    case :ets.lookup(@tab, key) do
      [{^key, value}] -> value
      [] -> nil
    end
  end

  def set_primary_path(t), do: set(:path, t)
  def get_primary_path(), do: get(:path)
  def set_primary(name), do: set(:primary, name)
  def get_primary() do
    primary = get(:primary)
    if is_nil(primary) do
      raise "No primary found!"
    end
    primary
  end

  def check_for_litefs_primary_file() do
    File.exists?(get_primary_path())
  end

  def update_primary() do
    primary_file_path = get_primary_path()

    primary_node =
      if !File.exists?(primary_file_path) do
        Node.self()
      else
        Enum.reduce_while(Node.list(), nil, fn x, acc ->
          if !Litefs.rpc(x, __MODULE__, :check_for_litefs_primary_file, []) do
            {:halt, x}
          else
            {:cont, acc}
          end
        end)
      end

    set_primary(primary_node)

    if !is_nil(primary_node), do: :ok, else: :error
  end

  def handle_continue(:update_primary, state) do
    update_primary()
    Process.send_after(self(), :update_primary, @primary_check_time)
    {:noreply, state}
  end

  def handle_info(:update_primary, state) do
    update_primary()
    Process.send_after(self(), :update_primary, @primary_check_time)
    {:noreply, state}
  end

  def handle_info({:nodeup, node_name}, state) do
    Logger.debug("nodeup #{node_name}")
    update_primary()
    {:noreply, state}
  end

  def handle_info({:nodedown, node_name}, state) do
    Logger.debug("nodedown #{node_name}")
    update_primary()
    {:noreply, state}
  end

  @doc """
  Executes the function on the remote node and waits for the response.

  Exits after `timeout` milliseconds.
  """
  @spec rpc(node, module, func :: atom(), args :: [any], non_neg_integer()) :: any()
  def rpc(node, module, func, args, timeout \\ 5000)

  def rpc(:primary, module, func, args, timeout) do
    rpc(get_primary(), module, func, args, timeout)
  end

  def rpc(node, module, func, args, timeout) do
    verbose_log(:info, fn ->
      "RPC REQ from #{Node.self()} to #{node}: #{mfa_string(module, func, args)}"
    end)

    caller = self()
    ref = make_ref()

    # Perform the RPC call to the remote node and wait for the response
    _pid =
      Node.spawn_link(node, __MODULE__, :__local_rpc__, [
        [caller, ref, module, func | args]
      ])

    receive do
      {^ref, result} ->
        verbose_log(:info, fn ->
          "RPC RECV response from #{node} to #{Node.self()}: #{mfa_string(module, func, args)}"
        end)

        result
    after
      timeout ->
        verbose_log(:error, fn ->
          "RPC TIMEOUT from #{node} to #{Node.self}: #{mfa_string(module, func, args)}"
        end)

        exit(:timeout)
    end
  end

  @doc false
  # Private function that can be executed on a remote node in the cluster. Used
  # to execute arbitrary function from a trusted caller.
  def __local_rpc__([caller, ref, module, func | args]) do
    result = apply(module, func, args)
    send(caller, {ref, result})
  end

  defp verbose_log(kind, func) do
    if !is_nil(Application.get_env(:litefs, :verbose_logging)) or @verbose_log == true do
      Logger.log(kind, func)
    end
  end

  @doc false
  # A "private" function that converts the MFA data into a string for logging.
  def mfa_string(module, func, args) do
    "#{Atom.to_string(module)}.#{Atom.to_string(func)}/#{length(args)}"
  end
end