Add ping command for PubSub connections #274

adrianke · 2024-11-01T13:54:38Z

This allows for periodic checking of the underlying TCP connection.

According to the Redis documentation, PING is one of very few commands that can be issued on a pubsub connection:

"Once the client enters the subscribed state it is not supposed to issue any other commands, except for additional SUBSCRIBE, SSUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, SUNSUBSCRIBE, PUNSUBSCRIBE, PING, RESET and QUIT commands"

I was not sure if the chosen flow is the best one, ie. sending a serialized PID as part of the payload and using that for sending the reply back to the waiting task.

whatyouhide

This is gonna be a useful addition. However, few notes:

We need to handle :ping calls when the gen_statem is in the disconnected state.
We need tests.
Instead of sending a PING command with a message, could we just keep a queue of pending PING commands, and for every PONG that the gen_statem receives we send it to the next caller waiting in the queue? Redis is single-threaded so it will reply to PINGs in order anyway.

whatyouhide · 2024-11-02T10:10:46Z

lib/redix/pubsub/connection.ex

+    {:ok, task_pid} =
+      Task.start(fn ->
+        res =
+          receive do
+            {:pong, ^random_string} -> :ok
+          after
+            5000 -> :error
+          end
+
+        :ok = :gen_statem.reply(from, res)
+      end)


We won't need the task here if we go with a queue of pending clients.

Thank you for the feedback.

Would the caller get an :error after 5000 ms without the task?

The flow to implement here is using gen_statem timeouts and processe aliases, see

Timeout specifies how long to wait for a response. If no response is received within the specified time, this function returns timeout. Assuming that the server executes on a node supporting aliases (introduced in OTP 24) the request will also be abandoned. That is, no response will be received after a time-out. Otherwise, a stray response might be received at a later time.

from the gen_statem docs. Basically you don't care about timeouts, you just reply to every pong and BEAM throws away the responses for procs that timed out already

This allows for periodic checking of the underlying TCP connection. According to the Redis documentation, PING is one of very few commands that can be issued on a pubsub connection: "Once the client enters the subscribed state it is not supposed to issue any other commands, except for additional SUBSCRIBE, SSUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, SUNSUBSCRIBE, PUNSUBSCRIBE, PING, RESET and QUIT commands"

adrianke · 2024-11-04T09:18:40Z

I've added some tests, and actually discovered that the reply format is different if you're subscribed to any channels or not.

I've also removed the task, and replaced it with a callback that is sent after 5000 ms.

EDIT:
And I've added a handler for the disconnected state, and removed the need for a subscription ref from the function signature.

adrianke · 2024-11-22T09:00:56Z

@whatyouhide Just a little reminder about this one. We've run my branch in production at work the last week and haven't encountered any troubles thus far.

whatyouhide

Going in the right direction, fantastic! Left a few more structural comments 🙃

whatyouhide · 2024-12-04T08:47:58Z

lib/redix/pubsub.ex

+  Upon a successful reply from the server, :ok will be returned. If no reply is received within 5 seconds, :error will be returned.
+


Suggested change

Upon a successful reply from the server, :ok will be returned. If no reply is received within 5 seconds, :error will be returned.

Upon a successful reply from the server, returns `:ok`. If no reply is received within the given timeout, returns `:error`.

whatyouhide · 2024-12-04T08:48:54Z

lib/redix/pubsub.ex

+
+  """
+  @spec ping(connection()) :: :ok | :error
+  def ping(conn) do


We need to make the timeout here configurable, and not hardcoded to 5 seconds. Do this:

def ping(conn, timeout \\ 5000) when is_integer(timeout) and timeout >= 0 do

and trickle that down to the implementation.

whatyouhide · 2024-12-04T08:49:20Z

lib/redix/pubsub/connection.ex

+    reply = :error
+    {:keep_state_and_data, {:reply, from, reply}}


Suggested change

reply = :error

{:keep_state_and_data, {:reply, from, reply}}

{:keep_state_and_data, {:reply, from, :error}}

whatyouhide · 2024-12-04T08:50:04Z

lib/redix/pubsub/connection.ex

@@ -230,6 +259,19 @@ defmodule Redix.PubSub.Connection do
    end
  end

+  def connected({:call, from}, :ping, data) do
+    random_string = :crypto.strong_rand_bytes(12) |> Base.encode64()


Let's do the same amount of bytes as in UUIDs.

Suggested change

random_string = :crypto.strong_rand_bytes(12) |> Base.encode64()

random_string = Base.encode64(:crypto.strong_rand_bytes(16))

whatyouhide · 2024-12-04T08:51:18Z

lib/redix/pubsub/connection.ex

+  def connected({:call, from}, :ping, data) do
+    random_string = :crypto.strong_rand_bytes(12) |> Base.encode64()
+    pipeline = [["PING", "ping:#{random_string}"]]
+    data.transport.send(data.socket, Enum.map(pipeline, &Protocol.pack/1))


Need to handle if this returns something that is not :ok here, and disconnect. We do it in other places too.

whatyouhide · 2024-12-04T08:52:08Z

lib/redix/pubsub/connection.ex

+    pipeline = [["PING", "ping:#{random_string}"]]
+    data.transport.send(data.socket, Enum.map(pipeline, &Protocol.pack/1))
+
+    Process.send_after(self(), {:ping_callback, random_string}, 5000)


Don't use send_after/3 here. Instead, use :gen_statem timeout actions. Return a timeout action from the return value of this function and handle the timeout instead of the message later on. Cancel the timeout if you get a ping with :cancel. The :gen_statem docs are the best resource for this 🙃

whatyouhide · 2024-12-04T08:52:32Z

lib/redix/pubsub/connection.ex

@@ -15,7 +15,8 @@ defmodule Redix.PubSub.Connection do
    :connected_address,
    :client_id,
    subscriptions: %{},
-    monitors: %{}
+    monitors: %{},
+    pings: []


I think you want a :queue here rather than a list, right?

whatyouhide reviewed Nov 2, 2024

View reviewed changes

adrianke force-pushed the main branch from 6ab1400 to dee490c Compare November 4, 2024 09:17

whatyouhide requested changes Dec 4, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add ping command for PubSub connections #274

Add ping command for PubSub connections #274

adrianke commented Nov 1, 2024

whatyouhide left a comment

whatyouhide Nov 2, 2024

adrianke Nov 4, 2024

whatyouhide Nov 4, 2024

adrianke commented Nov 4, 2024 •

edited

Loading

adrianke commented Nov 22, 2024

whatyouhide left a comment

whatyouhide Dec 4, 2024

whatyouhide Dec 4, 2024

whatyouhide Dec 4, 2024

whatyouhide Dec 4, 2024

whatyouhide Dec 4, 2024

whatyouhide Dec 4, 2024

whatyouhide Dec 4, 2024

		Upon a successful reply from the server, :ok will be returned. If no reply is received within 5 seconds, :error will be returned.

	reply = :error
	{:keep_state_and_data, {:reply, from, reply}}
	{:keep_state_and_data, {:reply, from, :error}}

	random_string = :crypto.strong_rand_bytes(12) \|> Base.encode64()
	random_string = Base.encode64(:crypto.strong_rand_bytes(16))

Add ping command for PubSub connections #274

Are you sure you want to change the base?

Add ping command for PubSub connections #274

Conversation

adrianke commented Nov 1, 2024

whatyouhide left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adrianke commented Nov 4, 2024 • edited Loading

adrianke commented Nov 22, 2024

whatyouhide left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adrianke commented Nov 4, 2024 •

edited

Loading