From c2583adfa164b2312a3661c65fee38d94073f39c Mon Sep 17 00:00:00 2001 From: Dave Miner Date: Fri, 12 Sep 2025 21:46:59 -0400 Subject: [PATCH 1/4] wait to set socket to active after chain_sync transitions to the correct state --- lib/chain_sync.ex | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/lib/chain_sync.ex b/lib/chain_sync.ex index e270d21..4c86544 100644 --- a/lib/chain_sync.ex +++ b/lib/chain_sync.ex @@ -255,7 +255,7 @@ defmodule Xander.ChainSync do end # The new_blocks state is when the client has caught up to the tip of the chain - # and passively receives new blocks from the node after msgAwaitReply responses. + # and receives new blocks from the node after msgAwaitReply responses. def new_blocks( :info, {_tcp_or_ssl, socket, data}, @@ -263,6 +263,7 @@ defmodule Xander.ChainSync do module_state ) do Logger.debug("handling new block") + :ok = setopts_lib(client).setopts(socket, active: :once) %{payload: payload, size: payload_length} = Util.plex!(data) remaining_payload_length = payload_length - byte_size(payload) @@ -289,7 +290,6 @@ defmodule Xander.ChainSync do case CSResponse.decode(combined_payload) do {:ok, %AwaitReply{}} -> - :ok = setopts_lib(client).setopts(socket, active: :once) :keep_state_and_data {:ok, %RollForward{header: header}} -> @@ -315,7 +315,6 @@ defmodule Xander.ChainSync do case CSResponse.decode(combined_payload) do {:ok, %AwaitReply{}} -> # Response should always be [1] msgAwaitReply - :ok = setopts_lib(client).setopts(socket, active: :once) {:keep_state, %{module_state | state: new_state}} error -> @@ -343,7 +342,6 @@ defmodule Xander.ChainSync do ) do {:ok, :next_block, new_state} -> :ok = client.send(socket, Messages.next_request()) - :ok = setopts_lib(client).setopts(socket, active: :once) {:keep_state, %{module_state | state: new_state}} {:ok, :stop} -> @@ -389,15 +387,11 @@ defmodule Xander.ChainSync do {:ok, %AwaitReply{}} -> Logger.debug("Awaiting reply") # This is the base case of the recursion and the function no - # longer recurses. It sets the socket to active mode so that - # data ingestion continues from the "new_blocks" state. - # The state transition from the "catching up" state to - # "new_blocks" state occurs in the caller of this function. - - # TODO: address race condition that takes place in case the - # node replies after socket is set to active but before the - # client has transitioned to the new state. - :ok = setopts_lib(client).setopts(socket, active: :once) + # longer recurses. The socket will be set to active mode + # after the state transition occurs in the caller of this function. + # This avoids the race condition where the node could reply + # after socket is set to active but before state transition. + :ok {:ok, %RollForward{header: header}} -> # This is the callback from the client module From ba33c93f7e58ef8ba2fc2a30531b58d1f1307ff6 Mon Sep 17 00:00:00 2001 From: Dave Miner Date: Fri, 26 Sep 2025 11:49:08 -0400 Subject: [PATCH 2/4] add state handlers for messages received during catching_up --- lib/chain_sync.ex | 144 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 137 insertions(+), 7 deletions(-) diff --git a/lib/chain_sync.ex b/lib/chain_sync.ex index 4c86544..1ebabe2 100644 --- a/lib/chain_sync.ex +++ b/lib/chain_sync.ex @@ -254,8 +254,132 @@ defmodule Xander.ChainSync do end end + # Handle socket messages that arrive during the race condition window + # between setting active mode and transitioning to new_blocks state + def catching_up( + :info, + {_tcp_or_ssl, socket, data}, + %__MODULE__{ + client: client, + socket: socket, + client_module: client_module, + state: client_state + } = + module_state + ) do + Logger.debug( + "Received socket message during catching_up state transition - handling gracefully" + ) + + %{payload: payload, size: payload_length} = Util.plex!(data) + remaining_payload_length = payload_length - byte_size(payload) + + # This ensures that if the entire payload has been read sent in data, + # we don't try to read anymore. + read_remaining_payload = fn + current_payload, 0 -> + {:ok, current_payload} + + current_payload, recv_payload_length -> + case client.recv(socket, recv_payload_length, @recv_timeout) do + {:ok, additional_payload} -> + {:ok, current_payload <> additional_payload} + + {:error, reason} -> + {:error, reason} + end + end + + case read_remaining_payload.(payload, remaining_payload_length) do + {:ok, combined_payload} -> + Logger.debug("read payload during state transition") + + case CSResponse.decode(combined_payload) do + {:ok, %AwaitReply{}} -> + # We're at the tip, transition to new_blocks and re-enable active mode + :ok = setopts_lib(client).setopts(socket, active: :once) + {:next_state, :new_blocks, module_state} + + {:ok, %RollForward{header: header}} -> + Logger.debug("decoded block during transition") + + case client_module.handle_block( + %{ + block_number: header.block_number, + size: header.block_body_size + }, + client_state + ) do + {:ok, :next_block, _new_state} -> + :ok = client.send(socket, Messages.next_request()) + :ok = setopts_lib(client).setopts(socket, active: :once) + :keep_state_and_data + + {:close, _new_state} -> + Logger.debug("Disconnecting from node") + :ok = client.close(socket) + {:next_state, :disconnected, module_state} + end + + {:ok, %RollBackward{point: point}} -> + Logger.debug( + "Calling client_module.handle_rollback during transition with #{point.slot_number}, #{point.hash}" + ) + + case client_module.handle_rollback( + %{ + slot_number: point.slot_number, + block_hash: point.hash + }, + client_state + ) do + {:ok, :next_block, _new_state} -> + :ok = client.send(socket, Messages.next_request()) + :ok = setopts_lib(client).setopts(socket, active: :once) + :keep_state_and_data + end + + {:error, _} -> + # If decoding fails, try to read another message + read_next_message_continue( + client, + socket, + combined_payload, + client_module, + client_state + ) + + :ok = setopts_lib(client).setopts(socket, active: :once) + :keep_state_and_data + + unknown_response -> + Logger.debug("Unknown message during transition: #{inspect(unknown_response)}") + :ok = setopts_lib(client).setopts(socket, active: :once) + :keep_state_and_data + end + + {:error, reason} -> + Logger.debug("Failed to read payload during transition: #{inspect(reason)}") + :ok = setopts_lib(client).setopts(socket, active: :once) + :keep_state_and_data + end + end + + # Handle socket close/error events during catching_up state + def catching_up(:info, {tcp_or_ssl_closed, _socket}, module_state) + when tcp_or_ssl_closed in [:tcp_closed, :ssl_closed] do + Logger.warning("Socket closed during catching_up state transition") + {:next_state, :disconnected, module_state} + end + + def catching_up(:info, {tcp_or_ssl_error, _socket, reason}, module_state) + when tcp_or_ssl_error in [:tcp_error, :ssl_error] do + Logger.error("Socket error during catching_up state transition: #{inspect(reason)}") + {:next_state, :disconnected, module_state} + end + # The new_blocks state is when the client has caught up to the tip of the chain - # and receives new blocks from the node after msgAwaitReply responses. + # and passively receives new blocks from the node after msgAwaitReply responses. def new_blocks( :info, {_tcp_or_ssl, socket, data}, @@ -263,7 +387,6 @@ defmodule Xander.ChainSync do module_state ) do Logger.debug("handling new block") - :ok = setopts_lib(client).setopts(socket, active: :once) %{payload: payload, size: payload_length} = Util.plex!(data) remaining_payload_length = payload_length - byte_size(payload) @@ -290,6 +413,7 @@ defmodule Xander.ChainSync do case CSResponse.decode(combined_payload) do {:ok, %AwaitReply{}} -> + :ok = setopts_lib(client).setopts(socket, active: :once) :keep_state_and_data {:ok, %RollForward{header: header}} -> @@ -315,6 +439,7 @@ defmodule Xander.ChainSync do case CSResponse.decode(combined_payload) do {:ok, %AwaitReply{}} -> # Response should always be [1] msgAwaitReply + :ok = setopts_lib(client).setopts(socket, active: :once) {:keep_state, %{module_state | state: new_state}} error -> @@ -342,6 +467,7 @@ defmodule Xander.ChainSync do ) do {:ok, :next_block, new_state} -> :ok = client.send(socket, Messages.next_request()) + :ok = setopts_lib(client).setopts(socket, active: :once) {:keep_state, %{module_state | state: new_state}} {:ok, :stop} -> @@ -387,11 +513,15 @@ defmodule Xander.ChainSync do {:ok, %AwaitReply{}} -> Logger.debug("Awaiting reply") # This is the base case of the recursion and the function no - # longer recurses. The socket will be set to active mode - # after the state transition occurs in the caller of this function. - # This avoids the race condition where the node could reply - # after socket is set to active but before state transition. - :ok + # longer recurses. It sets the socket to active mode so that + # data ingestion continues from the "new_blocks" state. + # The state transition from the "catching up" state to + # "new_blocks" state occurs in the caller of this function. + + # TODO: address race condition that takes place in case the + # node replies after socket is set to active but before the + # client has transitioned to the new state. + :ok = setopts_lib(client).setopts(socket, active: :once) {:ok, %RollForward{header: header}} -> # This is the callback from the client module From c6ea4d0757e7cc8b3e2ed03af460eea7bd8ca345 Mon Sep 17 00:00:00 2001 From: Dave Miner Date: Sat, 10 Jan 2026 00:48:49 -0500 Subject: [PATCH 3/4] use transport module in new chain_sync states --- lib/chain_sync.ex | 25 +++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/lib/chain_sync.ex b/lib/chain_sync.ex index 7204ff6..55cf76d 100644 --- a/lib/chain_sync.ex +++ b/lib/chain_sync.ex @@ -253,7 +253,8 @@ defmodule Xander.ChainSync do :info, {_tcp_or_ssl, socket, data}, %__MODULE__{ - client: client, + transport: transport, + # client: client, socket: socket, client_module: client_module, state: client_state @@ -274,7 +275,7 @@ defmodule Xander.ChainSync do {:ok, current_payload} current_payload, recv_payload_length -> - case client.recv(socket, recv_payload_length, @recv_timeout) do + case Transport.recv(transport, socket, recv_payload_length, @recv_timeout) do {:ok, additional_payload} -> {:ok, current_payload <> additional_payload} @@ -290,7 +291,7 @@ defmodule Xander.ChainSync do case CSResponse.decode(combined_payload) do {:ok, %AwaitReply{}} -> # We're at the tip, transition to new_blocks and re-enable active mode - :ok = setopts_lib(client).setopts(socket, active: :once) + :ok = Transport.setopts(transport, socket, active: :once) {:next_state, :new_blocks, module_state} {:ok, %RollForward{header: header}} -> @@ -304,13 +305,13 @@ defmodule Xander.ChainSync do client_state ) do {:ok, :next_block, _new_state} -> - :ok = client.send(socket, Messages.next_request()) - :ok = setopts_lib(client).setopts(socket, active: :once) + :ok = Transport.send(transport, socket, Messages.next_request()) + :ok = Transport.setopts(transport, socket, active: :once) :keep_state_and_data {:close, _new_state} -> Logger.debug("Disconnecting from node") - :ok = client.close(socket) + :ok = Transport.close(transport, socket) {:next_state, :disconnected, module_state} end @@ -327,33 +328,33 @@ defmodule Xander.ChainSync do client_state ) do {:ok, :next_block, _new_state} -> - :ok = client.send(socket, Messages.next_request()) - :ok = setopts_lib(client).setopts(socket, active: :once) + :ok = Transport.send(transport, socket, Messages.next_request()) + :ok = Transport.setopts(transport, socket, active: :once) :keep_state_and_data end {:error, _} -> # If decoding fails, try to read another message read_next_message_continue( - client, + transport, socket, combined_payload, client_module, client_state ) - :ok = setopts_lib(client).setopts(socket, active: :once) + :ok = Transport.setopts(transport, socket, active: :once) :keep_state_and_data unknown_response -> Logger.debug("Unknown message during transition: #{inspect(unknown_response)}") - :ok = setopts_lib(client).setopts(socket, active: :once) + :ok = Transport.setopts(transport, socket, active: :once) :keep_state_and_data end {:error, reason} -> Logger.debug("Failed to read payload during transition: #{inspect(reason)}") - :ok = setopts_lib(client).setopts(socket, active: :once) + :ok = Transport.setopts(transport, socket, active: :once) :keep_state_and_data end end From 2bed246cd47e0df3522bcf404c0736ca8e1a2173 Mon Sep 17 00:00:00 2001 From: Dave Miner Date: Sun, 11 Jan 2026 17:09:09 -0500 Subject: [PATCH 4/4] combine message handling for catching_up and new_block states --- lib/chain_sync.ex | 439 ++++++++++++++++++---------------------------- 1 file changed, 167 insertions(+), 272 deletions(-) diff --git a/lib/chain_sync.ex b/lib/chain_sync.ex index 55cf76d..ce711d5 100644 --- a/lib/chain_sync.ex +++ b/lib/chain_sync.ex @@ -197,6 +197,8 @@ defmodule Xander.ChainSync do :find_intersection, %__MODULE__{transport: transport, socket: socket, sync_from: sync_from} = data ) do + Logger.debug("finding intersection") + with {:ok, %IntersectionTarget{slot: slot, block_bytes: block_bytes}} <- Intersection.find_target(transport, socket, sync_from), {:ok, %IntersectFound{}} <- @@ -221,15 +223,8 @@ defmodule Xander.ChainSync do emit_initial_next_message = fn transport, socket -> with :ok <- Transport.send(transport, socket, Messages.next_request()), - {:ok, header_bytes} <- Transport.recv(transport, socket, 8, @recv_timeout), - # TODO: use Util.plex! - <<_timestamp::big-32, _mode::1, _protocol_id::15, payload_length::big-16>> <- - header_bytes, - {:ok, payload} <- Transport.recv(transport, socket, payload_length, @recv_timeout) do + {:ok, payload} <- recv_message(transport, socket) do CSResponse.decode(payload) - else - {:error, reason} -> - {:error, reason} end end @@ -252,89 +247,73 @@ defmodule Xander.ChainSync do def catching_up( :info, {_tcp_or_ssl, socket, data}, - %__MODULE__{ - transport: transport, - # client: client, - socket: socket, - client_module: client_module, - state: client_state - } = - module_state + %__MODULE__{socket: socket} = module_state ) do Logger.debug( "Received socket message during catching_up state transition - handling gracefully" ) - %{payload: payload, size: payload_length} = Util.plex!(data) - remaining_payload_length = payload_length - byte_size(payload) + case handle_socket_data(data, module_state) do + :ok -> {:next_state, :new_blocks, module_state} + result -> result + end + end - # This ensures that if the entire payload has been read sent in data, - # we don't try to read anymore. - read_remaining_payload = fn - current_payload, 0 -> - {:ok, current_payload} + # Handle socket close/error events during catching_up state + def catching_up(:info, {tcp_or_ssl_closed, _socket}, module_state) + when tcp_or_ssl_closed in [:tcp_closed, :ssl_closed] do + Logger.warning("Socket closed during catching_up state transition") + {:next_state, :disconnected, module_state} + end - current_payload, recv_payload_length -> - case Transport.recv(transport, socket, recv_payload_length, @recv_timeout) do - {:ok, additional_payload} -> - {:ok, current_payload <> additional_payload} + def catching_up(:info, {tcp_or_ssl_error, _socket, reason}, module_state) + when tcp_or_ssl_error in [:tcp_error, :ssl_error] do + Logger.error("Socket error during catching_up state transition: #{inspect(reason)}") + {:next_state, :disconnected, module_state} + end - {:error, reason} -> - {:error, reason} - end + # The new_blocks state is when the client has caught up to the tip of the chain + # and passively receives new blocks from the node after msgAwaitReply responses. + def new_blocks( + :info, + {_tcp_or_ssl, socket, data}, + %__MODULE__{socket: socket} = module_state + ) do + Logger.debug("handling new block") + + case handle_socket_data(data, module_state) do + :ok -> :keep_state_and_data + result -> result end + end - case read_remaining_payload.(payload, remaining_payload_length) do - {:ok, combined_payload} -> - Logger.debug("read payload during state transition") + # Common handler for socket data in both catching_up and new_blocks states + defp handle_socket_data( + data, + %__MODULE__{ + transport: transport, + socket: socket, + client_module: client_module, + state: client_state + } = module_state + ) do + %{payload: payload, size: payload_length} = Util.plex!(data) + remaining_payload_length = payload_length - byte_size(payload) + case read_remaining_payload(transport, socket, payload, remaining_payload_length) do + {:ok, combined_payload} -> case CSResponse.decode(combined_payload) do {:ok, %AwaitReply{}} -> - # We're at the tip, transition to new_blocks and re-enable active mode :ok = Transport.setopts(transport, socket, active: :once) - {:next_state, :new_blocks, module_state} + :ok {:ok, %RollForward{header: header}} -> - Logger.debug("decoded block during transition") - - case client_module.handle_block( - %{ - block_number: header.block_number, - size: header.block_body_size - }, - client_state - ) do - {:ok, :next_block, _new_state} -> - :ok = Transport.send(transport, socket, Messages.next_request()) - :ok = Transport.setopts(transport, socket, active: :once) - :keep_state_and_data - - {:close, _new_state} -> - Logger.debug("Disconnecting from node") - :ok = Transport.close(transport, socket) - {:next_state, :disconnected, module_state} - end + handle_roll_forward(header, module_state) {:ok, %RollBackward{point: point}} -> - Logger.debug( - "Calling client_module.handle_rollback during transition with #{point.slot_number}, #{point.hash}" - ) - - case client_module.handle_rollback( - %{ - slot_number: point.slot_number, - block_hash: point.hash - }, - client_state - ) do - {:ok, :next_block, _new_state} -> - :ok = Transport.send(transport, socket, Messages.next_request()) - :ok = Transport.setopts(transport, socket, active: :once) - :keep_state_and_data - end + handle_roll_backward(point, module_state) {:error, _} -> - # If decoding fails, try to read another message read_next_message_continue( transport, socket, @@ -347,77 +326,92 @@ defmodule Xander.ChainSync do :keep_state_and_data unknown_response -> - Logger.debug("Unknown message during transition: #{inspect(unknown_response)}") + Logger.debug("Unknown message: #{inspect(unknown_response)}") :ok = Transport.setopts(transport, socket, active: :once) :keep_state_and_data end {:error, reason} -> - Logger.debug("Failed to read payload during transition: #{inspect(reason)}") + Logger.debug("Failed to read payload: #{inspect(reason)}") :ok = Transport.setopts(transport, socket, active: :once) :keep_state_and_data end end - # Handle socket close/error events during catching_up state - def catching_up(:info, {tcp_or_ssl_closed, _socket}, module_state) - when tcp_or_ssl_closed in [:tcp_closed, :ssl_closed] do - Logger.warning("Socket closed during catching_up state transition") - {:next_state, :disconnected, module_state} - end + defp handle_roll_forward( + header, + %__MODULE__{ + transport: transport, + socket: socket, + client_module: client_module, + state: client_state + } = module_state + ) do + case client_module.handle_block( + %{block_number: header.block_number, size: header.block_body_size}, + client_state + ) do + {:ok, :next_block, new_state} -> + :ok = Transport.send(transport, socket, Messages.next_request()) + :ok = Transport.setopts(transport, socket, active: :once) + {:keep_state, %{module_state | state: new_state}} - def catching_up(:info, {tcp_or_ssl_error, _socket, reason}, module_state) - when tcp_or_ssl_error in [:tcp_error, :ssl_error] do - Logger.error("Socket error during catching_up state transition: #{inspect(reason)}") - {:next_state, :disconnected, module_state} + {:close, new_state} -> + Logger.debug("Disconnecting from node") + :ok = Transport.close(transport, socket) + {:next_state, :disconnected, %{module_state | state: new_state}} + end end - # The new_blocks state is when the client has caught up to the tip of the chain - # and passively receives new blocks from the node after msgAwaitReply responses. - def new_blocks( - :info, - {_tcp_or_ssl, socket, data}, - %__MODULE__{ - transport: transport, - socket: socket, - client_module: client_module, - state: state - } = - module_state - ) do - Logger.debug("handling new block") - - %{payload: payload, size: payload_length} = Util.plex!(data) - remaining_payload_length = payload_length - byte_size(payload) - - # This ensures that if the entire payload has been read sent in data, - # we don't try to read anymore. - read_remaining_payload = fn - current_payload, 0 -> - {:ok, current_payload} - - current_payload, recv_payload_length -> - case Transport.recv(transport, socket, recv_payload_length, @recv_timeout) do - {:ok, additional_payload} -> - {:ok, current_payload <> additional_payload} + defp handle_roll_backward( + point, + %__MODULE__{ + transport: transport, + socket: socket, + client_module: client_module, + state: client_state + } = module_state + ) do + Logger.debug("Calling client_module.handle_rollback with #{point.slot_number}, #{point.hash}") + + case client_module.handle_rollback( + %{slot_number: point.slot_number, block_hash: point.hash}, + client_state + ) do + {:ok, :next_block, new_state} -> + :ok = Transport.send(transport, socket, Messages.next_request()) + :ok = Transport.setopts(transport, socket, active: :once) + {:keep_state, %{module_state | state: new_state}} - {:error, reason} -> - {:error, reason} - end + {:ok, :stop} -> + {:next_state, :disconnected, module_state} end + end - case read_remaining_payload.(payload, remaining_payload_length) do - {:ok, combined_payload} -> - Logger.debug("read payload on new block") + # Helper function to read the next message + defp read_until_sync(transport, socket, client_module, state) do + Logger.debug("read_until_sync") - case CSResponse.decode(combined_payload) do + case recv_message(transport, socket) do + {:ok, payload} -> + case CSResponse.decode(payload) do + # When we receive a msgAwaitReply, this means we have reached + # the tip and are done with the sync. {:ok, %AwaitReply{}} -> + Logger.debug("Awaiting reply") + # This is the base case of the recursion and the function no + # longer recurses. It sets the socket to active mode so that + # data ingestion continues from the "new_blocks" state. + # The state transition from the "catching up" state to + # "new_blocks" state occurs in the caller of this function. + + # TODO: address race condition that takes place in case the + # node replies after socket is set to active but before the + # client has transitioned to the new state. :ok = Transport.setopts(transport, socket, active: :once) - :keep_state_and_data {:ok, %RollForward{header: header}} -> - Logger.debug("decoded block perfectly") - + # This is the callback from the client module case client_module.handle_block( %{ block_number: header.block_number, @@ -427,184 +421,85 @@ defmodule Xander.ChainSync do ) do {:ok, :next_block, new_state} -> :ok = Transport.send(transport, socket, Messages.next_request()) - - {:ok, data} = Transport.recv(transport, socket, 8, @recv_timeout) - %{payload: payload, size: payload_length} = Util.plex!(data) - remaining_payload_length = payload_length - byte_size(payload) - - {:ok, combined_payload} = - read_remaining_payload.(payload, remaining_payload_length) - - case CSResponse.decode(combined_payload) do - {:ok, %AwaitReply{}} -> - # Response should always be [1] msgAwaitReply - :ok = Transport.setopts(transport, socket, active: :once) - {:keep_state, %{module_state | state: new_state}} - - error -> - Logger.warning("Error decoding next request: #{inspect(error)}") - {:keep_state, %{module_state | state: new_state}} - end + read_until_sync(transport, socket, client_module, new_state) {:close, new_state} -> Logger.debug("Disconnecting from node") :ok = Transport.close(transport, socket) - {:next_state, :disconnected, %{module_state | state: new_state}} + {:next_state, :disconnected, new_state} end - {:ok, %RollBackward{point: point}} -> - Logger.debug( - "Calling client_module.handle_rollback with #{point.slot_number}, #{point.hash}" - ) - - case client_module.handle_rollback( - %{ - slot_number: point.slot_number, - block_hash: point.hash - }, - state - ) do - {:ok, :next_block, new_state} -> - :ok = Transport.send(transport, socket, Messages.next_request()) - :ok = Transport.setopts(transport, socket, active: :once) - {:keep_state, %{module_state | state: new_state}} - - {:ok, :stop} -> - {:next_state, :disconnected, module_state} - end - - {:error, _} -> + {:error, :incomplete_cbor_data} -> # If decoding fails, try to read another message - read_next_message_continue( - transport, - socket, - combined_payload, - client_module, - state - ) - - :keep_state_and_data - - unknown_response -> - Logger.debug("Unknown message: #{inspect(unknown_response)}") - :keep_state_and_data + read_next_message_continue(transport, socket, payload, client_module, state) end {:error, reason} -> - Logger.debug("Failed to read payload on new block: #{inspect(reason)}") - :keep_state_and_data + Logger.debug("Failed to recv message: #{inspect(reason)}") + {:error, reason} end end - # Helper function to read the next message - defp read_until_sync(transport, socket, client_module, state) do - # Read the header (8 bytes) - case Transport.recv(transport, socket, 8, @recv_timeout) do - {:ok, header_bytes} -> - # TODO: use Util.plex! - <<_timestamp::big-32, _mode::1, _protocol_id::15, payload_length::big-16>> = header_bytes - - case Transport.recv(transport, socket, payload_length, @recv_timeout) do - {:ok, payload} -> - case CSResponse.decode(payload) do - # When we receive a msgAwaitReply, this means we have reached - # the tip and are done with the sync. - {:ok, %AwaitReply{}} -> - Logger.debug("Awaiting reply") - # This is the base case of the recursion and the function no - # longer recurses. It sets the socket to active mode so that - # data ingestion continues from the "new_blocks" state. - # The state transition from the "catching up" state to - # "new_blocks" state occurs in the caller of this function. - - # TODO: address race condition that takes place in case the - # node replies after socket is set to active but before the - # client has transitioned to the new state. - :ok = Transport.setopts(transport, socket, active: :once) - - {:ok, %RollForward{header: header}} -> - # This is the callback from the client module - case client_module.handle_block( - %{ - block_number: header.block_number, - size: header.block_body_size - }, - state - ) do - {:ok, :next_block, new_state} -> - :ok = Transport.send(transport, socket, Messages.next_request()) - read_until_sync(transport, socket, client_module, new_state) - - {:close, new_state} -> - Logger.debug("Disconnecting from node") - :ok = Transport.close(transport, socket) - {:next_state, :disconnected, new_state} - end - - {:error, :incomplete_cbor_data} -> - # If decoding fails, try to read another message - read_next_message_continue(transport, socket, payload, client_module, state) - end + # Helper to read remaining payload bytes from socket, or return current payload if nothing left to read + defp read_remaining_payload(_transport, _socket, current_payload, 0), do: {:ok, current_payload} - {:error, reason} -> - Logger.debug("Failed to read payload: #{inspect(reason)}") - {:error, reason} - end + defp read_remaining_payload(transport, socket, current_payload, recv_payload_length) do + case Transport.recv(transport, socket, recv_payload_length, @recv_timeout) do + {:ok, additional_payload} -> + {:ok, current_payload <> additional_payload} {:error, reason} -> - Logger.debug("Failed to read header: #{inspect(reason)}") {:error, reason} end end + # Helper to read a complete multiplexed message (header + payload) from socket + defp recv_message(transport, socket) do + with {:ok, header_bytes} <- Transport.recv(transport, socket, 8, @recv_timeout), + %{size: payload_length} <- Util.plex!(header_bytes), + {:ok, payload} <- Transport.recv(transport, socket, payload_length, @recv_timeout) do + {:ok, payload} + end + end + # Helper function to continue reading if the first attempt fails defp read_next_message_continue(transport, socket, first_payload, client_module, state) do - # Read another header - case Transport.recv(transport, socket, 8, @recv_timeout) do - {:ok, header_bytes} -> - # TODO: use Util.plex! - <<_timestamp::big-32, _mode::1, _protocol_id::15, payload_length::big-16>> = header_bytes - - # Read another payload - case Transport.recv(transport, socket, payload_length, @recv_timeout) do - {:ok, second_payload} -> - # Combine the payloads and try to decode - combined_payload = first_payload <> second_payload - - case CSResponse.decode(combined_payload) do - {:ok, %RollForward{header: header}} -> - case client_module.handle_block( - %{ - block_number: header.block_number, - size: header.block_body_size - }, - state - ) do - {:ok, :next_block, new_state} -> - :ok = Transport.send(transport, socket, Messages.next_request()) - read_until_sync(transport, socket, client_module, new_state) - - {:ok, :stop} -> - :ok - end - - {:error, :incomplete_cbor_data} -> - read_next_message_continue( - transport, - socket, - combined_payload, - client_module, - state - ) + Logger.debug("read_next_message_continue") + + case recv_message(transport, socket) do + {:ok, second_payload} -> + # Combine the payloads and try to decode + combined_payload = first_payload <> second_payload + + case CSResponse.decode(combined_payload) do + {:ok, %RollForward{header: header}} -> + case client_module.handle_block( + %{ + block_number: header.block_number, + size: header.block_body_size + }, + state + ) do + {:ok, :next_block, new_state} -> + :ok = Transport.send(transport, socket, Messages.next_request()) + read_until_sync(transport, socket, client_module, new_state) + + {:ok, :stop} -> + :ok end - {:error, reason} -> - Logger.debug("Failed to read second payload: #{inspect(reason)}") - {:error, reason} + {:error, :incomplete_cbor_data} -> + read_next_message_continue( + transport, + socket, + combined_payload, + client_module, + state + ) end {:error, reason} -> - Logger.debug("Failed to read second header: #{inspect(reason)}") + Logger.debug("Failed to recv message: #{inspect(reason)}") {:error, reason} end end