From 5a5f4883ce268fd0255ab1b0c8a0c43bc3aaddd6 Mon Sep 17 00:00:00 2001 From: Oleksii Sholik Date: Tue, 3 Dec 2024 13:41:21 +0200 Subject: [PATCH 1/4] Fallback to using unencrypted DB connection when possible Fix #1792. --- .../lib/electric/connection/manager.ex | 79 ++++++++++++------- 1 file changed, 51 insertions(+), 28 deletions(-) diff --git a/packages/sync-service/lib/electric/connection/manager.ex b/packages/sync-service/lib/electric/connection/manager.ex index e9438ee033..343121e957 100644 --- a/packages/sync-service/lib/electric/connection/manager.ex +++ b/packages/sync-service/lib/electric/connection/manager.ex @@ -242,12 +242,16 @@ defmodule Electric.Connection.Manager do @impl true def handle_continue(:start_lock_connection, %State{lock_connection_pid: nil} = state) do - case Electric.Postgres.LockConnection.start_link( - connection_opts: state.connection_opts, - connection_manager: self(), - lock_name: Keyword.fetch!(state.replication_opts, :slot_name) - ) do - {:ok, lock_connection_pid} -> + opts = [ + connection_opts: state.connection_opts, + connection_manager: self(), + lock_name: Keyword.fetch!(state.replication_opts, :slot_name) + ] + + case start_lock_connection(opts) do + {:ok, pid, connection_opts} -> + state = %{state | lock_connection_pid: pid, connection_opts: connection_opts} + Electric.StackSupervisor.dispatch_stack_event( state.stack_events_registry, state.stack_id, @@ -255,7 +259,7 @@ defmodule Electric.Connection.Manager do ) Process.send_after(self(), :log_lock_connection_status, @lock_status_logging_interval) - {:noreply, %{state | lock_connection_pid: lock_connection_pid}} + {:noreply, state} {:error, reason} -> handle_connection_error(reason, state, "lock_connection") @@ -435,32 +439,27 @@ defmodule Electric.Connection.Manager do }} end - defp start_replication_client(opts) do - case Electric.Postgres.ReplicationClient.start_link(opts) do + defp start_lock_connection(opts) do + case Electric.Postgres.LockConnection.start_link(opts) do {:ok, pid} -> - {:ok, pid, Keyword.fetch!(opts, :connection_opts)} - - {:error, %Postgrex.Error{message: "ssl not available"}} = error -> - sslmode = get_in(opts, [:connection_opts, :sslmode]) + {:ok, pid, opts[:connection_opts]} - if sslmode == :require do - error - else - if not is_nil(sslmode) do - # Only log a warning when there's an explicit sslmode parameter in the database - # config, meaning the user has requested a certain sslmode. - Logger.warning( - "Failed to connect to the database using SSL. Trying again, using an unencrypted connection." - ) - end - - opts - |> Keyword.update!(:connection_opts, &Keyword.put(&1, :ssl, false)) - |> start_replication_client() + error -> + with {:ok, opts} <- maybe_fallback_to_no_ssl(error, opts) do + start_lock_connection(opts) end + end + end + + defp start_replication_client(opts) do + case Electric.Postgres.ReplicationClient.start_link(opts) do + {:ok, pid} -> + {:ok, pid, opts[:connection_opts]} error -> - error + with {:ok, opts} <- maybe_fallback_to_no_ssl(error, opts) do + start_replication_client(opts) + end end end @@ -478,6 +477,30 @@ defmodule Electric.Connection.Manager do ) end + defp maybe_fallback_to_no_ssl( + {:error, %Postgrex.Error{message: "ssl not available"}} = error, + opts + ) do + sslmode = get_in(opts, [:connection_opts, :sslmode]) + + if sslmode == :require do + error + else + if not is_nil(sslmode) do + # Only log a warning when there's an explicit sslmode parameter in the database + # config, meaning the user has requested a certain sslmode. + Logger.warning( + "Failed to connect to the database using SSL. Trying again, using an unencrypted connection." + ) + end + + opts = Keyword.update!(opts, :connection_opts, &Keyword.put(&1, :ssl, false)) + {:ok, opts} + end + end + + defp maybe_fallback_to_no_ssl(error, _opts), do: error + defp handle_connection_error( {:shutdown, {:failed_to_start_child, Electric.Postgres.ReplicationClient, error}}, state, From 99c7be62ec4a289dd6bb84eb03ac799d2c14a9e5 Mon Sep 17 00:00:00 2001 From: Oleksii Sholik Date: Tue, 3 Dec 2024 14:10:14 +0200 Subject: [PATCH 2/4] Remove the mention of ?sslmode=disable where it's not needed --- examples/gatekeeper-auth/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/gatekeeper-auth/README.md b/examples/gatekeeper-auth/README.md index ce7ca4d512..73bc1bf7a0 100644 --- a/examples/gatekeeper-auth/README.md +++ b/examples/gatekeeper-auth/README.md @@ -157,7 +157,7 @@ $ curl -sv --header "Authorization: Bearer ${AUTH_TOKEN}" \ Note that we got an empty response when successfully proxied through to Electric above because there are no `items` in the database. If you like, you can create some, e.g. using `psql`: ```console -$ psql "postgresql://postgres:password@localhost:54321/electric?sslmode=disable" +$ psql "postgresql://postgres:password@localhost:54321/electric" psql (16.4) Type "help" for help. From ae20ae13eedadfb03fb899ceafb3caa2eebb41fa Mon Sep 17 00:00:00 2001 From: Oleksii Sholik Date: Tue, 3 Dec 2024 14:10:55 +0200 Subject: [PATCH 3/4] Add changeset --- .changeset/stupid-weeks-look.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/stupid-weeks-look.md diff --git a/.changeset/stupid-weeks-look.md b/.changeset/stupid-weeks-look.md new file mode 100644 index 0000000000..a400d0f6e3 --- /dev/null +++ b/.changeset/stupid-weeks-look.md @@ -0,0 +1,5 @@ +--- +"@core/sync-service": patch +--- + +Restore the automatic fallback to unencrypted database connections when SSL isn't available. From 63676b94704496875a68a4581571296282742c3a Mon Sep 17 00:00:00 2001 From: Oleksii Sholik Date: Tue, 3 Dec 2024 14:07:00 +0200 Subject: [PATCH 4/4] Add an optional DATABASE_POOL_URL config When configured, the lock connection and the database pool will connect to this pooled URL. By default, when DATABASE_POOL_URL is not set, all connections are open using the direct database connection URL configured with DATABASE_URL. --- packages/sync-service/config/runtime.exs | 8 +++-- .../sync-service/lib/electric/application.ex | 20 +++++++++---- .../lib/electric/connection/manager.ex | 16 ++++++---- .../lib/electric/stack_supervisor.ex | 29 ++++++++++--------- .../test/support/component_setup.ex | 1 + 5 files changed, 47 insertions(+), 27 deletions(-) diff --git a/packages/sync-service/config/runtime.exs b/packages/sync-service/config/runtime.exs index b1b6b79f2b..0da658f066 100644 --- a/packages/sync-service/config/runtime.exs +++ b/packages/sync-service/config/runtime.exs @@ -105,11 +105,15 @@ database_ipv6_config = env!("ELECTRIC_DATABASE_USE_IPV6", :boolean, false) {:ok, database_url_config} = Electric.ConfigParser.parse_postgresql_uri(database_url) - connection_opts = database_url_config ++ [ipv6: database_ipv6_config] - config :electric, connection_opts: Electric.Utils.obfuscate_password(connection_opts) +if database_pool_url = env!("DATABASE_POOL_URL", :string, nil) do + {:ok, database_pool_url_config} = Electric.ConfigParser.parse_postgresql_uri(database_pool_url) + connection_opts = database_pool_url_config ++ [ipv6: database_ipv6_config] + config :electric, pool_connection_opts: Electric.Utils.obfuscate_password(connection_opts) +end + enable_integration_testing = env!("ELECTRIC_ENABLE_INTEGRATION_TESTING", :boolean, false) cache_max_age = env!("ELECTRIC_CACHE_MAX_AGE", :integer, 60) cache_stale_age = env!("ELECTRIC_CACHE_STALE_AGE", :integer, 60 * 5) diff --git a/packages/sync-service/lib/electric/application.ex b/packages/sync-service/lib/electric/application.ex index 44bf27b09a..9a83ec340f 100644 --- a/packages/sync-service/lib/electric/application.ex +++ b/packages/sync-service/lib/electric/application.ex @@ -36,6 +36,18 @@ defmodule Electric.Application do publication_name = "electric_publication_#{replication_stream_id}" slot_name = "electric_slot_#{replication_stream_id}" + replication_connection_opts = Application.fetch_env!(:electric, :connection_opts) + pool_connection_opts = Application.get_env(:electric, :pool_connection_opts) + + connection_opts = pool_connection_opts || replication_connection_opts + + replication_opts = [ + connection_opts: replication_connection_opts, + publication_name: publication_name, + slot_name: slot_name, + slot_temporary?: Application.fetch_env!(:electric, :replication_slot_temporary?) + ] + # The root application supervisor starts the core global processes, including the HTTP # server and the database connection manager. The latter is responsible for establishing # all needed connections to the database (acquiring the exclusive access lock, opening a @@ -53,13 +65,9 @@ defmodule Electric.Application do {Electric.StackSupervisor, stack_id: stack_id, stack_events_registry: Registry.StackEvents, - connection_opts: Application.fetch_env!(:electric, :connection_opts), persistent_kv: persistent_kv, - replication_opts: [ - publication_name: publication_name, - slot_name: slot_name, - slot_temporary?: Application.fetch_env!(:electric, :replication_slot_temporary?) - ], + connection_opts: connection_opts, + replication_opts: replication_opts, pool_opts: [pool_size: Application.fetch_env!(:electric, :db_pool_size)], storage: Application.fetch_env!(:electric, :storage), chunk_bytes_threshold: Application.fetch_env!(:electric, :chunk_bytes_threshold)}, diff --git a/packages/sync-service/lib/electric/connection/manager.ex b/packages/sync-service/lib/electric/connection/manager.ex index 343121e957..8f8c88601e 100644 --- a/packages/sync-service/lib/electric/connection/manager.ex +++ b/packages/sync-service/lib/electric/connection/manager.ex @@ -267,16 +267,20 @@ defmodule Electric.Connection.Manager do end def handle_continue(:start_replication_client, %State{replication_client_pid: nil} = state) do - opts = - state - |> Map.take([:stack_id, :replication_opts, :connection_opts]) - |> Map.to_list() - Logger.debug("Starting replication client for stack #{state.stack_id}") + {connection_opts, replication_opts} = Keyword.pop(state.replication_opts, :connection_opts) + + opts = [ + connection_opts: connection_opts, + replication_opts: replication_opts, + stack_id: state.stack_id + ] + case start_replication_client(opts) do {:ok, pid, connection_opts} -> - state = %{state | replication_client_pid: pid, connection_opts: connection_opts} + replication_opts = Keyword.put(replication_opts, :connection_opts, connection_opts) + state = %{state | replication_client_pid: pid, replication_opts: replication_opts} if is_nil(state.pool_pid) do # This is the case where Connection.Manager starts connections from the initial state. diff --git a/packages/sync-service/lib/electric/stack_supervisor.ex b/packages/sync-service/lib/electric/stack_supervisor.ex index b8c669e29e..4ad82933c3 100644 --- a/packages/sync-service/lib/electric/stack_supervisor.ex +++ b/packages/sync-service/lib/electric/stack_supervisor.ex @@ -29,28 +29,31 @@ defmodule Electric.StackSupervisor do """ use Supervisor, restart: :transient + @connection_opts_schema [ + type: :keyword_list, + required: true, + keys: [ + hostname: [type: :string, required: true], + port: [type: :integer, required: true], + database: [type: :string, required: true], + username: [type: :string, required: true], + password: [type: {:fun, 0}, required: true], + sslmode: [type: :atom, required: false], + ipv6: [type: :boolean, required: false] + ] + ] + @opts_schema NimbleOptions.new!( name: [type: :any, required: false], stack_id: [type: :string, required: true], persistent_kv: [type: :any, required: true], stack_events_registry: [type: :atom, required: true], - connection_opts: [ - type: :keyword_list, - required: true, - keys: [ - hostname: [type: :string, required: true], - port: [type: :integer, required: true], - database: [type: :string, required: true], - username: [type: :string, required: true], - password: [type: {:fun, 0}, required: true], - sslmode: [type: :atom, required: false], - ipv6: [type: :boolean, required: false] - ] - ], + connection_opts: @connection_opts_schema, replication_opts: [ type: :keyword_list, required: true, keys: [ + connection_opts: @connection_opts_schema, publication_name: [type: :string, required: true], slot_name: [type: :string, required: true], slot_temporary?: [type: :boolean, default: false], diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index 7dc47a28c4..b3ada65e35 100644 --- a/packages/sync-service/test/support/component_setup.ex +++ b/packages/sync-service/test/support/component_setup.ex @@ -163,6 +163,7 @@ defmodule Support.ComponentSetup do storage: storage, connection_opts: ctx.db_config, replication_opts: [ + connection_opts: ctx.db_config, slot_name: "electric_test_slot_#{:erlang.phash2(stack_id)}", publication_name: "electric_test_pub_#{:erlang.phash2(stack_id)}", try_creating_publication?: true,