Skip to content

Commit 6e917b9

Browse files
tsgalcoclaude
authored
Add TCP keepalive options for database connections (#4748)
This adds the following config options. These are set on the underlying TCP socket. * `ELECTRIC_DATABASE_TCP_KEEPALIVE_IDLE` - Idle time before probing. Enables `SO_KEEPALIVE` * `ELECTRIC_DATABASE_TCP_KEEPALIVE_INTERVAL` - Gap between probes. * `ELECTRIC_DATABASE_TCP_KEEPALIVE_COUNT` - Unanswered probes before drop. * `ELECTRIC_DATABASE_TCP_USER_TIMEOUT` - Max time data may stay unacknowledged. All unset by default, so behaviour is unchanged unless configured. Durations accept the usual 30s / 500ms forms. Sample config: ``` ELECTRIC_DATABASE_TCP_KEEPALIVE_IDLE=30s ELECTRIC_DATABASE_TCP_KEEPALIVE_INTERVAL=10s ELECTRIC_DATABASE_TCP_KEEPALIVE_COUNT=3 ELECTRIC_DATABASE_TCP_USER_TIMEOUT=60s ``` Implementation: `ConnectionResolver.populate_tcp_opts/1` now appends the options to `:socket_options`. `SO_KEEPALIVE` goes through :inet's portable `{:keepalive, true}`; the rest use :raw with the Linux `IPPROTO_TCP` option numbers, guarded on `:os.type()` so non-Linux developer machines skip them rather than failing with `einval`. Note: I've generated the PR with AI, because I don't know Elixir, but the change looks fairly straightforward to me, so I'm hoping it's all correct. The one thing that feels a bit risky is what happens on non-Linux kernels. The code attempts to disable this automatically in that case. <!-- codesmith:footer --> --- <a href="https://app.blacksmith.sh/electric-sql/codesmith/electric/pr/4748"><picture><source media="(prefers-color-scheme: dark)" srcset="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-dark-v2.svg"><source media="(prefers-color-scheme: light)" srcset="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-light-v2.svg"><img alt="View with [code]smith" src="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-dark-v2.svg"></picture></a> <a href="https://backend.blacksmith.sh/track/enable-autofix?expires=1788095454&installation_model_id=8736&pr_number=4748&repository=electric-sql%2Felectric&return_to=https%3A%2F%2Fgithub.com%2Felectric-sql%2Felectric%2Fpull%2F4748&signature=e8372d8dc66f8eb1d71e56d1b2c4fb88d8f8244348fd8c04b075ff5b4a4c57d3"><picture><source media="(prefers-color-scheme: dark)" srcset="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-light.svg"><img alt="Autofix with [code]smith" src="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-dark.svg"></picture></a> <sup>Need help on this PR? Tag <code>@codesmith-bot</code> with what you need. Autofix is disabled.</sup> <!-- codesmith:autofix:disabled --> <!-- /codesmith:footer --> --------- Co-authored-by: Oleksii Sholik <oleksii@sholik.dev> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent 77a22fa commit 6e917b9

10 files changed

Lines changed: 332 additions & 16 deletions

File tree

.changeset/db-tcp-keepalive.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@core/sync-service": patch
3+
---
4+
5+
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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -256,6 +256,25 @@ config :electric,
256256
prometheus_port: prometheus_port,
257257
live_dashboard_port: live_dashboard_port,
258258
db_pool_size: env!("ELECTRIC_DB_POOL_SIZE", :integer, nil),
259+
db_tcp_keepalive_idle:
260+
env!(
261+
"ELECTRIC_DATABASE_TCP_KEEPALIVE_IDLE",
262+
&Electric.Config.parse_human_readable_time!/1,
263+
nil
264+
),
265+
db_tcp_keepalive_interval:
266+
env!(
267+
"ELECTRIC_DATABASE_TCP_KEEPALIVE_INTERVAL",
268+
&Electric.Config.parse_human_readable_time!/1,
269+
nil
270+
),
271+
db_tcp_keepalive_count: env!("ELECTRIC_DATABASE_TCP_KEEPALIVE_COUNT", :integer, nil),
272+
db_tcp_user_timeout:
273+
env!(
274+
"ELECTRIC_DATABASE_TCP_USER_TIMEOUT",
275+
&Electric.Config.parse_human_readable_time!/1,
276+
nil
277+
),
259278
replication_stream_id: replication_stream_id,
260279
replication_slot_temporary?: env!("CLEANUP_REPLICATION_SLOTS_ON_SHUTDOWN", :boolean, nil),
261280
replication_slot_temporary_random_name?:

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

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,7 @@ defmodule Electric.Application do
140140
],
141141
pool_opts:
142142
get_env_lazy(opts, :pool_opts, fn -> [pool_size: get_env(opts, :db_pool_size)] end),
143+
tcp_opts: tcp_opts(opts),
143144
chunk_bytes_threshold: get_env(opts, :chunk_bytes_threshold),
144145
telemetry_opts: telemetry_opts([instance_id: instance_id] ++ opts),
145146
max_shapes: get_env(opts, :max_shapes),
@@ -240,6 +241,39 @@ defmodule Electric.Application do
240241
]
241242
end
242243

244+
# The keepidle, keepintvl, keepcnt and user_timeout socket options are only
245+
# compiled into Erlang's inet driver where the OS headers define the
246+
# corresponding TCP_* constants. Where they are missing, setting any of them
247+
# makes the driver reject the connection with einval. They are all available
248+
# on Linux; elsewhere we drop them, so that the database connection still
249+
# works, and tell the user.
250+
defp tcp_opts(opts) do
251+
tcp_opts = [
252+
keepalive_idle: get_env(opts, :db_tcp_keepalive_idle),
253+
keepalive_interval: get_env(opts, :db_tcp_keepalive_interval),
254+
keepalive_count: get_env(opts, :db_tcp_keepalive_count),
255+
user_timeout: get_env(opts, :db_tcp_user_timeout)
256+
]
257+
258+
configured = Enum.reject(tcp_opts, fn {_key, val} -> is_nil(val) end)
259+
260+
cond do
261+
configured == [] ->
262+
[]
263+
264+
:os.type() == {:unix, :linux} ->
265+
configured
266+
267+
true ->
268+
Logger.warning(
269+
"Ignoring database TCP keepalive/user timeout settings " <>
270+
"#{inspect(Keyword.keys(configured))}: they are only supported on Linux."
271+
)
272+
273+
[]
274+
end
275+
end
276+
243277
defp get_env(opts, key) do
244278
get_env(opts, key, key)
245279
end

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,13 @@ defmodule Electric.Config do
4343
## Database
4444
provided_database_id: "single_stack",
4545
db_pool_size: 20,
46+
# TCP-level liveness detection for database connections. All nil by
47+
# default, leaving the OS defaults in place. See
48+
# Electric.Connection.Manager.ConnectionResolver for what these do.
49+
db_tcp_keepalive_idle: nil,
50+
db_tcp_keepalive_interval: nil,
51+
db_tcp_keepalive_count: nil,
52+
db_tcp_user_timeout: nil,
4653
replication_stream_id: "default",
4754
replication_slot_temporary?: false,
4855
replication_slot_temporary_random_name?: false,

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

Lines changed: 70 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,10 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
5151
connection_mod =
5252
Keyword.get(opts, :connection_mod, {Postgrex.SimpleConnection, :start_link, []})
5353

54-
{:ok, %{connection_mod: connection_mod, stack_id: stack_id}, {:continue, :notify_ready}}
54+
tcp_opts = Keyword.get(opts, :tcp_opts, [])
55+
56+
{:ok, %{connection_mod: connection_mod, stack_id: stack_id, tcp_opts: tcp_opts},
57+
{:continue, :notify_ready}}
5558
end
5659

5760
@impl GenServer
@@ -64,7 +67,7 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
6467
@impl GenServer
6568
def handle_call({:validate, connection}, _from, state) do
6669
# convert to postgrex style for return to conn.manager
67-
connection = populate_connection_opts(connection)
70+
connection = populate_connection_opts(connection, state.tcp_opts)
6871

6972
result = attempt_connection({:cont, connection}, state)
7073

@@ -100,7 +103,7 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
100103

101104
{:error, error} ->
102105
error
103-
|> mutate_based_on_error(conn_opts)
106+
|> mutate_based_on_error(conn_opts, state.tcp_opts)
104107
|> attempt_connection(state)
105108
end
106109
end
@@ -109,8 +112,8 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
109112
{:error, error}
110113
end
111114

112-
defp populate_connection_opts(conn_opts) do
113-
conn_opts |> populate_ssl_opts() |> populate_tcp_opts()
115+
defp populate_connection_opts(conn_opts, tcp_opts) do
116+
conn_opts |> populate_ssl_opts() |> populate_tcp_opts(tcp_opts)
114117
end
115118

116119
defp populate_ssl_opts(connection_opts) do
@@ -172,36 +175,87 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
172175
]
173176
end
174177

175-
defp populate_tcp_opts(connection_opts) do
176-
tcp_opts =
178+
defp populate_tcp_opts(connection_opts, tcp_opts) do
179+
inet_opts =
177180
if connection_opts[:ipv6] do
178181
[:inet6]
179182
else
180183
[]
181184
end
182185

183-
Keyword.put(connection_opts, :socket_options, tcp_opts)
186+
Keyword.put(
187+
connection_opts,
188+
:socket_options,
189+
inet_opts ++ tcp_liveness_opts(tcp_opts)
190+
)
184191
end
185192

186-
defp mutate_based_on_error(%Postgrex.Error{message: "ssl not available"} = error, conn_opts) do
193+
# Options for configuring TCP keepalives and TCP user timeout.
194+
#
195+
# SO_KEEPALIVE makes the kernel probe an idle connection, and TCP_USER_TIMEOUT
196+
# caps how long unacknowledged data may stay outstanding before the connection
197+
# is dropped -- the latter also bounds detection while data *is* being sent,
198+
# which keepalive alone does not.
199+
#
200+
# Everything here is opt-in: with no configuration we emit no options and
201+
# inherit the OS defaults. Platform support is checked upstream, in
202+
# Electric.Application, before these options reach us.
203+
#
204+
# Time values are in milliseconds, as produced by parse_human_readable_time!.
205+
# TCP_KEEPIDLE and TCP_KEEPINTVL are expressed in seconds by the kernel, while
206+
# TCP_USER_TIMEOUT takes milliseconds.
207+
@doc false
208+
def tcp_liveness_opts(config) do
209+
keepalive_idle = Keyword.get(config, :keepalive_idle)
210+
keepalive_interval = Keyword.get(config, :keepalive_interval)
211+
keepalive_count = Keyword.get(config, :keepalive_count)
212+
user_timeout = Keyword.get(config, :user_timeout)
213+
214+
keepalive_opt =
215+
if is_nil(keepalive_idle) and is_nil(keepalive_interval) and is_nil(keepalive_count) do
216+
[]
217+
else
218+
[{:keepalive, true}]
219+
end
220+
221+
keepalive_opt ++
222+
tcp_opt(:keepidle, ms_to_sec(keepalive_idle)) ++
223+
tcp_opt(:keepintvl, ms_to_sec(keepalive_interval)) ++
224+
tcp_opt(:keepcnt, keepalive_count) ++
225+
tcp_opt(:user_timeout, user_timeout)
226+
end
227+
228+
defp tcp_opt(_opt, nil), do: []
229+
defp tcp_opt(opt, value) when is_integer(value), do: [{opt, value}]
230+
231+
defp ms_to_sec(nil), do: nil
232+
defp ms_to_sec(ms) when is_integer(ms), do: max(div(ms, 1000), 1)
233+
234+
defp mutate_based_on_error(
235+
%Postgrex.Error{message: "ssl not available"} = error,
236+
conn_opts,
237+
_tcp_opts
238+
) do
187239
maybe_fallback_to_no_ssl(conn_opts, error)
188240
end
189241

190242
defp mutate_based_on_error(
191243
%DBConnection.ConnectionError{message: "ssl connect: closed"} = error,
192-
conn_opts
244+
conn_opts,
245+
_tcp_opts
193246
) do
194247
maybe_fallback_to_no_ssl(conn_opts, error)
195248
end
196249

197250
defp mutate_based_on_error(
198251
%DBConnection.ConnectionError{severity: :error} = error,
199-
conn_opts
252+
conn_opts,
253+
tcp_opts
200254
) do
201-
maybe_fallback_to_ipv4(error, conn_opts)
255+
maybe_fallback_to_ipv4(error, conn_opts, tcp_opts)
202256
end
203257

204-
defp mutate_based_on_error(error, _conn_opts) do
258+
defp mutate_based_on_error(error, _conn_opts, _tcp_opts) do
205259
{:halt, error}
206260
end
207261

@@ -225,7 +279,8 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
225279

226280
defp maybe_fallback_to_ipv4(
227281
%DBConnection.ConnectionError{message: message, severity: :error} = error,
228-
conn_opts
282+
conn_opts,
283+
tcp_opts
229284
) do
230285
# If network is unreachable, IPv6 is not enabled on the machine
231286
# If domain cannot be resolved, assume there is no AAAA record for it
@@ -239,7 +294,7 @@ defmodule Electric.Connection.Manager.ConnectionResolver do
239294
"Database connection failed to find valid IPv6 address for #{conn_opts[:hostname]} - falling back to IPv4"
240295
)
241296

242-
{:cont, conn_opts |> Keyword.put(:ipv6, false) |> populate_tcp_opts()}
297+
{:cont, conn_opts |> Keyword.put(:ipv6, false) |> populate_tcp_opts(tcp_opts)}
243298
else
244299
{:halt, error}
245300
end

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,8 @@ defmodule Electric.Connection.Manager.Supervisor do
2020

2121
children = [
2222
{Electric.Connection.Manager, opts},
23-
{Electric.Connection.Manager.ConnectionResolver, stack_id: opts[:stack_id]}
23+
{Electric.Connection.Manager.ConnectionResolver,
24+
stack_id: opts[:stack_id], tcp_opts: Keyword.get(opts, :tcp_opts, [])}
2425
]
2526

2627
# Electric.Connection.Manager is a permanent child of the supervisor, so when it dies, the

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,18 @@ defmodule Electric.StackSupervisor do
102102
doc:
103103
"will be passed on to the Postgrex connection pool. See `t:Postgrex.start_option()`, apart from the connection options."
104104
],
105+
tcp_opts: [
106+
type: :keyword_list,
107+
default: [],
108+
doc:
109+
"TCP-level liveness settings applied to every database connection socket. Any option left unset keeps the OS default.",
110+
keys: [
111+
keepalive_idle: [type: {:or, [:pos_integer, nil]}, default: nil],
112+
keepalive_interval: [type: {:or, [:pos_integer, nil]}, default: nil],
113+
keepalive_count: [type: {:or, [:pos_integer, nil]}, default: nil],
114+
user_timeout: [type: {:or, [:pos_integer, nil]}, default: nil]
115+
]
116+
],
105117
storage: [type: :mod_arg, required: true],
106118
storage_dir: [type: :string, required: true],
107119
chunk_bytes_threshold: [
@@ -383,6 +395,7 @@ defmodule Electric.StackSupervisor do
383395
handle_event: {Electric.Replication.ShapeLogCollector, :handle_event_async, [stack_id]}
384396
] ++ config.replication_opts,
385397
pool_opts: [types: PgInterop.Postgrex.Types] ++ config.pool_opts,
398+
tcp_opts: config.tcp_opts,
386399
timeline_opts: [
387400
stack_id: stack_id,
388401
persistent_kv: config.persistent_kv

0 commit comments

Comments
 (0)