b77331e2a0e8e8f5efc35b36ee1258afe7460002

Committed 05 Jul 2026 09:40AM UTC coverage: 90.084% (-0.06%) from 90.14%

Build # b77331e2a0e8e8f5efc35b36ee1258afe7460002

Build Type

push

github

Committed by

whatyouhide

Commit Message

Track errors for each cluster node

Closes #339.

Coverage Stats

8 of 9 new or added lines in 1 file covered. (88.89%)

1 existing line in 1 file now uncovered.

1290 of 1432 relevant lines covered (90.08%)

2361.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.43

/test/support/cluster_fake_node.ex

defmodule Redix.Cluster.FakeNode do
  @moduledoc false

  # A scriptable fake RESP server for Redix Cluster tests.
  #
  # It lets the cluster test suite exercise topology and redirection edge cases
  # that a healthy Docker cluster won't emit on demand (issues #295, #306, #314,
  # #318, #319, #323, #328) without needing the Docker cluster at all. A node
  # listens on an ephemeral loopback port and replies to each parsed RESP command
  # with whatever its `handler` returns — a raw RESP binary. The listener and
  # every accepted connection are `spawn_link`ed to the calling test, so they die
  # with it.
  #
  # ## Single-shot node
  #
  #     node = FakeNode.start(fn ["GET", _] -> "$3\r\nbar\r\n" end)
  #     "#{node}"   # => "127.0.0.1:53123" — the node id (FakeNode implements String.Chars)
  #     node.id     # => same
  #
  # ## Nodes that reference each other (e.g. a redirect chain)
  #
  # Reserve the ports first so each node id is known before any handler is set,
  # then attach handlers:
  #
  #     a = FakeNode.reserve()
  #     b = FakeNode.reserve()
  #     FakeNode.serve(a, fn ["GET", _] -> "-ASK 0 #{b}\r\n" end)
  #     FakeNode.serve(b, fn ["GET", _] -> "$3\r\nbar\r\n" end)
  #
  # ## Wiring a real connection (redirection tests)
  #
  # `connect/2` registers a supervised `Redix` connection under the node id in a
  # cluster's registry, so redirection code that looks the node up by id resolves
  # it. `start_connected/3` does reserve + connect + serve in one step.
  #
  # ## Simulating an unreachable node
  #
  # A node starts `:up`. `set_status(node, :down)` makes it close accepted
  # connections immediately (so a topology fetch fails at the socket level);
  # flipping back to `:up` resumes serving. Pass `status: :down` to `reserve/1`
  # to start down.

  alias Redix.Cluster.FakeNode

  @accept_timeout 10_000
  @recv_timeout 10_000

  @enforce_keys [:id, :host, :port, :listen, :inet6, :status, :accepted]
  defstruct [:id, :host, :port, :listen, :inet6, :status, :accepted]

  @type t :: %__MODULE__{
          id: String.t(),
          host: String.t(),
          port: :inet.port_number(),
          listen: :gen_tcp.socket(),
          inet6: boolean(),
          status: pid(),
          accepted: pid()
        }

  @type handler :: ([binary()] -> iodata())

  ## Lifecycle

  # Reserve an ephemeral loopback port and node id without serving yet. Pass
  # `inet6: true` to bind the IPv6 loopback — the node id is then the unbracketed
  # "::1:port" form Redis uses in MOVED/ASK replies (issue #306). Pass
  # `status: :down` to start unreachable (see `set_status/2`).
  @spec reserve(keyword()) :: t()
  def reserve(opts \\ []) do
    inet6? = Keyword.get(opts, :inet6, false)
    {host, family_opts} = if inet6?, do: {"::1", [:inet6]}, else: {"127.0.0.1", []}

    {:ok, listen} =
      :gen_tcp.listen(
        0,
        [:binary, active: false, reuseaddr: true, packet: :raw] ++ family_opts
      )

    {:ok, port} = :inet.port(listen)
    {:ok, status} = Agent.start_link(fn -> Keyword.get(opts, :status, :up) end)
    {:ok, accepted} = Agent.start_link(fn -> 0 end)

    %FakeNode{
      id: "#{host}:#{port}",
      host: host,
      port: port,
      listen: listen,
      inet6: inet6?,
      status: status,
      accepted: accepted
    }
  end

  # Attach a handler to a reserved node and start accepting connections. The
  # handler maps a parsed command (a list of binaries) to a raw RESP reply.
  # Returns the node so it can be threaded through a pipe.
  @spec serve(t(), handler()) :: t()
  def serve(%FakeNode{} = node, handler) when is_function(handler, 1) do
    spawn_link(fn -> accept_loop(node, handler) end)
    node
  end

  # Reserve + serve in one step, for a node whose id no one needs in advance.
  @spec start(handler(), keyword()) :: t()
  def start(handler, opts \\ []) when is_function(handler, 1) do
    opts |> reserve() |> serve(handler)
  end

  # Register a supervised `Redix` connection to the node under its id in
  # `cluster`'s registry, so redirection code can resolve it by id. Must be
  # called from the test process (uses `start_supervised!`). Returns the node.
  @spec connect(t(), atom()) :: t()
  def connect(%FakeNode{} = node, cluster) do
    socket_opts = if node.inet6, do: [socket_opts: [:inet6]], else: []

    ExUnit.Callbacks.start_supervised!(
      {Redix,
       [
         host: node.host,
         port: node.port,
         sync_connect: true,
         name: {:via, Registry, {:"#{cluster}_registry", node.id}}
       ] ++ socket_opts},
      id: {:conn, node.id}
    )

    node
  end

  # reserve + connect + serve, for a fully wired node (redirection tests).
  @spec start_connected(atom(), handler(), keyword()) :: t()
  def start_connected(cluster, handler, opts \\ []) when is_function(handler, 1) do
    opts |> reserve() |> connect(cluster) |> serve(handler)
  end

  # Flip a node between serving (`:up`) and closing connections immediately
  # (`:down`), to simulate a node going down/up under a running test.
  @spec set_status(t(), :up | :down) :: :ok
  def set_status(%FakeNode{status: status}, value) when value in [:up, :down] do
    Agent.update(status, fn _ -> value end)
  end

  # Total number of TCP connections ever accepted (transient topology-fetch
  # sockets and managed Redix connections alike). Used to assert that no *new*
  # dial happens — e.g. a redirect resolving to an already-connected node instead
  # of opening a duplicate connection under a different address form (#340).
  @spec connections_accepted(t()) :: non_neg_integer()
  def connections_accepted(%FakeNode{accepted: accepted}) do
    Agent.get(accepted, & &1)
  end

  ## CLUSTER SLOTS encoding

  # Encode a `CLUSTER SLOTS` reply. Each range is `{start, stop, primary}` or
  # `{start, stop, primary, replicas}`, where a node is a `FakeNode`, a
  # "host:port" string, or a `{host_or_nil, port}` tuple. A `nil` host is encoded
  # as a RESP null bulk string, like Redis 7+ does with
  # "cluster-preferred-endpoint-type unknown-endpoint" (issue #328). Mirrors the
  # real RESP shape: an array of ranges, each `[start, stop, [host, port, id],
  # [replica_host, replica_port, id]...]`.
  @spec cluster_slots([tuple()]) :: binary()
  def cluster_slots(ranges) do
    body =
      Enum.map(ranges, fn range ->
        {start, stop, primary, replicas} =
          case range do
            {start, stop, primary} -> {start, stop, primary, []}
            {start, stop, primary, replicas} -> {start, stop, primary, replicas}
          end

        nodes = [primary | replicas]

        [
          "*#{2 + length(nodes)}\r\n",
          ":#{start}\r\n",
          ":#{stop}\r\n",
          Enum.map(nodes, &encode_node/1)
        ]
      end)

    IO.iodata_to_binary(["*#{length(ranges)}\r\n", body])
  end

  defp encode_node(node) do
    {host, port} = host_port(node)

    [
      "*3\r\n",
      encode_host(host),
      ":#{port}\r\n",
      "$40\r\n",
      String.duplicate("a", 40),
      "\r\n"
    ]
  end

  defp host_port(%FakeNode{host: host, port: port}), do: {host, port}
  defp host_port({_host_or_nil, _port} = host_port), do: host_port

  defp host_port(node_id) when is_binary(node_id) do
    [host, port] = String.split(node_id, ":")
    {host, String.to_integer(port)}
  end

  defp encode_host(nil), do: "$-1\r\n"
  defp encode_host(host), do: ["$#{byte_size(host)}\r\n", host, "\r\n"]

  ## Polling helper

  # Poll `fun` until it returns truthy or the timeout elapses (then flunk). Used
  # by the fake-node tests to wait for the cluster to converge after a topology
  # change, since reactive refreshes are async.
  @spec wait_until((-> as_boolean(term())), timeout()) :: :ok
  def wait_until(fun, timeout \\ 2_000)

  def wait_until(_fun, timeout) when timeout <= 0 do
    ExUnit.Assertions.flunk("condition not met in time")
  end

  def wait_until(fun, timeout) do
    if fun.() do
      :ok
    else
      Process.sleep(20)
      wait_until(fun, timeout - 20)
    end
  end

  ## Accept / serve loop

  # Accepts connections until the listen socket closes (when the owning test
  # exits) or no client shows up before the timeout. Some fake nodes get more
  # than one connection — e.g. the Manager's transient topology-fetch socket plus
  # the managed Redix connection. While the node is `:down`, accepted connections
  # are closed immediately to simulate an unreachable node.
  defp accept_loop(node, handler) do
    case :gen_tcp.accept(node.listen, @accept_timeout) do
      {:ok, socket} ->
        case Agent.get(node.status, & &1) do
          :up ->
            Agent.update(node.accepted, &(&1 + 1))
            spawn_link(fn -> loop(socket, handler, "") end)

          :down ->
            :gen_tcp.close(socket)
        end

        accept_loop(node, handler)

      {:error, _reason} ->
        :ok
    end
  end

  defp loop(socket, handler, buffer) do
    case :gen_tcp.recv(socket, 0, @recv_timeout) do
      {:ok, data} ->
        {commands, rest} = parse_commands(buffer <> data, [])
        Enum.each(commands, &:gen_tcp.send(socket, handler.(&1)))
        loop(socket, handler, rest)

      {:error, _reason} ->
        :ok
    end
  end

  ## Minimal RESP request parser

  # Pulls as many complete `*N\r\n$len\r\n...` commands as the buffer holds,
  # returning the parsed commands and any leftover bytes.
  defp parse_commands(buffer, acc) do
    case parse_command(buffer) do
      {:ok, command, rest} -> parse_commands(rest, [command | acc])
      :incomplete -> {Enum.reverse(acc), buffer}
    end
  end

  defp parse_command("*" <> rest) do
    with {:ok, count, rest} <- parse_int_line(rest) do
      parse_bulk_strings(count, rest, [])
    end
  end

  defp parse_command(_other), do: :incomplete

  defp parse_bulk_strings(0, rest, acc), do: {:ok, Enum.reverse(acc), rest}

  defp parse_bulk_strings(count, "$" <> rest, acc) do
    with {:ok, length, rest} <- parse_int_line(rest) do
      case rest do
        <<value::binary-size(^length), "\r\n", rest::binary>> ->
          parse_bulk_strings(count - 1, rest, [value | acc])

        _incomplete ->
          :incomplete
      end
    end
  end

  defp parse_bulk_strings(_count, _rest, _acc), do: :incomplete

  defp parse_int_line(binary) do
    case :binary.split(binary, "\r\n") do
      [int_string, rest] -> {:ok, String.to_integer(int_string), rest}
      [_no_crlf_yet] -> :incomplete
    end
  end
end

defimpl String.Chars, for: Redix.Cluster.FakeNode do
  def to_string(%Redix.Cluster.FakeNode{id: id}), do: id
end

1	defmodule Redix.Cluster.FakeNode do
2	@moduledoc false
3
4	# A scriptable fake RESP server for Redix Cluster tests.
5	#
6	# It lets the cluster test suite exercise topology and redirection edge cases
7	# that a healthy Docker cluster won't emit on demand (issues #295, #306, #314,
8	# #318, #319, #323, #328) without needing the Docker cluster at all. A node
9	# listens on an ephemeral loopback port and replies to each parsed RESP command
10	# with whatever its `handler` returns — a raw RESP binary. The listener and
11	# every accepted connection are `spawn_link`ed to the calling test, so they die
12	# with it.
13	#
14	# ## Single-shot node
15	#
16	# node = FakeNode.start(fn ["GET", _] -> "$3\r\nbar\r\n" end)
17	# "#{node}" # => "127.0.0.1:53123" — the node id (FakeNode implements String.Chars)
18	# node.id # => same
19	#
20	# ## Nodes that reference each other (e.g. a redirect chain)
21	#
22	# Reserve the ports first so each node id is known before any handler is set,
23	# then attach handlers:
24	#
25	# a = FakeNode.reserve()
26	# b = FakeNode.reserve()
27	# FakeNode.serve(a, fn ["GET", _] -> "-ASK 0 #{b}\r\n" end)
28	# FakeNode.serve(b, fn ["GET", _] -> "$3\r\nbar\r\n" end)
29	#
30	# ## Wiring a real connection (redirection tests)
31	#
32	# `connect/2` registers a supervised `Redix` connection under the node id in a
33	# cluster's registry, so redirection code that looks the node up by id resolves
34	# it. `start_connected/3` does reserve + connect + serve in one step.
35	#
36	# ## Simulating an unreachable node
37	#
38	# A node starts `:up`. `set_status(node, :down)` makes it close accepted
39	# connections immediately (so a topology fetch fails at the socket level);
40	# flipping back to `:up` resumes serving. Pass `status: :down` to `reserve/1`
41	# to start down.
42
43	alias Redix.Cluster.FakeNode
44
45	@accept_timeout 10_000
46	@recv_timeout 10_000
47
48	@enforce_keys [:id, :host, :port, :listen, :inet6, :status, :accepted]
49	defstruct [:id, :host, :port, :listen, :inet6, :status, :accepted]
50
51	@type t :: %__MODULE__{
52	id: String.t(),
53	host: String.t(),
54	port: :inet.port_number(),
55	listen: :gen_tcp.socket(),
56	inet6: boolean(),
57	status: pid(),
58	accepted: pid()
59	}
60
61	@type handler :: ([binary()] -> iodata())
62
63	## Lifecycle
64
65	# Reserve an ephemeral loopback port and node id without serving yet. Pass
66	# `inet6: true` to bind the IPv6 loopback — the node id is then the unbracketed
67	# "::1:port" form Redis uses in MOVED/ASK replies (issue #306). Pass
68	# `status: :down` to start unreachable (see `set_status/2`).
69	@spec reserve(keyword()) :: t()
70	def reserve(opts \\ []) do
71	inet6? = Keyword.get(opts, :inet6, false)	38✔
72	{host, family_opts} = if inet6?, do: {"::1", [:inet6]}, else: {"127.0.0.1", []}	38✔
73
74	{:ok, listen} =	38✔
75	:gen_tcp.listen(
76	0,
77	[:binary, active: false, reuseaddr: true, packet: :raw] ++ family_opts
78	)
79
80	{:ok, port} = :inet.port(listen)	38✔
81	{:ok, status} = Agent.start_link(fn -> Keyword.get(opts, :status, :up) end)	38✔
82	{:ok, accepted} = Agent.start_link(fn -> 0 end)	38✔
83
84	%FakeNode{	38✔
85	id: "#{host}:#{port}",	38✔
86	host: host,
87	port: port,
88	listen: listen,
89	inet6: inet6?,
90	status: status,
91	accepted: accepted
92	}
93	end
94
95	# Attach a handler to a reserved node and start accepting connections. The
96	# handler maps a parsed command (a list of binaries) to a raw RESP reply.
97	# Returns the node so it can be threaded through a pipe.
98	@spec serve(t(), handler()) :: t()
99	def serve(%FakeNode{} = node, handler) when is_function(handler, 1) do
100	spawn_link(fn -> accept_loop(node, handler) end)	38✔
101	node	38✔
102	end
103
104	# Reserve + serve in one step, for a node whose id no one needs in advance.
105	@spec start(handler(), keyword()) :: t()
106	def start(handler, opts \\ []) when is_function(handler, 1) do
107	opts \|> reserve() \|> serve(handler)	1✔
108	end
109
110	# Register a supervised `Redix` connection to the node under its id in
111	# `cluster`'s registry, so redirection code can resolve it by id. Must be
112	# called from the test process (uses `start_supervised!`). Returns the node.
113	@spec connect(t(), atom()) :: t()
114	def connect(%FakeNode{} = node, cluster) do
115	socket_opts = if node.inet6, do: [socket_opts: [:inet6]], else: []	27✔
116
117	ExUnit.Callbacks.start_supervised!(	27✔
118	{Redix,
119	[
120	host: node.host,	27✔
121	port: node.port,	27✔
122	sync_connect: true,
123	name: {:via, Registry, {:"#{cluster}_registry", node.id}}	27✔
124	] ++ socket_opts},
125	id: {:conn, node.id}	27✔
126	)
127
128	node	27✔
129	end
130
131	# reserve + connect + serve, for a fully wired node (redirection tests).
132	@spec start_connected(atom(), handler(), keyword()) :: t()
133	def start_connected(cluster, handler, opts \\ []) when is_function(handler, 1) do
134	opts \|> reserve() \|> connect(cluster) \|> serve(handler)	21✔
135	end
136
137	# Flip a node between serving (`:up`) and closing connections immediately
138	# (`:down`), to simulate a node going down/up under a running test.
139	@spec set_status(t(), :up \| :down) :: :ok
140	def set_status(%FakeNode{status: status}, value) when value in [:up, :down] do
141	Agent.update(status, fn _ -> value end)	1✔
142	end
143
144	# Total number of TCP connections ever accepted (transient topology-fetch
145	# sockets and managed Redix connections alike). Used to assert that no new
146	# dial happens — e.g. a redirect resolving to an already-connected node instead
147	# of opening a duplicate connection under a different address form (#340).
148	@spec connections_accepted(t()) :: non_neg_integer()
149	def connections_accepted(%FakeNode{accepted: accepted}) do
150	Agent.get(accepted, & &1)	2✔
151	end
152
153	## CLUSTER SLOTS encoding
154
155	# Encode a `CLUSTER SLOTS` reply. Each range is `{start, stop, primary}` or
156	# `{start, stop, primary, replicas}`, where a node is a `FakeNode`, a
157	# "host:port" string, or a `{host_or_nil, port}` tuple. A `nil` host is encoded
158	# as a RESP null bulk string, like Redis 7+ does with
159	# "cluster-preferred-endpoint-type unknown-endpoint" (issue #328). Mirrors the
160	# real RESP shape: an array of ranges, each `[start, stop, [host, port, id],
161	# [replica_host, replica_port, id]...]`.
162	@spec cluster_slots([tuple()]) :: binary()
163	def cluster_slots(ranges) do
164	body =	11✔
165	Enum.map(ranges, fn range ->
166	{start, stop, primary, replicas} =	11✔
167	case range do
168	{start, stop, primary} -> {start, stop, primary, []}	9✔
169	{start, stop, primary, replicas} -> {start, stop, primary, replicas}	2✔
170	end
171
172	nodes = [primary \| replicas]	11✔
173
174	[
175	"*#{2 + length(nodes)}\r\n",	11✔
176	":#{start}\r\n",	11✔
177	":#{stop}\r\n",	11✔
178	Enum.map(nodes, &encode_node/1)
179	]
180	end)
181
182	IO.iodata_to_binary(["*#{length(ranges)}\r\n", body])	11✔
183	end
184
185	defp encode_node(node) do
186	{host, port} = host_port(node)	13✔
187
188	[
189	"*3\r\n",
190	encode_host(host),
191	":#{port}\r\n",	13✔
192	"$40\r\n",
193	String.duplicate("a", 40),
194	"\r\n"
195	]
196	end
197
198	defp host_port(%FakeNode{host: host, port: port}), do: {host, port}	12✔
199	defp host_port({_host_or_nil, _port} = host_port), do: host_port	1✔
200
201	defp host_port(node_id) when is_binary(node_id) do
202	[host, port] = String.split(node_id, ":")	×
203	{host, String.to_integer(port)}
204	end
205
206	defp encode_host(nil), do: "$-1\r\n"	1✔
207	defp encode_host(host), do: ["$#{byte_size(host)}\r\n", host, "\r\n"]	12✔
208
209	## Polling helper
210
211	# Poll `fun` until it returns truthy or the timeout elapses (then flunk). Used
212	# by the fake-node tests to wait for the cluster to converge after a topology
213	# change, since reactive refreshes are async.
214	@spec wait_until((-> as_boolean(term())), timeout()) :: :ok
215	def wait_until(fun, timeout \\ 2_000)	5✔
216
217	def wait_until(_fun, timeout) when timeout <= 0 do
218	ExUnit.Assertions.flunk("condition not met in time")	×
219	end
220
221	def wait_until(fun, timeout) do
222	if fun.() do	8✔
223	:ok
224	else
225	Process.sleep(20)	3✔
226	wait_until(fun, timeout - 20)	3✔
227	end
228	end
229
230	## Accept / serve loop
231
232	# Accepts connections until the listen socket closes (when the owning test
233	# exits) or no client shows up before the timeout. Some fake nodes get more
234	# than one connection — e.g. the Manager's transient topology-fetch socket plus
235	# the managed Redix connection. While the node is `:down`, accepted connections
236	# are closed immediately to simulate an unreachable node.
237	defp accept_loop(node, handler) do
238	case :gen_tcp.accept(node.listen, @accept_timeout) do	94✔
239	{:ok, socket} ->
240	case Agent.get(node.status, & &1) do	56✔
241	:up ->
242	Agent.update(node.accepted, &(&1 + 1))	55✔
243	spawn_link(fn -> loop(socket, handler, "") end)	55✔
244
245	:down ->
246	:gen_tcp.close(socket)	1✔
247	end
248
249	accept_loop(node, handler)	56✔
250
UNCOV 251	{:error, _reason} ->	×
252	:ok
253	end
254	end
255
256	defp loop(socket, handler, buffer) do
257	case :gen_tcp.recv(socket, 0, @recv_timeout) do	120✔
258	{:ok, data} ->
259	{commands, rest} = parse_commands(buffer <> data, [])	65✔
260	Enum.each(commands, &:gen_tcp.send(socket, handler.(&1)))	65✔
261	loop(socket, handler, rest)	65✔
262
263	{:error, _reason} ->	17✔
264	:ok
265	end
266	end
267
268	## Minimal RESP request parser
269
270	# Pulls as many complete `*N\r\n$len\r\n...` commands as the buffer holds,
271	# returning the parsed commands and any leftover bytes.
272	defp parse_commands(buffer, acc) do
273	case parse_command(buffer) do	167✔
274	{:ok, command, rest} -> parse_commands(rest, [command \| acc])	102✔
275	:incomplete -> {Enum.reverse(acc), buffer}	65✔
276	end
277	end
278
279	defp parse_command("*" <> rest) do
280	with {:ok, count, rest} <- parse_int_line(rest) do	102✔
281	parse_bulk_strings(count, rest, [])	102✔
282	end
283	end
284
285	defp parse_command(_other), do: :incomplete	65✔
286
287	defp parse_bulk_strings(0, rest, acc), do: {:ok, Enum.reverse(acc), rest}	102✔
288
289	defp parse_bulk_strings(count, "$" <> rest, acc) do
290	with {:ok, length, rest} <- parse_int_line(rest) do	172✔
291	case rest do	172✔
292	<<value::binary-size(^length), "\r\n", rest::binary>> ->
293	parse_bulk_strings(count - 1, rest, [value \| acc])	172✔
294
295	_incomplete ->	×
296	:incomplete
297	end
298	end
299	end
300
301	defp parse_bulk_strings(_count, _rest, _acc), do: :incomplete	×
302
303	defp parse_int_line(binary) do
304	case :binary.split(binary, "\r\n") do	274✔
305	[int_string, rest] -> {:ok, String.to_integer(int_string), rest}	274✔
306	[_no_crlf_yet] -> :incomplete	×
307	end
308	end
309	end
310
311	defimpl String.Chars, for: Redix.Cluster.FakeNode do
312	def to_string(%Redix.Cluster.FakeNode{id: id}), do: id	53✔
313	end

whatyouhide / redix / b77331e2a0e8e8f5efc35b36ee1258afe7460002

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous