Add TCP keepalive options for database connections

.changeset/db-tcp-keepalive.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Add `ELECTRIC_DATABASE_TCP_KEEPALIVE_IDLE`, `ELECTRIC_DATABASE_TCP_KEEPALIVE_INTERVAL`, `ELECTRIC_DATABASE_TCP_KEEPALIVE_COUNT` and `ELECTRIC_DATABASE_TCP_USER_TIMEOUT` for configuring TCP keepalive and user timeout on database connections. All of them are opt-in; when unset the OS defaults are kept.

packages/sync-service/config/runtime.exs

@@ -265,6 +265,25 @@ config :electric,
   prometheus_port: prometheus_port,
   live_dashboard_port: live_dashboard_port,
   db_pool_size: env!("ELECTRIC_DB_POOL_SIZE", :integer, nil),
+  db_tcp_keepalive_idle:
+    env!(
+      "ELECTRIC_DATABASE_TCP_KEEPALIVE_IDLE",
+      &Electric.Config.parse_human_readable_time!/1,
+      nil
+    ),
+  db_tcp_keepalive_interval:
+    env!(
+      "ELECTRIC_DATABASE_TCP_KEEPALIVE_INTERVAL",
+      &Electric.Config.parse_human_readable_time!/1,
+      nil
+    ),
+  db_tcp_keepalive_count: env!("ELECTRIC_DATABASE_TCP_KEEPALIVE_COUNT", :integer, nil),
+  db_tcp_user_timeout:
+    env!(
+      "ELECTRIC_DATABASE_TCP_USER_TIMEOUT",
+      &Electric.Config.parse_human_readable_time!/1,
+      nil
+    ),
   replication_stream_id: replication_stream_id,
   replication_slot_temporary?: env!("CLEANUP_REPLICATION_SLOTS_ON_SHUTDOWN", :boolean, nil),
   replication_slot_temporary_random_name?:

packages/sync-service/lib/electric/application.ex

@@ -140,6 +140,7 @@ defmodule Electric.Application do
       ],
       pool_opts:
         get_env_lazy(opts, :pool_opts, fn -> [pool_size: get_env(opts, :db_pool_size)] end),
+      tcp_opts: tcp_opts(opts),
       chunk_bytes_threshold: get_env(opts, :chunk_bytes_threshold),
       telemetry_opts:
         telemetry_opts([instance_id: instance_id, installation_id: installation_id] ++ opts),
@@ -241,6 +242,39 @@ defmodule Electric.Application do
     ]
   end
 
+  # The keepidle, keepintvl, keepcnt and user_timeout socket options are only
+  # compiled into Erlang's inet driver where the OS headers define the
+  # corresponding TCP_* constants. Where they are missing, setting any of them
+  # makes the driver reject the connection with einval. They are all available
+  # on Linux; elsewhere we drop them, so that the database connection still
+  # works, and tell the user.
+  defp tcp_opts(opts) do
+    tcp_opts = [
+      keepalive_idle: get_env(opts, :db_tcp_keepalive_idle),
+      keepalive_interval: get_env(opts, :db_tcp_keepalive_interval),
+      keepalive_count: get_env(opts, :db_tcp_keepalive_count),
+      user_timeout: get_env(opts, :db_tcp_user_timeout)
+    ]
+
+    configured = Enum.reject(tcp_opts, fn {_key, val} -> is_nil(val) end)
+
+    cond do
+      configured == [] ->
+        []
+
+      :os.type() == {:unix, :linux} ->
+        configured
+
+      true ->
+        Logger.warning(
+          "Ignoring database TCP keepalive/user timeout settings " <>
+            "#{inspect(Keyword.keys(configured))}: they are only supported on Linux."
+        )
+
+        []
+    end
+  end
+
   defp get_env(opts, key) do
     get_env(opts, key, key)
   end

packages/sync-service/lib/electric/config.ex

@@ -45,6 +45,13 @@ defmodule Electric.Config do
     ## Database
     provided_database_id: "single_stack",
     db_pool_size: 20,
+    # TCP-level liveness detection for database connections. All nil by
+    # default, leaving the OS defaults in place. See
+    # Electric.Connection.Manager.ConnectionResolver for what these do.
+    db_tcp_keepalive_idle: nil,
+    db_tcp_keepalive_interval: nil,
+    db_tcp_keepalive_count: nil,
+    db_tcp_user_timeout: nil,
     replication_stream_id: "default",
     replication_slot_temporary?: false,
     replication_slot_temporary_random_name?: false,

packages/sync-service/lib/electric/connection/manager/connection_resolver.ex

@@ -51,7 +51,10 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
       connection_mod =
       Keyword.get(opts, :connection_mod, {Postgrex.SimpleConnection, :start_link, []})
 
-    {:ok, %{connection_mod: connection_mod, stack_id: stack_id}, {:continue, :notify_ready}}
+    tcp_opts = Keyword.get(opts, :tcp_opts, [])
+
+    {:ok, %{connection_mod: connection_mod, stack_id: stack_id, tcp_opts: tcp_opts},
+     {:continue, :notify_ready}}
   end
 
   @impl GenServer
@@ -64,7 +67,7 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
   @impl GenServer
   def handle_call({:validate, connection}, _from, state) do
     # convert to postgrex style for return to conn.manager
-    connection = populate_connection_opts(connection)
+    connection = populate_connection_opts(connection, state.tcp_opts)
 
     result = attempt_connection({:cont, connection}, state)
 
@@ -100,7 +103,7 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
 
       {:error, error} ->
         error
-        |> mutate_based_on_error(conn_opts)
+        |> mutate_based_on_error(conn_opts, state.tcp_opts)
         |> attempt_connection(state)
     end
   end
@@ -109,8 +112,8 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
     {:error, error}
   end
 
-  defp populate_connection_opts(conn_opts) do
-    conn_opts |> populate_ssl_opts() |> populate_tcp_opts()
+  defp populate_connection_opts(conn_opts, tcp_opts) do
+    conn_opts |> populate_ssl_opts() |> populate_tcp_opts(tcp_opts)
   end
 
   defp populate_ssl_opts(connection_opts) do
@@ -172,36 +175,87 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
     ]
   end
 
-  defp populate_tcp_opts(connection_opts) do
-    tcp_opts =
+  defp populate_tcp_opts(connection_opts, tcp_opts) do
+    inet_opts =
       if connection_opts[:ipv6] do
         [:inet6]
       else
         []
       end
 
-    Keyword.put(connection_opts, :socket_options, tcp_opts)
+    Keyword.put(
+      connection_opts,
+      :socket_options,
+      inet_opts ++ tcp_liveness_opts(tcp_opts)
+    )
   end
 
-  defp mutate_based_on_error(%Postgrex.Error{message: "ssl not available"} = error, conn_opts) do
+  # Options for configuring TCP keepalives and TCP user timeout.
+  #
+  # SO_KEEPALIVE makes the kernel probe an idle connection, and TCP_USER_TIMEOUT
+  # caps how long unacknowledged data may stay outstanding before the connection
+  # is dropped -- the latter also bounds detection while data *is* being sent,
+  # which keepalive alone does not.
+  #
+  # Everything here is opt-in: with no configuration we emit no options and
+  # inherit the OS defaults. Platform support is checked upstream, in
+  # Electric.Application, before these options reach us.
+  #
+  # Time values are in milliseconds, as produced by parse_human_readable_time!.
+  # TCP_KEEPIDLE and TCP_KEEPINTVL are expressed in seconds by the kernel, while
+  # TCP_USER_TIMEOUT takes milliseconds.
+  @doc false
+  def tcp_liveness_opts(config) do
+    keepalive_idle = Keyword.get(config, :keepalive_idle)
+    keepalive_interval = Keyword.get(config, :keepalive_interval)
+    keepalive_count = Keyword.get(config, :keepalive_count)
+    user_timeout = Keyword.get(config, :user_timeout)
+
+    keepalive_opt =
+      if is_nil(keepalive_idle) and is_nil(keepalive_interval) and is_nil(keepalive_count) do
+        []
+      else
+        [{:keepalive, true}]
+      end
+
+    keepalive_opt ++
+      tcp_opt(:keepidle, ms_to_sec(keepalive_idle)) ++
+      tcp_opt(:keepintvl, ms_to_sec(keepalive_interval)) ++
+      tcp_opt(:keepcnt, keepalive_count) ++
+      tcp_opt(:user_timeout, user_timeout)
+  end
+
+  defp tcp_opt(_opt, nil), do: []
+  defp tcp_opt(opt, value) when is_integer(value), do: [{opt, value}]
+
+  defp ms_to_sec(nil), do: nil
+  defp ms_to_sec(ms) when is_integer(ms), do: max(div(ms, 1000), 1)
+
+  defp mutate_based_on_error(
+         %Postgrex.Error{message: "ssl not available"} = error,
+         conn_opts,
+         _tcp_opts
+       ) do
     maybe_fallback_to_no_ssl(conn_opts, error)
   end
 
   defp mutate_based_on_error(
          %DBConnection.ConnectionError{message: "ssl connect: closed"} = error,
-         conn_opts
+         conn_opts,
+         _tcp_opts
        ) do
     maybe_fallback_to_no_ssl(conn_opts, error)
   end
 
   defp mutate_based_on_error(
          %DBConnection.ConnectionError{severity: :error} = error,
-         conn_opts
+         conn_opts,
+         tcp_opts
        ) do
-    maybe_fallback_to_ipv4(error, conn_opts)
+    maybe_fallback_to_ipv4(error, conn_opts, tcp_opts)
   end
 
-  defp mutate_based_on_error(error, _conn_opts) do
+  defp mutate_based_on_error(error, _conn_opts, _tcp_opts) do
     {:halt, error}
   end
 
@@ -225,7 +279,8 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
 
   defp maybe_fallback_to_ipv4(
          %DBConnection.ConnectionError{message: message, severity: :error} = error,
-         conn_opts
+         conn_opts,
+         tcp_opts
        ) do
     # If network is unreachable, IPv6 is not enabled on the machine
     # If domain cannot be resolved, assume there is no AAAA record for it
@@ -239,7 +294,7 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
         "Database connection failed to find valid IPv6 address for #{conn_opts[:hostname]} - falling back to IPv4"
       )
 
-      {:cont, conn_opts |> Keyword.put(:ipv6, false) |> populate_tcp_opts()}
+      {:cont, conn_opts |> Keyword.put(:ipv6, false) |> populate_tcp_opts(tcp_opts)}
     else
       {:halt, error}
     end

packages/sync-service/lib/electric/connection/manager/supervisor.ex

@@ -20,7 +20,8 @@ defmodule Electric.Connection.Manager.Supervisor do
 
     children = [
       {Electric.Connection.Manager, opts},
-      {Electric.Connection.Manager.ConnectionResolver, stack_id: opts[:stack_id]}
+      {Electric.Connection.Manager.ConnectionResolver,
+       stack_id: opts[:stack_id], tcp_opts: Keyword.get(opts, :tcp_opts, [])}
     ]
 
     # Electric.Connection.Manager is a permanent child of the supervisor, so when it dies, the

packages/sync-service/lib/electric/stack_supervisor.ex

@@ -102,6 +102,18 @@ defmodule Electric.StackSupervisor do
                    doc:
                      "will be passed on to the Postgrex connection pool. See `t:Postgrex.start_option()`, apart from the connection options."
                  ],
+                 tcp_opts: [
+                   type: :keyword_list,
+                   default: [],
+                   doc:
+                     "TCP-level liveness settings applied to every database connection socket. Any option left unset keeps the OS default.",
+                   keys: [
+                     keepalive_idle: [type: {:or, [:pos_integer, nil]}, default: nil],
+                     keepalive_interval: [type: {:or, [:pos_integer, nil]}, default: nil],
+                     keepalive_count: [type: {:or, [:pos_integer, nil]}, default: nil],
+                     user_timeout: [type: {:or, [:pos_integer, nil]}, default: nil]
+                   ]
+                 ],
                  storage: [type: :mod_arg, required: true],
                  storage_dir: [type: :string, required: true],
                  chunk_bytes_threshold: [
@@ -383,6 +395,7 @@ defmodule Electric.StackSupervisor do
           handle_event: {Electric.Replication.ShapeLogCollector, :handle_event_async, [stack_id]}
         ] ++ config.replication_opts,
       pool_opts: [types: PgInterop.Postgrex.Types] ++ config.pool_opts,
+      tcp_opts: config.tcp_opts,
       timeline_opts: [
         stack_id: stack_id,
         persistent_kv: config.persistent_kv