fix: Fix REPLICATION_STREAM_ID parsing (#1901)

electric-sql · Oct 29, 2024 · 1cf8bf9 · 1cf8bf9
1 parent d43c375
commit 1cf8bf9
Show file tree

Hide file tree

Showing 5 changed files with 26 additions and 7 deletions.
diff --git a/.changeset/big-shoes-care.md b/.changeset/big-shoes-care.md
@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Fix `REPLICATION_STREAM_ID` not being able to be set because of incorrect parsing
diff --git a/integration-tests/scripts/electric_dev.sh b/integration-tests/scripts/electric_dev.sh
@@ -5,4 +5,4 @@ set -e
 SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &> /dev/null && pwd)
 
 cd "$SCRIPT_DIR"/../../packages/sync-service
-STORAGE_DIR="$SCRIPT_DIR/../_storage" iex -S mix
+STORAGE_DIR="$SCRIPT_DIR/../_storage" REPLICATION_STREAM_ID=integration iex -S mix
diff --git a/integration-tests/tests/invalidated-replication-slot.lux b/integration-tests/tests/invalidated-replication-slot.lux
@@ -7,7 +7,7 @@
 [my invalidated_slot_error=
   """
   [error] GenServer Electric.Connection.Manager terminating
-  ** (Postgrex.Error) ERROR 55000 (object_not_in_prerequisite_state) cannot read from logical replication slot "electric_slot_default"
+  ** (Postgrex.Error) ERROR 55000 (object_not_in_prerequisite_state) cannot read from logical replication slot "electric_slot_integration"
 
   This slot has been invalidated because it exceeded the maximum reserved size.
   """]
@@ -34,7 +34,7 @@
 
 ## Confirm slot invalidation in Postgres.
 [shell pg]
-  ?invalidating slot "electric_slot_default" because its restart_lsn [\d\w]+/[\d\w]+ exceeds max_slot_wal_keep_size
+  ?invalidating slot "electric_slot_integration" because its restart_lsn [\d\w]+/[\d\w]+ exceeds max_slot_wal_keep_size
 
 ## Observe the fatal connection error.
 [shell electric]

diff --git a/integration-tests/tests/rolling-deploy.lux b/integration-tests/tests/rolling-deploy.lux
@@ -13,8 +13,8 @@
 [invoke setup_electric_shell "electric_1" "3000" ""]
 
 [shell electric_1]
-  ??[info] Acquiring lock from postgres with name electric_slot_default
-  ??[info] Lock acquired from postgres with name electric_slot_default
+  ??[info] Acquiring lock from postgres with name electric_slot_integration
+  ??[info] Lock acquired from postgres with name electric_slot_integration
   ??[info] Starting replication from postgres
 
 # First service should be health and active
@@ -29,7 +29,7 @@
 ## in the second electric
 [shell electric_2]
   -Lock acquired from postgres|Starting replication from postgres|$fail_pattern
-  ??[info] Acquiring lock from postgres with name electric_slot_default
+  ??[info] Acquiring lock from postgres with name electric_slot_integration
   [sleep 2]
 
 
@@ -50,7 +50,7 @@
 ## Lock should now be acquired and replication starting
 [shell electric_2]
   -$fail_pattern
-  ??[info] Lock acquired from postgres with name electric_slot_default
+  ??[info] Lock acquired from postgres with name electric_slot_integration
   ??[info] Starting replication from postgres
 
 # Second service is now healthy and active

diff --git a/packages/sync-service/lib/electric/postgres/identifiers.ex b/packages/sync-service/lib/electric/postgres/identifiers.ex
@@ -86,6 +86,20 @@ defmodule Electric.Postgres.Identifiers do
       else: {:ok, unescape_quotes(ident)}
   end
 
+  @doc """
+  Parse an unquoted PostgreSQL identifier, downcasing characters and failing if any
+  special characters are present
+
+  ## Examples
+      iex> Electric.Postgres.Identifiers.parse_unquoted_identifier("FooBar")
+      {:ok, "foobar"}
+
+      iex> Electric.Postgres.Identifiers.parse_unquoted_identifier("foob@r")
+      {:error, ~S|Invalid unquoted identifier contains special characters: foob@r|}
+  """
+  @spec parse(binary(), boolean(), boolean()) :: {:ok, binary()} | {:error, term()}
+  def parse_unquoted_identifier(ident, truncate \\ false, single_byte_encoding \\ false)
+
   def parse_unquoted_identifier("", _, _), do: parse_quoted_identifier("")
 
   def parse_unquoted_identifier(ident, truncate, single_byte_encoding) do