Fix tests in CI

.github/workflows/main.yml

@@ -105,14 +105,13 @@ jobs:
             elixir: "1.15.4"
             test_only_cassandra: true
             cassandra_version: "4.1"
-            cassandra_native_protocol: "v4"
+            cassandra_native_protocol: "v5"
 
     env:
       CASSANDRA_VERSION: ${{ matrix.server_versions.cassandra }}
       SCYLLA_VERSION: ${{ matrix.server_versions.scylla }}
       CASSANDRA_NATIVE_PROTOCOL: ${{ matrix.cassandra_native_protocol }}
       LOG_LEVEL: debug
-      XANDRA_DEBUG: true
 
     steps:
       - name: Clone the repository
@@ -162,11 +161,11 @@ jobs:
       # TODO: eventually figure out why we can't run encryption tests on CI.
       - name: Run tests for Cassandra and Scylla
         if: ${{ !matrix.test_only_cassandra }}
-        run: mix test.all --trace --exclude encryption --exclude ccm
+        run: mix test.all --trace --exclude encryption --exclude ccm --exclude toxiproxy
 
       - name: Run tests for Cassandra only
         if: ${{ matrix.test_only_cassandra }}
-        run: mix test --trace --exclude encryption --exclude ccm
+        run: mix test --trace --exclude encryption --exclude ccm --exclude toxiproxy
 
       - name: Dump Docker logs on failure
         uses: jwalton/gh-docker-logs@v1

docker-compose.yml

@@ -90,7 +90,7 @@ services:
         max-size: 50m
 
   toxiproxy:
-    image: ghcr.io/shopify/toxiproxy
+    image: ghcr.io/shopify/toxiproxy:2.6.0
     command: -config=/config/toxiproxy.json -host 0.0.0.0
     volumes:
       - ./test/toxiproxy_config.json:/config/toxiproxy.json

lib/xandra/cluster/pool.ex

@@ -350,12 +350,16 @@ defmodule Xandra.Cluster.Pool do
   end
 
   # Handle the control connection shutting itself down.
+  # If we don't have a control connection PID (yet?) and some other PID shuts down
+  # with reason :shutdown, we assume it's the control connection we go on to
+  # starting the next control connection.
   def handle_event(
         :info,
         {:EXIT, control_connection_pid, {:shutdown, _reason}},
         _state,
-        %__MODULE__{control_connection: control_connection_pid}
-      ) do
+        %__MODULE__{control_connection: data_pid}
+      )
+      when data_pid == control_connection_pid or is_nil(data_pid) do
     {:keep_state_and_data, {:next_event, :internal, :start_control_connection}}
   end
 

lib/xandra/connection.ex

@@ -132,10 +132,7 @@ defmodule Xandra.Connection do
       |> protocol_module.encode_request(prepared)
       |> Frame.encode(protocol_module)
 
-    case Transport.send(transport, payload) do
-      :ok -> :ok
-      {:error, reason} -> {:error, ConnectionError.new("prepare", reason)}
-    end
+    Transport.send(transport, payload)
   end
 
   @spec execute(:gen_statem.server_ref(), Batch.t(), nil, keyword()) ::

lib/xandra/telemetry.ex

@@ -145,6 +145,9 @@ defmodule Xandra.Telemetry do
       [:pool, :restarted] ->
         Logger.debug("Pool restarted", logger_meta)
 
+      [:pool, :stopped] ->
+        Logger.debug("Pool stopped", logger_meta)
+
       [:discovered_peers] ->
         Logger.debug("Discovered peers: #{inspect(measurements.peers)}", logger_meta)
     end

test/integration/warning_test.exs

@@ -50,7 +50,7 @@ defmodule WarningTest do
     assert_receive {[:xandra, :server_warnings], ^ref, measurements, metadata}
     assert %{warnings: [warning]} = measurements
     assert warning =~ "Unlogged batch covering 11 partitions"
-    assert metadata.address == ~c"127.0.0.1"
+    assert metadata.address == "127.0.0.1"
     assert metadata.port == 9052
     assert metadata.current_keyspace == keyspace
     assert inspect(metadata.query) == inspect(batch)

test/toxiproxy_config.json

@@ -1,13 +1,13 @@
 [
     {
         "name": "xandra_test_cassandra",
-        "listen": "0.0.0.0:9042",
+        "listen": "[::]:9042",
         "upstream": "cassandra:9042",
         "enabled": true
     },
     {
         "name": "xandra_test_cassandra_sliced",
-        "listen": "0.0.0.0:9043",
+        "listen": "[::]:9043",
         "upstream": "cassandra:9042",
         "enabled": true
     }

test/xandra/cluster_test.exs

@@ -14,7 +14,14 @@ defmodule Xandra.ClusterTest do
   defmodule PoolMock do
     use Supervisor
 
-    def start_link(opts), do: Supervisor.start_link(__MODULE__, Map.new(opts))
+    def start_link(opts) do
+      map_opts =
+        opts
+        |> Map.new()
+        |> Map.update(:connection_options, %{}, &Map.new/1)
+
+      Supervisor.start_link(__MODULE__, map_opts)
+    end
 
     @impl true
     def init(opts) do
@@ -137,9 +144,17 @@ defmodule Xandra.ClusterTest do
   end
 
   describe "start_link/1" do
-    test "doesn't fail to start if the control connection fails to connect", %{base_options: opts} do
+    @tag telemetry_events: [[:xandra, :cluster, :control_connection, :failed_to_connect]]
+    test "doesn't fail to start if the control connection fails to connect",
+         %{base_options: opts, telemetry_ref: telemetry_ref} do
       opts = Keyword.merge(opts, nodes: ["127.0.0.1:8092"], sync_connect: false)
       pid = start_link_supervised!({Cluster, opts})
+
+      assert_telemetry [:control_connection, :failed_to_connect], %{
+        cluster_pid: ^pid,
+        reason: :econnrefused
+      }
+
       :sys.get_state(pid)
     end
 
@@ -305,10 +320,9 @@ defmodule Xandra.ClusterTest do
       pid = start_link_supervised!({Cluster, opts})
 
       assert_receive {^test_ref, PoolMock, :init_called,
-                      %{
-                        pool_size: 1,
-                        connection_options: [nodes: ["127.0.0.1:9052"], cluster_pid: ^pid]
-                      }}
+                      %{pool_size: 1, connection_options: %{cluster_pid: ^pid} = conn_opts}}
+
+      assert conn_opts[:nodes] == ["127.0.0.1:#{@port}"]
 
       remote_host = %Host{address: {198, 10, 0, 1}, port: @port, data_center: "remote_dc"}
       local_host1 = %Host{address: {198, 0, 0, 1}, port: @port, data_center: "local_dc"}
@@ -330,13 +344,13 @@ defmodule Xandra.ClusterTest do
       assert_receive {^test_ref, PoolMock, :init_called,
                       %{
                         pool_size: 1,
-                        connection_options: [nodes: ["198.0.0.1:9052"], cluster_pid: ^pid]
+                        connection_options: %{nodes: ["198.0.0.1:9052"], cluster_pid: ^pid}
                       }}
 
       assert_receive {^test_ref, PoolMock, :init_called,
                       %{
                         pool_size: 1,
-                        connection_options: [nodes: ["198.0.0.2:9052"], cluster_pid: ^pid]
+                        connection_options: %{nodes: ["198.0.0.2:9052"], cluster_pid: ^pid}
                       }}
 
       refute_receive {[:xandra, :cluster, :pool, :started], ^telemetry_ref, %{},
@@ -911,6 +925,13 @@ defmodule Xandra.ClusterTest do
     end
   end
 
+  @tag :tmp
+  @tag :toxiproxy
+  test "[tmp] Toxiproxy goes through", %{base_options: opts} do
+    opts = Keyword.merge(opts, sync_connect: 1000, nodes: ["127.0.0.1:19052"])
+    assert {:ok, _pid} = start_supervised({Cluster, opts})
+  end
+
   defp get_state(cluster) do
     assert {_state, data} = :sys.get_state(cluster)
     data

test/xandra_test.exs

@@ -62,17 +62,6 @@ defmodule XandraTest do
                Xandra.execute(conn, "USE some_keyspace")
     end
 
-    test "supports the :connect_timeout option", %{start_options: start_options} do
-      assert {:ok, conn} =
-               start_supervised(
-                 {Xandra, [connect_timeout: 0, backoff_type: :stop] ++ start_options}
-               )
-
-      ref = Process.monitor(conn)
-      assert_receive {:DOWN, ^ref, _, _, reason}
-      assert reason == :timeout
-    end
-
     test "supports the :name option as an atom", %{start_options: start_options} do
       assert {:ok, conn} = start_supervised({Xandra, [name: :my_test_conn] ++ start_options})
       assert Process.whereis(:my_test_conn) == conn

test/xandra_toxiproxy_test.exs

@@ -35,10 +35,29 @@ defmodule XandraToxiproxyTest do
     conn = start_supervised!({Xandra, opts})
 
     ToxiproxyEx.get!(:xandra_test_cassandra)
-    |> ToxiproxyEx.toxic(:limit_data, bytes: 120)
+    |> ToxiproxyEx.toxic(:limit_data, bytes: 500)
     |> ToxiproxyEx.apply!(fn ->
       assert {:error, %ConnectionError{reason: :disconnected}} =
                Xandra.prepare(conn, "SELECT * FROM system.local")
     end)
   end
+
+  test "start_link/1 supports the :connect_timeout option", %{start_options: opts} do
+    opts =
+      Keyword.merge(opts,
+        connect_timeout: 0,
+        backoff_type: :stop,
+        nodes: ["127.0.0.1:19052"]
+      )
+
+    ToxiproxyEx.get!(:xandra_test_cassandra)
+    |> ToxiproxyEx.toxic(:timeout, timeout: 0)
+    |> ToxiproxyEx.apply!(fn ->
+      assert {:ok, conn} = start_supervised({Xandra, opts})
+
+      ref = Process.monitor(conn)
+      assert_receive {:DOWN, ^ref, _, _, reason}
+      assert reason == :timeout
+    end)
+  end
 end