defmodule Object.RobustNetworkTransport do @moduledoc """ A more robust network transport layer that handles startup failures gracefully. This module addresses common issues in network startup by: - Implementing graceful degradation when transport services fail - Providing fallback mechanisms for connection establishment - Better error handling and recovery strategies - Improved process lifecycle management """ use GenServer require Logger @type transport_type :: :tcp | :udp | :websocket | :grpc @type connection_opts :: [ host: String.t(), port: non_neg_integer(), transport: transport_type(), ssl: boolean(), pool_size: pos_integer(), timeout: timeout(), reconnect_interval: pos_integer(), max_reconnect_attempts: pos_integer() ] @type connection_id :: String.t() @type connection_state :: :connecting | :connected | :disconnected | :failed defstruct [ :id, :socket, :transport, :host, :port, :state, :last_activity, :created_at, :metrics, :reconnect_attempts, :max_reconnect_attempts ] # Public API def start_link(opts \\ []) do name = Keyword.get(opts, :name, __MODULE__) GenServer.start_link(__MODULE__, opts, name: name) end @doc """ Connects to a remote endpoint with robust error handling. """ def connect(host, port, opts \\ []) do transport_name = Keyword.get(opts, :transport_name, __MODULE__) case GenServer.whereis(transport_name) do nil -> Logger.error("NetworkTransport process not available, attempting fallback connection") fallback_connect(host, port, opts) pid -> try do GenServer.call(pid, {:connect, host, port, opts}, 30_000) catch :exit, {:noproc, _} -> Logger.warning("NetworkTransport process died during call, using fallback") fallback_connect(host, port, opts) :exit, {:timeout, _} -> Logger.warning("NetworkTransport call timed out, using fallback") fallback_connect(host, port, opts) end end end @doc """ Sends data with automatic retry and fallback mechanisms. """ def send_data(connection_id, data, opts \\ []) do transport_name = Keyword.get(opts, :transport_name, __MODULE__) case GenServer.whereis(transport_name) do nil -> Logger.warning("NetworkTransport not available for send_data") {:error, :transport_unavailable} pid -> try do GenServer.call(pid, {:send, connection_id, data}, 10_000) catch :exit, reason -> Logger.warning("Send failed: #{inspect(reason)}") {:error, :send_failed} end end end @doc """ Checks if the transport service is available and healthy. """ def health_check(transport_name \\ __MODULE__) do case GenServer.whereis(transport_name) do nil -> {:error, :not_running} pid when is_pid(pid) -> try do GenServer.call(pid, :ping, 5_000) catch :exit, _ -> {:error, :unhealthy} end end end @doc """ Gracefully starts the transport service with fallback options. """ def ensure_started(opts \\ []) do case start_link(opts) do {:ok, pid} -> Logger.info("NetworkTransport started successfully") {:ok, pid} {:error, {:already_started, pid}} -> Logger.debug("NetworkTransport already running") {:ok, pid} {:error, reason} -> Logger.error("Failed to start NetworkTransport: #{inspect(reason)}") # Try to start with minimal configuration minimal_opts = [ pool_size: 1, timeout: 5_000, fallback_mode: true ] case start_link(minimal_opts) do {:ok, pid} -> Logger.warning("NetworkTransport started in fallback mode") {:ok, pid} error -> Logger.error("Failed to start even minimal NetworkTransport: #{inspect(error)}") error end end end # GenServer Callbacks @impl true def init(opts) do fallback_mode = Keyword.get(opts, :fallback_mode, false) config = %{ pool_size: Keyword.get(opts, :pool_size, 5), timeout: Keyword.get(opts, :timeout, 30_000), max_reconnect_attempts: Keyword.get(opts, :max_reconnect_attempts, 3), fallback_mode: fallback_mode } state = %{ connections: %{}, pools: %{}, config: config, metrics: init_metrics(), health_check_ref: nil, fallback_connections: %{} } # Start health check timer if not in fallback mode state = if fallback_mode do Logger.warning("Starting NetworkTransport in fallback mode - limited functionality") state else ref = Process.send_after(self(), :health_check, 10_000) %{state | health_check_ref: ref} end {:ok, state} end @impl true def handle_call(:ping, _from, state) do {:reply, :pong, state} end @impl true def handle_call({:connect, host, port, opts}, _from, state) do if state.config.fallback_mode do handle_fallback_connect(host, port, opts, state) else handle_normal_connect(host, port, opts, state) end end @impl true def handle_call({:send, conn_id, data}, _from, state) do cond do state.config.fallback_mode -> handle_fallback_send(conn_id, data, state) Map.has_key?(state.connections, conn_id) -> handle_normal_send(conn_id, data, state) true -> {:reply, {:error, :connection_not_found}, state} end end @impl true def handle_call(:get_metrics, _from, state) do {:reply, state.metrics, state} end @impl true def handle_cast({:disconnect, conn_id}, state) do new_state = close_connection(state, conn_id) {:noreply, new_state} end @impl true def handle_info(:health_check, state) do # Perform health checks on connections new_state = perform_health_checks(state) # Schedule next health check ref = Process.send_after(self(), :health_check, 30_000) new_state = %{new_state | health_check_ref: ref} {:noreply, new_state} end @impl true def handle_info({:tcp, socket, data}, state) do # Handle incoming TCP data handle_incoming_data(socket, data, state) end @impl true def handle_info({:tcp_closed, socket}, state) do # Handle TCP connection closed handle_connection_closed(socket, state) end @impl true def handle_info({:tcp_error, socket, reason}, state) do # Handle TCP connection error handle_connection_error(socket, reason, state) end # Private Functions defp handle_normal_connect(host, port, opts, state) do connection_id = generate_connection_id(host, port) transport = Keyword.get(opts, :transport, :tcp) case establish_tcp_connection(host, port, opts) do {:ok, socket} -> connection = %__MODULE__{ id: connection_id, socket: socket, transport: transport, host: host, port: port, state: :connected, last_activity: DateTime.utc_now(), created_at: DateTime.utc_now(), metrics: %{bytes_sent: 0, bytes_received: 0, messages: 0}, reconnect_attempts: 0, max_reconnect_attempts: state.config.max_reconnect_attempts } new_state = put_in(state.connections[connection_id], connection) new_state = update_metrics(new_state, :connection_established) {:reply, {:ok, connection_id}, new_state} {:error, reason} -> Logger.warning("Failed to establish connection to #{host}:#{port} - #{inspect(reason)}") {:reply, {:error, reason}, state} end end defp handle_fallback_connect(host, port, opts, state) do # In fallback mode, create a mock connection that can be used for testing connection_id = generate_connection_id(host, port) fallback_connection = %{ id: connection_id, host: host, port: port, type: :fallback, created_at: DateTime.utc_now() } new_state = put_in(state.fallback_connections[connection_id], fallback_connection) Logger.debug("Created fallback connection to #{host}:#{port}") {:reply, {:ok, connection_id}, new_state} end defp handle_normal_send(conn_id, data, state) do connection = state.connections[conn_id] case do_send_tcp_data(connection.socket, data) do :ok -> # Update connection metrics updated_connection = update_connection_activity(connection, byte_size(data)) new_state = put_in(state.connections[conn_id], updated_connection) new_state = update_metrics(new_state, :message_sent, byte_size(data)) {:reply, :ok, new_state} {:error, reason} -> Logger.warning("Failed to send data on connection #{conn_id}: #{inspect(reason)}") {:reply, {:error, reason}, state} end end defp handle_fallback_send(conn_id, data, state) do case state.fallback_connections[conn_id] do nil -> {:reply, {:error, :connection_not_found}, state} _connection -> # In fallback mode, just log the send attempt Logger.debug("Fallback send: #{byte_size(data)} bytes to #{conn_id}") new_state = update_metrics(state, :message_sent, byte_size(data)) {:reply, :ok, new_state} end end defp establish_tcp_connection(host, port, opts) do timeout = Keyword.get(opts, :timeout, 10_000) tcp_opts = [ :binary, {:active, true}, {:packet, 0}, {:nodelay, true}, {:keepalive, true} ] case :gen_tcp.connect(String.to_charlist(host), port, tcp_opts, timeout) do {:ok, socket} -> {:ok, socket} {:error, reason} -> {:error, reason} end end defp do_send_tcp_data(socket, data) do :gen_tcp.send(socket, data) end defp generate_connection_id(host, port) do "#{host}:#{port}:#{:erlang.unique_integer([:positive])}" end defp update_connection_activity(connection, bytes_sent) do updated_metrics = %{ connection.metrics | bytes_sent: connection.metrics.bytes_sent + bytes_sent, messages: connection.metrics.messages + 1 } %{connection | last_activity: DateTime.utc_now(), metrics: updated_metrics } end defp update_metrics(state, event, data \\ 0) do metrics = case event do :connection_established -> %{state.metrics | total_connections: state.metrics.total_connections + 1, active_connections: map_size(state.connections) } :connection_closed -> %{state.metrics | active_connections: map_size(state.connections) } :message_sent -> %{state.metrics | total_bytes_sent: state.metrics.total_bytes_sent + data, total_messages: state.metrics.total_messages + 1 } _ -> state.metrics end %{state | metrics: metrics} end defp init_metrics do %{ total_connections: 0, active_connections: 0, total_bytes_sent: 0, total_bytes_received: 0, total_messages: 0, avg_latency_ms: 0.0, uptime_seconds: 0, last_health_check: DateTime.utc_now() } end defp close_connection(state, conn_id) do case state.connections[conn_id] do nil -> state connection -> if connection.socket do :gen_tcp.close(connection.socket) end new_connections = Map.delete(state.connections, conn_id) new_state = %{state | connections: new_connections} update_metrics(new_state, :connection_closed) end end defp perform_health_checks(state) do now = DateTime.utc_now() timeout_threshold = 60_000 # 60 seconds {healthy_connections, unhealthy_connections} = Enum.split_with(state.connections, fn {_id, conn} -> time_diff = DateTime.diff(now, conn.last_activity, :millisecond) time_diff < timeout_threshold end) # Close unhealthy connections for {conn_id, connection} <- unhealthy_connections do Logger.warning("Closing unhealthy connection: #{conn_id}") if connection.socket do :gen_tcp.close(connection.socket) end end new_connections = Map.new(healthy_connections) new_state = %{state | connections: new_connections} # Update health check metrics metrics = %{new_state.metrics | active_connections: map_size(new_connections), last_health_check: now } %{new_state | metrics: metrics} end defp handle_incoming_data(socket, data, state) do # Find connection by socket case find_connection_by_socket(state, socket) do {conn_id, connection} -> # Update received data metrics updated_metrics = %{connection.metrics | bytes_received: connection.metrics.bytes_received + byte_size(data) } updated_connection = %{connection | last_activity: DateTime.utc_now(), metrics: updated_metrics } new_state = put_in(state.connections[conn_id], updated_connection) new_state = update_metrics(new_state, :message_received, byte_size(data)) # TODO: Handle the actual data (forward to appropriate handler) Logger.debug("Received #{byte_size(data)} bytes on connection #{conn_id}") {:noreply, new_state} nil -> Logger.warning("Received data on unknown socket") {:noreply, state} end end defp handle_connection_closed(socket, state) do case find_connection_by_socket(state, socket) do {conn_id, _connection} -> Logger.info("Connection #{conn_id} closed by remote") new_state = close_connection(state, conn_id) {:noreply, new_state} nil -> {:noreply, state} end end defp handle_connection_error(socket, reason, state) do case find_connection_by_socket(state, socket) do {conn_id, _connection} -> Logger.error("Connection #{conn_id} error: #{inspect(reason)}") new_state = close_connection(state, conn_id) {:noreply, new_state} nil -> {:noreply, state} end end defp find_connection_by_socket(state, socket) do Enum.find(state.connections, fn {_id, conn} -> conn.socket == socket end) end # Fallback connection functions (when main transport is unavailable) defp fallback_connect(host, port, opts) do Logger.warning("Using fallback connection to #{host}:#{port}") # Attempt direct TCP connection without going through GenServer case establish_tcp_connection(host, port, opts) do {:ok, socket} -> connection_id = generate_connection_id(host, port) # Store fallback connection info in process dictionary Process.put({:fallback_connection, connection_id}, %{ socket: socket, host: host, port: port, created_at: DateTime.utc_now() }) {:ok, connection_id} {:error, reason} -> {:error, reason} end end end