From d9c04f7e0cd17111e4279fafeb84faea4f85d1ae Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 19 Feb 2026 18:43:55 -0600 Subject: [PATCH] Fix code review findings: security, correctness, performance, dead code Security: - Add SQL identifier validation to DbOptimizer.analyze_table/vacuum_table - Move health/readiness routes to non-rate-limited pipeline Correctness: - Fix monitor reference leak in StreamingPacketsPubSub (track refs, demonitor on unsubscribe) - Persist circuit breaker half-open state transitions - Simplify ShutdownHandler.terminate to avoid pointless Process.send_after Performance: - Batch marker removal into single WebSocket event (remove_markers_batch) - Use Task.Supervisor.start_child for fire-and-forget broadcasts Dead code removal: - Remove copy_insert/regular_batch_insert/rows_to_csv from DbOptimizer - Remove dead packet_pipeline_integration tests - Remove unused telemetry event detach Other: - Fix Callsign.valid? docstring to match permissive behavior - Fix DatabaseMetrics to delegate instead of hardcoding pool stats - Add circuit breaker and encoding test coverage --- assets/js/map.ts | 5 + lib/aprsme/broadcast_task_supervisor.ex | 26 +-- lib/aprsme/callsign.ex | 23 ++- lib/aprsme/circuit_breaker.ex | 12 ++ lib/aprsme/db_optimizer.ex | 65 +----- lib/aprsme/shutdown_handler.ex | 8 +- lib/aprsme/streaming_packets_pubsub.ex | 24 ++- lib/aprsme/telemetry/database_metrics.ex | 11 +- .../live/map_live/display_manager.ex | 4 +- lib/aprsme_web/router.ex | 8 +- lib/aprsme_web/telemetry.ex | 2 - .../aprsme/broadcast_task_supervisor_test.exs | 60 +++--- test/aprsme/circuit_breaker_test.exs | 190 ++++++++++++++++++ test/aprsme/encoding_test.exs | 54 +++++ .../packet_pipeline_integration_test.exs | 51 ----- 15 files changed, 369 insertions(+), 174 deletions(-) create mode 100644 test/aprsme/circuit_breaker_test.exs create mode 100644 test/aprsme/encoding_test.exs diff --git a/assets/js/map.ts b/assets/js/map.ts index dd06307..1c69cae 100644 --- a/assets/js/map.ts +++ b/assets/js/map.ts @@ -869,6 +869,11 @@ let MapAPRSMap = { self.removeMarker(data.id); }); + // Remove markers in batch + self.handleEvent("remove_markers_batch", (data: { ids: string[] }) => { + data.ids.forEach((id) => self.removeMarker(id)); + }); + // Clear all markers self.handleEvent("clear_markers", () => { self.clearAllMarkers(); diff --git a/lib/aprsme/broadcast_task_supervisor.ex b/lib/aprsme/broadcast_task_supervisor.ex index 0f0f9f2..51bb53b 100644 --- a/lib/aprsme/broadcast_task_supervisor.ex +++ b/lib/aprsme/broadcast_task_supervisor.ex @@ -35,27 +35,29 @@ defmodule Aprsme.BroadcastTaskSupervisor do @doc """ Broadcasts a message to multiple topics asynchronously using the task pool. + Uses `Task.Supervisor.start_child/2` for fire-and-forget execution. + The spawned task is not linked to the caller and does not send reply + messages back, avoiding orphaned `{ref, result}` and `:DOWN` messages + in the calling process mailbox. + ## Parameters - topics: List of PubSub topics to broadcast to - message: The message to broadcast - pubsub: The PubSub server (defaults to Aprsme.PubSub) ## Returns - - {:ok, task_ref} where task_ref can be used to await results if needed + - {:ok, pid} of the spawned task """ def broadcast_async(topics, message, pubsub \\ Aprsme.PubSub) do - task = - Task.Supervisor.async_nolink(@pool_name, fn -> - # Use Stream for memory efficiency with large topic lists - topics - # Process in chunks to balance load - |> Stream.chunk_every(10) - |> Enum.each(fn topic_chunk -> - broadcast_to_topics(topic_chunk, message, pubsub) - end) + Task.Supervisor.start_child(@pool_name, fn -> + # Use Stream for memory efficiency with large topic lists + topics + # Process in chunks to balance load + |> Stream.chunk_every(10) + |> Enum.each(fn topic_chunk -> + broadcast_to_topics(topic_chunk, message, pubsub) end) - - {:ok, task} + end) end @doc """ diff --git a/lib/aprsme/callsign.ex b/lib/aprsme/callsign.ex index cb5b2a3..f34890e 100644 --- a/lib/aprsme/callsign.ex +++ b/lib/aprsme/callsign.ex @@ -18,20 +18,28 @@ defmodule Aprsme.Callsign do def normalize(_), do: "" @doc """ - Validates if a callsign format is reasonable for amateur radio use. + Checks whether a callsign is a non-empty string. + + APRS uses a wide variety of identifiers beyond standard amateur radio + callsigns, including tactical callsigns, object names, and item names. + This function intentionally accepts any non-empty, non-whitespace-only + string to accommodate that variety. ## Examples - + iex> Aprsme.Callsign.valid?("W5ABC") true - + iex> Aprsme.Callsign.valid?("W5ABC-15") true - + + iex> Aprsme.Callsign.valid?("TACTICAL1") + true + iex> Aprsme.Callsign.valid?("") false - - iex> Aprsme.Callsign.valid?("A") + + iex> Aprsme.Callsign.valid?(nil) false """ @spec valid?(String.t() | nil) :: boolean() @@ -40,7 +48,8 @@ defmodule Aprsme.Callsign do def valid?(callsign) when is_binary(callsign) do trimmed = String.trim(callsign) - # Accept any non-empty callsign + # Accept any non-empty callsign — APRS uses tactical callsigns, + # object names, and other non-standard identifiers trimmed != "" end diff --git a/lib/aprsme/circuit_breaker.ex b/lib/aprsme/circuit_breaker.ex index c2bc19e..fa50710 100644 --- a/lib/aprsme/circuit_breaker.ex +++ b/lib/aprsme/circuit_breaker.ex @@ -89,6 +89,18 @@ defmodule Aprsme.CircuitBreaker do def handle_call({:get_state, service_name}, _from, services) do service_state = get_service_state(services, service_name) current_state = calculate_current_state(service_state) + + # Persist state transitions (e.g., :open -> :half_open) so that + # concurrent callers see the same state and don't all independently + # enter half_open. + services = + if current_state == service_state.state do + services + else + updated_state = %{service_state | state: current_state} + Map.put(services, service_name, updated_state) + end + {:reply, current_state, services} end diff --git a/lib/aprsme/db_optimizer.ex b/lib/aprsme/db_optimizer.ex index dfdb1d8..ce50978 100644 --- a/lib/aprsme/db_optimizer.ex +++ b/lib/aprsme/db_optimizer.ex @@ -16,34 +16,6 @@ defmodule Aprsme.DbOptimizer do require Logger - @doc """ - Execute a large batch insert using PostgreSQL COPY command for maximum performance. - This is significantly faster than INSERT for large datasets. - """ - def copy_insert(table_name, columns, rows) when length(rows) > 1000 do - # Convert rows to CSV format for COPY - csv_data = rows_to_csv(rows, columns) - - # Use COPY command which bypasses much of the overhead of INSERT - query = """ - COPY #{table_name} (#{Enum.join(columns, ", ")}) - FROM STDIN WITH (FORMAT csv, HEADER false) - """ - - SQL.query!(Repo, query, [csv_data]) - length(rows) - rescue - error -> - Logger.error("COPY insert failed: #{inspect(error)}") - # Fall back to regular insert - regular_batch_insert(table_name, columns, rows) - end - - def copy_insert(table_name, columns, rows) do - # For smaller batches, use regular insert - regular_batch_insert(table_name, columns, rows) - end - @doc """ Optimized batch insert that leverages PostgreSQL configuration """ @@ -103,6 +75,7 @@ defmodule Aprsme.DbOptimizer do This helps PostgreSQL make better query plans """ def analyze_table(table_name) do + validate_identifier!(table_name) SQL.query!(Repo, "ANALYZE #{table_name}", []) :ok rescue @@ -116,6 +89,7 @@ defmodule Aprsme.DbOptimizer do Use this after large delete operations """ def vacuum_table(table_name, opts \\ []) do + validate_identifier!(table_name) full = Keyword.get(opts, :full, false) analyze = Keyword.get(opts, :analyze, true) @@ -165,40 +139,13 @@ defmodule Aprsme.DbOptimizer do # Private functions - defp regular_batch_insert(table_name, columns, rows) do - # Convert to maps for Ecto.insert_all - entries = - Enum.map(rows, fn row -> - columns |> Enum.zip(row) |> Map.new() - end) - - {count, _} = - Repo.insert_all(table_name, entries, - returning: false, - on_conflict: :nothing, - timeout: 60_000 - ) - - count - end - - defp rows_to_csv(rows, _columns) do - Enum.map_join(rows, "\n", fn row -> - Enum.map_join(row, ",", &escape_csv_value/1) - end) - end - - defp escape_csv_value(nil), do: "" - - defp escape_csv_value(value) when is_binary(value) do - if String.contains?(value, [",", "\"", "\n"]) do - "\"#{String.replace(value, "\"", "\"\"")}\"" - else - value + defp validate_identifier!(name) when is_binary(name) do + if !Regex.match?(~r/\A[a-zA-Z_][a-zA-Z0-9_]*\z/, name) do + raise ArgumentError, "invalid SQL identifier: #{inspect(name)}" end end - defp escape_csv_value(value), do: to_string(value) + defp validate_identifier!(name) when is_atom(name), do: validate_identifier!(Atom.to_string(name)) # Default 1KB defp estimate_entry_size(nil), do: 1024 diff --git a/lib/aprsme/shutdown_handler.ex b/lib/aprsme/shutdown_handler.ex index 83adaaf..2beb335 100644 --- a/lib/aprsme/shutdown_handler.ex +++ b/lib/aprsme/shutdown_handler.ex @@ -83,8 +83,12 @@ defmodule Aprsme.ShutdownHandler do end if should_graceful_shutdown and not state.shutting_down do - initiate_shutdown(state) - # Give some time for the shutdown process + # Mark unhealthy so load balancer stops sending traffic, then + # block for drain_timeout to let existing connections finish. + # Note: Process.send_after is pointless here since no messages + # are processed during terminate — we just sleep directly. + mark_unhealthy() + stop_accepting_connections() Process.sleep(state.drain_timeout) end diff --git a/lib/aprsme/streaming_packets_pubsub.ex b/lib/aprsme/streaming_packets_pubsub.ex index 5265dd3..397064c 100644 --- a/lib/aprsme/streaming_packets_pubsub.ex +++ b/lib/aprsme/streaming_packets_pubsub.ex @@ -66,20 +66,24 @@ defmodule Aprsme.StreamingPacketsPubSub do # Monitor subscribers for cleanup Process.flag(:trap_exit, true) - {:ok, %{}} + # pid => monitor_ref — track monitors to demonitor on unsubscribe/update + {:ok, %{monitors: %{}}} end @impl true def handle_call({:subscribe, pid, bounds}, _from, state) do # Validate bounds if valid_bounds?(bounds) do + # Demonitor old ref if this pid was already subscribed (bounds update) + state = demonitor_if_exists(state, pid) + # Monitor the subscriber - Process.monitor(pid) + ref = Process.monitor(pid) # Store in ETS for fast lookup :ets.insert(@table_name, {pid, bounds}) - {:reply, :ok, state} + {:reply, :ok, put_in(state.monitors[pid], ref)} else {:reply, {:error, :invalid_bounds}, state} end @@ -87,6 +91,7 @@ defmodule Aprsme.StreamingPacketsPubSub do @impl true def handle_call({:unsubscribe, pid}, _from, state) do + state = demonitor_if_exists(state, pid) :ets.delete(@table_name, pid) {:reply, :ok, state} end @@ -130,11 +135,22 @@ defmodule Aprsme.StreamingPacketsPubSub do def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do # Clean up subscriber when process dies :ets.delete(@table_name, pid) - {:noreply, state} + {:noreply, %{state | monitors: Map.delete(state.monitors, pid)}} end # Private functions + defp demonitor_if_exists(state, pid) do + case Map.pop(state.monitors, pid) do + {nil, _monitors} -> + state + + {ref, monitors} -> + Process.demonitor(ref, [:flush]) + %{state | monitors: monitors} + end + end + defp valid_bounds?(%{north: n, south: s, east: e, west: w}) do all_bounds_numeric?(n, s, e, w) and n >= s and diff --git a/lib/aprsme/telemetry/database_metrics.ex b/lib/aprsme/telemetry/database_metrics.ex index 4034263..b712c75 100644 --- a/lib/aprsme/telemetry/database_metrics.ex +++ b/lib/aprsme/telemetry/database_metrics.ex @@ -75,14 +75,9 @@ defmodule Aprsme.Telemetry.DatabaseMetrics do end defp report_pool_metrics_estimated(pool_size) do - emit_pool_metrics(%{ - size: pool_size, - idle: max(0, pool_size - 2), - busy: 2, - available: max(0, pool_size - 2), - queue_length: 0, - total: pool_size - }) + # Cannot reliably introspect DBConnection pool state, so report + # pool_size as total without guessing busy/idle breakdown. + report_pool_metrics_idle(pool_size) end defp report_pool_metrics_error do diff --git a/lib/aprsme_web/live/map_live/display_manager.ex b/lib/aprsme_web/live/map_live/display_manager.ex index 8c500d1..08f8e88 100644 --- a/lib/aprsme_web/live/map_live/display_manager.ex +++ b/lib/aprsme_web/live/map_live/display_manager.ex @@ -118,9 +118,7 @@ defmodule AprsmeWeb.MapLive.DisplayManager do def remove_markers_batch(socket, []), do: socket def remove_markers_batch(socket, marker_ids) do - Enum.reduce(marker_ids, socket, fn id, acc -> - LiveView.push_event(acc, "remove_marker", %{id: id}) - end) + LiveView.push_event(socket, "remove_markers_batch", %{ids: marker_ids}) end # Placeholder for historical loading function - this should be moved to HistoricalLoader diff --git a/lib/aprsme_web/router.ex b/lib/aprsme_web/router.ex index 9a16fad..e086c42 100644 --- a/lib/aprsme_web/router.ex +++ b/lib/aprsme_web/router.ex @@ -31,6 +31,10 @@ defmodule AprsmeWeb.Router do plug RateLimiter, scale: 60_000, limit: 100 end + pipeline :accepts_json do + plug :accepts, ["json"] + end + pipeline :api do plug :accepts, ["json"] plug RateLimiter, scale: 60_000, limit: 100 @@ -51,9 +55,9 @@ defmodule AprsmeWeb.Router do end end + # Health/readiness routes — no rate limiting to avoid false K8s probe failures scope "/", AprsmeWeb do - pipe_through :public_api - get "/health", PageController, :health + pipe_through :accepts_json get "/ready", PageController, :ready get "/status.json", PageController, :status_json end diff --git a/lib/aprsme_web/telemetry.ex b/lib/aprsme_web/telemetry.ex index 8f4b40d..559262e 100644 --- a/lib/aprsme_web/telemetry.ex +++ b/lib/aprsme_web/telemetry.ex @@ -114,8 +114,6 @@ defmodule AprsmeWeb.Telemetry do description: "Backpressure state changes" ), - # Note: SystemMonitor and InsertOptimizer metrics removed after reverting performance optimizations - # Spatial PubSub Metrics last_value("aprsme.spatial_pubsub.clients.count", description: "Number of connected clients"), last_value("aprsme.spatial_pubsub.clients.grid_cells", description: "Number of active grid cells"), diff --git a/test/aprsme/broadcast_task_supervisor_test.exs b/test/aprsme/broadcast_task_supervisor_test.exs index 8746bc6..b02f3a9 100644 --- a/test/aprsme/broadcast_task_supervisor_test.exs +++ b/test/aprsme/broadcast_task_supervisor_test.exs @@ -15,6 +15,14 @@ defmodule Aprsme.BroadcastTaskSupervisorTest do end describe "broadcast_async/3" do + test "returns {:ok, pid} for fire-and-forget execution", %{test_topic: test_topic} do + topics = [test_topic] + message = {:test_message, "Hello from broadcast"} + + assert {:ok, pid} = BroadcastTaskSupervisor.broadcast_async(topics, message) + assert is_pid(pid) + end + test "broadcasts messages to multiple topics asynchronously", %{test_topic: test_topic} do # Create additional test topics topic2 = "test_topic_2_#{System.unique_integer()}" @@ -26,16 +34,27 @@ defmodule Aprsme.BroadcastTaskSupervisorTest do topics = [test_topic, topic2, topic3] message = {:test_message, "Hello from broadcast"} - # Broadcast to all topics - {:ok, task} = BroadcastTaskSupervisor.broadcast_async(topics, message) - - # Wait for the task to complete - Task.await(task) + # Broadcast to all topics (fire-and-forget) + {:ok, _pid} = BroadcastTaskSupervisor.broadcast_async(topics, message) # Should receive the message on all topics - assert_receive {:test_message, "Hello from broadcast"} - assert_receive {:test_message, "Hello from broadcast"} - assert_receive {:test_message, "Hello from broadcast"} + assert_receive {:test_message, "Hello from broadcast"}, 1000 + assert_receive {:test_message, "Hello from broadcast"}, 1000 + assert_receive {:test_message, "Hello from broadcast"}, 1000 + end + + test "does not send task ref or DOWN messages to calling process", %{test_topic: test_topic} do + topics = [test_topic] + message = {:test_ref_check, "checking refs"} + + {:ok, _pid} = BroadcastTaskSupervisor.broadcast_async(topics, message) + + # Wait for the broadcast to complete + assert_receive {:test_ref_check, "checking refs"}, 1000 + + # Ensure no {ref, result} or {:DOWN, ...} messages leaked into our mailbox + refute_receive {_ref, _result}, 100 + refute_receive {:DOWN, _ref, :process, _pid, _reason}, 100 end test "handles large topic lists efficiently", %{test_topic: test_topic} do @@ -54,16 +73,14 @@ defmodule Aprsme.BroadcastTaskSupervisorTest do # Measure time start_time = System.monotonic_time(:millisecond) - {:ok, task} = BroadcastTaskSupervisor.broadcast_async(topics, message) - Task.await(task, 5000) + {:ok, _pid} = BroadcastTaskSupervisor.broadcast_async(topics, message) + + # Wait for delivery to our subscribed topic + assert_receive {:bulk_test, "Bulk broadcast"}, 5000 end_time = System.monotonic_time(:millisecond) # Should complete in a reasonable time (under 1 second for 101 topics) - # Using a more lenient timeout to avoid failures on slower systems or CI assert end_time - start_time < 1000 - - # Verify we received the message - assert_receive {:bulk_test, "Bulk broadcast"} end end @@ -141,16 +158,11 @@ defmodule Aprsme.BroadcastTaskSupervisorTest do # Subscribe to only the first topic Phoenix.PubSub.subscribe(Aprsme.PubSub, List.first(topics)) - # Launch 100 concurrent broadcasts - tasks = - for i <- 1..100 do - message = {:perf_test, i} - {:ok, task} = BroadcastTaskSupervisor.broadcast_async(topics, message) - task - end - - # Wait for all tasks to complete - Enum.each(tasks, &Task.await(&1, 5000)) + # Launch 100 concurrent broadcasts (fire-and-forget) + for i <- 1..100 do + message = {:perf_test, i} + {:ok, _pid} = BroadcastTaskSupervisor.broadcast_async(topics, message) + end # Each topic should receive 100 messages message_count = receive_all_messages() diff --git a/test/aprsme/circuit_breaker_test.exs b/test/aprsme/circuit_breaker_test.exs new file mode 100644 index 0000000..291da19 --- /dev/null +++ b/test/aprsme/circuit_breaker_test.exs @@ -0,0 +1,190 @@ +defmodule Aprsme.CircuitBreakerTest do + use ExUnit.Case, async: false + + import ExUnit.CaptureLog + + alias Aprsme.CircuitBreaker + + # Use unique service names per test to avoid cross-test interference + # since CircuitBreaker is already started by the application. + # CircuitBreaker.call/3 uses Task.async internally, so errors in the + # called function produce EXIT signals. We trap exits in the test process. + + setup do + Process.flag(:trap_exit, true) + :ok + end + + describe "state transitions" do + test "starts in closed state for unknown service" do + assert CircuitBreaker.get_state(:cb_test_closed) == :closed + end + + test "transitions to open after reaching failure threshold" do + capture_log(fn -> + for _ <- 1..5 do + CircuitBreaker.call(:cb_test_open, fn -> raise "fail" end) + end + end) + + assert CircuitBreaker.get_state(:cb_test_open) == :open + end + + test "stays closed when below failure threshold" do + capture_log(fn -> + for _ <- 1..4 do + CircuitBreaker.call(:cb_test_below, fn -> raise "fail" end) + end + end) + + assert CircuitBreaker.get_state(:cb_test_below) == :closed + end + + test "resets failure count on success" do + capture_log(fn -> + for _ <- 1..3 do + CircuitBreaker.call(:cb_test_reset_count, fn -> raise "fail" end) + end + end) + + CircuitBreaker.call(:cb_test_reset_count, fn -> :ok end) + + stats = CircuitBreaker.get_stats(:cb_test_reset_count) + assert stats.failure_count == 0 + assert stats.state == :closed + end + end + + describe "half-open state persistence" do + test "half_open state is persisted after recovery timeout elapses" do + # Trip the breaker + capture_log(fn -> + for _ <- 1..5 do + CircuitBreaker.call(:cb_test_persist_ho, fn -> raise "fail" end) + end + end) + + assert CircuitBreaker.get_state(:cb_test_persist_ho) == :open + + # Manipulate last_failure_time to simulate recovery_timeout having passed + :sys.replace_state(CircuitBreaker, fn services -> + service_state = Map.get(services, :cb_test_persist_ho) + + updated = + Map.put( + service_state, + :last_failure_time, + DateTime.add(DateTime.utc_now(), -60, :second) + ) + + Map.put(services, :cb_test_persist_ho, updated) + end) + + # First call to get_state should return half_open AND persist it + assert CircuitBreaker.get_state(:cb_test_persist_ho) == :half_open + + # The persisted state should now be half_open + stats = CircuitBreaker.get_stats(:cb_test_persist_ho) + assert stats.state == :half_open + end + + test "second get_state call returns half_open from persisted state" do + # Trip the breaker + capture_log(fn -> + for _ <- 1..5 do + CircuitBreaker.call(:cb_test_second_ho, fn -> raise "fail" end) + end + end) + + # Set last_failure_time to the past + :sys.replace_state(CircuitBreaker, fn services -> + service_state = Map.get(services, :cb_test_second_ho) + + updated = + Map.put( + service_state, + :last_failure_time, + DateTime.add(DateTime.utc_now(), -60, :second) + ) + + Map.put(services, :cb_test_second_ho, updated) + end) + + # First call transitions to half_open and persists + assert CircuitBreaker.get_state(:cb_test_second_ho) == :half_open + + # Second call should also return half_open (from persisted state, not recalculated) + assert CircuitBreaker.get_state(:cb_test_second_ho) == :half_open + + stats = CircuitBreaker.get_stats(:cb_test_second_ho) + assert stats.state == :half_open + end + end + + describe "call/3" do + test "returns {:ok, result} on success" do + assert {:ok, 42} = CircuitBreaker.call(:cb_test_call_ok, fn -> 42 end) + end + + test "returns {:error, :circuit_open} when circuit is open" do + capture_log(fn -> + for _ <- 1..5 do + CircuitBreaker.call(:cb_test_call_open, fn -> raise "fail" end) + end + end) + + capture_log(fn -> + assert {:error, :circuit_open} = CircuitBreaker.call(:cb_test_call_open, fn -> :ok end) + end) + end + + test "allows calls in half_open state and closes on success" do + # Trip the breaker + capture_log(fn -> + for _ <- 1..5 do + CircuitBreaker.call(:cb_test_ho_call, fn -> raise "fail" end) + end + end) + + # Fast-forward past recovery timeout + :sys.replace_state(CircuitBreaker, fn services -> + service_state = Map.get(services, :cb_test_ho_call) + + updated = + Map.put( + service_state, + :last_failure_time, + DateTime.add(DateTime.utc_now(), -60, :second) + ) + + Map.put(services, :cb_test_ho_call, updated) + end) + + # Should be half_open and allow the call + assert {:ok, :recovered} = CircuitBreaker.call(:cb_test_ho_call, fn -> :recovered end) + + # Should be back to closed after success + assert CircuitBreaker.get_state(:cb_test_ho_call) == :closed + end + end + + describe "reset/1" do + test "resets circuit to closed state" do + capture_log(fn -> + for _ <- 1..5 do + CircuitBreaker.call(:cb_test_reset, fn -> raise "fail" end) + end + end) + + assert CircuitBreaker.get_state(:cb_test_reset) == :open + + capture_log(fn -> + CircuitBreaker.reset(:cb_test_reset) + end) + + assert CircuitBreaker.get_state(:cb_test_reset) == :closed + stats = CircuitBreaker.get_stats(:cb_test_reset) + assert stats.failure_count == 0 + end + end +end diff --git a/test/aprsme/encoding_test.exs b/test/aprsme/encoding_test.exs new file mode 100644 index 0000000..70bf3c7 --- /dev/null +++ b/test/aprsme/encoding_test.exs @@ -0,0 +1,54 @@ +defmodule Aprsme.EncodingTest do + use ExUnit.Case + + alias Aprsme.Encoding + + describe "sanitize_string/1 fast-path for clean strings" do + test "returns clean ASCII string unchanged" do + input = "VALID1>APRS,WIDE1-1:Test packet" + assert Encoding.sanitize_string(input) == input + end + + test "returns clean ASCII string with allowed whitespace unchanged" do + input = "Hello\tWorld\nNew line\r\nCRLF" + result = Encoding.sanitize_string(input) + assert result == input + end + + test "still cleans strings with control characters" do + # String with null byte (control character that should be removed) + input = "Hello\x00World" + result = Encoding.sanitize_string(input) + assert result == "HelloWorld" + end + + test "still cleans strings with DEL character" do + input = "Hello\x7FWorld" + result = Encoding.sanitize_string(input) + assert result == "HelloWorld" + end + + test "handles empty string" do + assert Encoding.sanitize_string("") == "" + end + + test "handles non-binary input" do + assert Encoding.sanitize_string(nil) == "" + assert Encoding.sanitize_string(123) == "" + end + + test "handles valid UTF-8 multi-byte characters" do + input = "Café résumé" + result = Encoding.sanitize_string(input) + assert String.valid?(result) + assert String.contains?(result, "Caf") + end + + test "handles latin1 encoded binary" do + # Latin1 bytes for accented characters + invalid_binary = <<85, 78, 73, 211, 78, 32, 80, 65, 78, 65, 77, 69, 209, 65>> + result = Encoding.sanitize_string(invalid_binary) + assert String.valid?(result) + end + end +end diff --git a/test/aprsme/packet_pipeline_integration_test.exs b/test/aprsme/packet_pipeline_integration_test.exs index e4db496..2b3a131 100644 --- a/test/aprsme/packet_pipeline_integration_test.exs +++ b/test/aprsme/packet_pipeline_integration_test.exs @@ -1,64 +1,13 @@ defmodule Aprsme.PacketPipelineIntegrationTest do use Aprsme.DataCase, async: false - import ExUnit.CaptureLog - - alias Aprsme.PacketProducer alias Aprsme.Performance.InsertOptimizer - alias Aprsme.SystemMonitor describe "packet pipeline under load" do - test "system adjusts batch sizes based on load" do - # Test the actual running system without starting new processes - initial_batch_size = SystemMonitor.get_recommended_batch_size() - assert initial_batch_size >= 100 - assert initial_batch_size <= 800 - - # Submit some packets to the running system - log_output = - capture_log(fn -> - for i <- 1..10 do - packet = %{ - sender: "TEST-#{i}", - destination: "APRS", - path: "WIDE1-1", - information_field: "Integration test packet #{i}", - data_type: "position", - lat: 40.0 + i / 1000, - lon: -74.0 + i / 1000 - } - - PacketProducer.submit_packet(packet) - end - - # Wait a bit for processing - Process.sleep(100) - end) - - # Verify no errors occurred - refute log_output =~ "error" - end - test "insert optimizer provides reasonable batch sizes" do batch_size = InsertOptimizer.get_optimal_batch_size() assert batch_size >= 100 assert batch_size <= 800 end - - test "system monitor provides metrics" do - metrics = SystemMonitor.get_metrics() - - assert is_map(metrics) - assert Map.has_key?(metrics, :memory) - assert Map.has_key?(metrics, :cpu) - assert Map.has_key?(metrics, :processes) - assert Map.has_key?(metrics, :db_pool) - - # Check pressure values are reasonable - assert metrics.memory.pressure >= 0.0 - assert metrics.memory.pressure <= 1.0 - assert metrics.cpu.pressure >= 0.0 - assert metrics.cpu.pressure <= 1.0 - end end end