perf: optimize prefix scan from O(N) to O(log N + K)

Switch :concord_store ETS table from :set to :ordered_set to enable
lexicographic range traversal. Add a server-side {:prefix_scan, prefix}
Ra query that uses :ets.select/2 with range guards to scan only the
matching key range, filtering expired entries inline. Wire Query.keys/1
to dispatch this query when the prefix: option is present, avoiding the
previous full-table load. Expose Concord.prefix_scan/2 as a public API.

Author: GSMLG-BOT

lib/concord.ex

@@ -516,6 +516,33 @@ defmodule Concord do
     end
   end
 
+  @doc """
+  Returns key-value pairs for all keys starting with the given prefix.
+
+  Uses an efficient server-side scan on the ordered ETS table, avoiding
+  loading all keys into memory. O(log N + K) where K is the number of
+  matching keys.
+
+  ## Options
+  - `:timeout` - Operation timeout in milliseconds (default: 5000)
+  - `:consistency` - Read consistency level (default: :leader)
+
+  ## Examples
+      iex> Concord.prefix_scan("user:")
+      {:ok, [{"user:1", %{name: "Alice"}}, {"user:2", %{name: "Bob"}}]}
+  """
+  def prefix_scan(prefix, opts \\ []) when is_binary(prefix) do
+    timeout = Keyword.get(opts, :timeout, @timeout)
+    consistency = Keyword.get(opts, :consistency, default_consistency())
+
+    case query({:prefix_scan, prefix}, timeout, consistency) do
+      {:ok, {{_index, _term}, query_result}, _} -> query_result
+      {:timeout, _} -> {:error, :timeout}
+      {:error, :noproc} -> {:error, :cluster_not_ready}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
   @doc """
   Stores multiple key-value pairs in the cluster atomically.
 

lib/concord/query.ex

@@ -95,7 +95,13 @@ defmodule Concord.Query do
   """
   @spec keys(query_opts()) :: {:ok, [key()]} | {:error, term()}
   def keys(opts \\ []) do
-    with {:ok, all_keys} <- get_all_keys() do
+    fetch_fn =
+      case Keyword.get(opts, :prefix) do
+        nil -> fn -> get_all_keys() end
+        prefix -> fn -> get_prefix_keys(prefix) end
+      end
+
+    with {:ok, all_keys} <- fetch_fn.() do
       filtered_keys =
         all_keys
         |> apply_key_filters(opts)
@@ -207,6 +213,17 @@ defmodule Concord.Query do
     end
   end
 
+  defp get_prefix_keys(prefix) do
+    case Concord.prefix_scan(prefix) do
+      {:ok, pairs} ->
+        keys = pairs |> Enum.map(fn {k, _v} -> k end) |> Enum.sort()
+        {:ok, keys}
+
+      error ->
+        error
+    end
+  end
+
   defp get_key_values(keys) do
     case Concord.get_many(keys) do
       {:ok, results} ->

lib/concord/state_machine.ex

@@ -79,7 +79,7 @@ defmodule Concord.StateMachine do
 
   @impl :ra_machine
   def init(_config) do
-    ensure_ets_table(:concord_store, [:set, :public, :named_table])
+    ensure_ets_table(:concord_store, [:ordered_set, :public, :named_table])
 
     {:concord_kv, default_state_fields()}
   end
@@ -743,6 +743,33 @@ defmodule Concord.StateMachine do
     {:ok, Map.new(results)}
   end
 
+  def query({:prefix_scan, prefix}, {:concord_kv, _data}) when is_binary(prefix) do
+    now = System.system_time(:second)
+    # Upper bound: any key starting with `prefix` is lexicographically < prefix <> <<255>>
+    # because byte values after the prefix will be < 255 for all normal string characters.
+    end_key = prefix <> <<255>>
+
+    match_spec = [
+      {{:"$1", :"$2"},
+       [{:">=", :"$1", prefix}, {:"<", :"$1", end_key}],
+       [{{:"$1", :"$2"}}]}
+    ]
+
+    results =
+      :ets.select(:concord_store, match_spec)
+      |> Enum.reduce([], fn {key, stored_data}, acc ->
+        case extract_value(stored_data) do
+          {value, expires_at} ->
+            if expired?(expires_at, now), do: acc, else: [{key, value} | acc]
+
+          _ ->
+            acc
+        end
+      end)
+
+    {:ok, results}
+  end
+
   def query(:stats, {:concord_kv, _data}) do
     info = :ets.info(:concord_store)
 
@@ -843,7 +870,7 @@ defmodule Concord.StateMachine do
 
       # V1/V2 legacy: bare list of KV tuples
       data when is_list(data) ->
-        ensure_ets_table(:concord_store)
+        ensure_ets_table(:concord_store, [:ordered_set, :public, :named_table])
         :ets.delete_all_objects(:concord_store)
         Enum.each(data, fn entry -> :ets.insert(:concord_store, entry) end)
     end
@@ -859,7 +886,7 @@ defmodule Concord.StateMachine do
 
   defp rebuild_all_ets_from_snapshot(data) do
     # Rebuild main KV store
-    ensure_ets_table(:concord_store)
+    ensure_ets_table(:concord_store, [:ordered_set, :public, :named_table])
     :ets.delete_all_objects(:concord_store)
 
     Enum.each(Map.get(data, :__kv_data__, []), fn entry ->

test/concord/query_test.exs

@@ -317,6 +317,46 @@ defmodule Concord.QueryTest do
     end
   end
 
+  describe "Concord.prefix_scan/2" do
+    test "returns key-value pairs for matching prefix" do
+      assert {:ok, pairs} = Concord.prefix_scan("user:")
+      assert length(pairs) == 4
+      keys = Enum.map(pairs, fn {k, _v} -> k end)
+      assert "user:1" in keys
+      assert "user:2" in keys
+      assert "user:10" in keys
+      assert "user:100" in keys
+    end
+
+    test "does not return keys outside the prefix" do
+      assert {:ok, pairs} = Concord.prefix_scan("product:")
+      keys = Enum.map(pairs, fn {k, _v} -> k end)
+      assert Enum.all?(keys, &String.starts_with?(&1, "product:"))
+      refute Enum.any?(keys, &String.starts_with?(&1, "user:"))
+    end
+
+    test "returns empty list when no keys match" do
+      assert {:ok, []} = Concord.prefix_scan("nonexistent:")
+    end
+
+    test "excludes expired keys" do
+      :ok = Concord.put_with_ttl("expiring:1", "value", 1)
+      :ok = Concord.put("expiring:2", "permanent")
+
+      # Both visible before expiry
+      assert {:ok, pairs} = Concord.prefix_scan("expiring:")
+      assert length(pairs) == 2
+
+      # Expire the key — TTL of 1s stores expires_at = now + 1.
+      # The expired? check is `now > expires_at` (strict), so we need at least 2s.
+      Process.sleep(2_100)
+      assert {:ok, pairs} = Concord.prefix_scan("expiring:")
+      keys = Enum.map(pairs, fn {k, _v} -> k end)
+      refute "expiring:1" in keys
+      assert "expiring:2" in keys
+    end
+  end
+
   describe "performance with larger datasets" do
     test "handles queries on larger datasets efficiently" do
       # Insert 100 additional keys