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
This commit is contained in:
parent
0084539bb1
commit
d9c04f7e0c
15 changed files with 369 additions and 174 deletions
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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 """
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"),
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
190
test/aprsme/circuit_breaker_test.exs
Normal file
190
test/aprsme/circuit_breaker_test.exs
Normal file
|
|
@ -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
|
||||
54
test/aprsme/encoding_test.exs
Normal file
54
test/aprsme/encoding_test.exs
Normal file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue