Release v0.7.2: Replace global locks with ETS-based locking for improved concurrency

This release replaces :global.trans-based locking with a custom
ETS-based locking mechanism in ConcurrencyGate and State modules,
improving performance and reliability under high concurrency.

Key improvements:

- ETS-based locking eliminates dependency on distributed Erlang's
  global name server, which can introduce latency and contention in
  single-node deployments

- Dedicated lock tables (gemini_concurrency_locks and
  gemini_rate_limit_locks) track lock ownership per model/key with
  automatic cleanup of dead lock holders

- Lock acquisition uses insert_new for atomic compare-and-set
  semantics; contending processes spin-wait with 5ms intervals

- Dead lock holder cleanup uses delete_object to prevent TOCTOU races
  when reclaiming locks from terminated processes

- Lock release is explicit and guaranteed via try/after blocks around
  critical sections

Implementation details:

- ConcurrencyGate.with_lock/2 acquires model-specific lock before
  permit operations (acquire, release, holder tracking)

- State.with_lock/2 acquires budget-key-specific lock before
  reservation operations (try_reserve_budget, reconcile_reservation)

- Both modules create secondary ETS tables with write_concurrency: true
  for lock storage during lazy initialization

- Lock acquisition retries indefinitely with Process.alive? checks to
  detect and clean up stale lock holders

- Test coverage expanded with high-iteration concurrency repro script
  (repro_concurrency_gate.exs) demonstrating no permit overlap across
  500 iterations with 8 concurrent tasks

Performance characteristics:

- Reduced latency for uncontended lock acquisition (ETS insert_new vs
  global name server round-trip)

- Bounded spin-wait interval (5ms) prevents CPU saturation during lock
  contention

- Automatic lock cleanup eliminates need for manual lock recovery after
  process crashes

Breaking changes: None

All existing tests pass; new test validates atomic reservation with
improved non_blocking mode timing to avoid flaky race conditions.

Author: nshkrdotcom

CHANGELOG.md

@@ -5,6 +5,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.7.2] - 2025-12-06
+
+### Fixed
+- **Rate limiter race condition**: Replaced `:global.trans/2` with ETS-based spinlock using `:ets.insert_new/2` for proper single-node mutex semantics in both `ConcurrencyGate` and `State` modules
+- **TOCTOU race in lock cleanup**: Use `:ets.delete_object/2` instead of `:ets.delete/2` to atomically delete only if PID still matches, preventing lock theft
+- **ETS table options**: Changed lock tables to use `write_concurrency: true` instead of `read_concurrency: true` for write-heavy workloads
+- **Test synchronization**: Removed flaky `Process.sleep` from atomic reservation test; now awaits non-blocking task2 before releasing task1 for deterministic synchronization
+
+### Changed
+- Lock acquisition retry sleep increased from 1ms to 5ms to reduce CPU usage under contention
+
 ## [0.7.1] - 2025-12-05
 
 ### Added

README.md

@@ -51,7 +51,7 @@ Add `gemini` to your list of dependencies in `mix.exs`:
 ```elixir
 def deps do
   [
-    {:gemini_ex, "~> 0.7.1"}
+    {:gemini_ex, "~> 0.7.2"}
   ]
 end
 ```

lib/gemini/rate_limiter/concurrency_gate.ex

@@ -16,6 +16,7 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   alias Gemini.RateLimiter.Config
 
   @ets_table :gemini_concurrency_permits
+  @lock_table :gemini_concurrency_locks
   @adaptive_backoff_factor 0.75
   @adaptive_raise_amount 1
 
@@ -37,12 +38,12 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   """
   @spec init() :: :ok
   def init do
-    ensure_table_exists()
+    ensure_tables_exist()
     :ok
   end
 
   # Lazy initialization - ensures table exists before any operation
-  defp ensure_table_exists do
+  defp ensure_tables_exist do
     case :ets.whereis(@ets_table) do
       :undefined ->
         try do
@@ -54,6 +55,18 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
       _ref ->
         :ok
     end
+
+    case :ets.whereis(@lock_table) do
+      :undefined ->
+        try do
+          :ets.new(@lock_table, [:named_table, :public, :set, write_concurrency: true])
+        catch
+          :error, :badarg -> :ok
+        end
+
+      _ref ->
+        :ok
+    end
   end
 
   @doc """
@@ -76,7 +89,7 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   """
   @spec acquire(model_key(), Config.t()) :: :ok | {:error, atom()}
   def acquire(model, %Config{} = config) do
-    ensure_table_exists()
+    ensure_tables_exist()
 
     unless Config.concurrency_enabled?(config) do
       {:error, :concurrency_disabled}
@@ -93,7 +106,7 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   """
   @spec release(model_key()) :: :ok
   def release(model) do
-    ensure_table_exists()
+    ensure_tables_exist()
 
     with_lock(model, fn ->
       case :ets.lookup(@ets_table, model) do
@@ -118,7 +131,7 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   """
   @spec signal_429(model_key(), Config.t()) :: :ok
   def signal_429(model, %Config{adaptive_concurrency: true} = config) do
-    ensure_table_exists()
+    ensure_tables_exist()
 
     case :ets.lookup(@ets_table, model) do
       [{^model, state}] ->
@@ -149,7 +162,7 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   """
   @spec signal_success(model_key(), Config.t()) :: :ok
   def signal_success(model, %Config{adaptive_concurrency: true} = config) do
-    ensure_table_exists()
+    ensure_tables_exist()
 
     case :ets.lookup(@ets_table, model) do
       [{^model, state}] when not is_nil(state.adaptive_max) ->
@@ -173,7 +186,7 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   """
   @spec get_state(model_key()) :: permit_state() | nil
   def get_state(model) do
-    ensure_table_exists()
+    ensure_tables_exist()
 
     case :ets.lookup(@ets_table, model) do
       [{^model, state}] -> state
@@ -186,7 +199,7 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   """
   @spec available_permits(model_key(), Config.t()) :: non_neg_integer()
   def available_permits(model, %Config{} = config) do
-    ensure_table_exists()
+    ensure_tables_exist()
 
     case :ets.lookup(@ets_table, model) do
       [{^model, state}] ->
@@ -217,6 +230,11 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
       _ref -> :ets.delete_all_objects(@ets_table)
     end
 
+    case :ets.whereis(@lock_table) do
+      :undefined -> :ok
+      _ref -> :ets.delete_all_objects(@lock_table)
+    end
+
     :ok
   end
 
@@ -335,7 +353,7 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   end
 
   def handle_holder_down(model, holder_pid) do
-    ensure_table_exists()
+    ensure_tables_exist()
 
     with_lock(model, fn ->
       case :ets.lookup(@ets_table, model) do
@@ -371,6 +389,45 @@ defmodule Gemini.RateLimiter.ConcurrencyGate do
   end
 
   defp with_lock(model, fun) do
-    :global.trans({@ets_table, model}, fun)
+    acquire_lock(model)
+
+    try do
+      fun.()
+    after
+      release_lock(model)
+    end
+  end
+
+  defp acquire_lock(model) do
+    ensure_tables_exist()
+
+    case :ets.insert_new(@lock_table, {model, self()}) do
+      true ->
+        :ok
+
+      false ->
+        cleanup_dead_lock_holder(model)
+        Process.sleep(5)
+        acquire_lock(model)
+    end
+  end
+
+  defp release_lock(model) do
+    :ets.delete(@lock_table, model)
+    :ok
+  end
+
+  defp cleanup_dead_lock_holder(model) do
+    case :ets.lookup(@lock_table, model) do
+      [{^model, pid}] ->
+        unless Process.alive?(pid) do
+          # Use delete_object to atomically delete only if PID still matches
+          # This prevents TOCTOU race where another process acquired the lock
+          :ets.delete_object(@lock_table, {model, pid})
+        end
+
+      _ ->
+        :ok
+    end
   end
 end

lib/gemini/rate_limiter/state.ex

@@ -36,6 +36,7 @@ defmodule Gemini.RateLimiter.State do
         }
 
   @ets_table :gemini_rate_limit_state
+  @lock_table :gemini_rate_limit_locks
   @default_window_duration_ms 60_000
   @default_location "us-central1"
 
@@ -52,7 +53,7 @@ defmodule Gemini.RateLimiter.State do
     :ok
   end
 
-  # Lazy initialization - ensures table exists before any operation
+  # Lazy initialization - ensures tables exist before any operation
   defp ensure_table_exists do
     case :ets.whereis(@ets_table) do
       :undefined ->
@@ -66,6 +67,18 @@ defmodule Gemini.RateLimiter.State do
       _ref ->
         :ok
     end
+
+    case :ets.whereis(@lock_table) do
+      :undefined ->
+        try do
+          :ets.new(@lock_table, [:named_table, :public, :set, write_concurrency: true])
+        catch
+          :error, :badarg -> :ok
+        end
+
+      _ref ->
+        :ok
+    end
   end
 
   @doc """
@@ -181,7 +194,7 @@ defmodule Gemini.RateLimiter.State do
     now = DateTime.utc_now()
     window_duration = Keyword.get(opts, :window_duration_ms, @default_window_duration_ms)
 
-    :global.trans({@ets_table, {:budget, key}}, fn ->
+    with_lock({:budget, key}, fn ->
       current_window = current_or_new_window(key, now, window_duration)
 
       updated =
@@ -273,7 +286,7 @@ defmodule Gemini.RateLimiter.State do
     window_duration = Keyword.get(opts, :window_duration_ms, @default_window_duration_ms)
     reserved_tokens = scaled_tokens(estimated_total_tokens, multiplier)
 
-    :global.trans({@ets_table, {:budget, key}}, fn ->
+    with_lock({:budget, key}, fn ->
       now = DateTime.utc_now()
       window = current_or_new_window(key, now, window_duration)
       window_end = DateTime.add(window.window_start, window.window_duration_ms, :millisecond)
@@ -334,7 +347,7 @@ defmodule Gemini.RateLimiter.State do
     actual_input = Map.get(usage_map || %{}, :input_tokens, 0)
     actual_output = Map.get(usage_map || %{}, :output_tokens, 0)
 
-    :global.trans({@ets_table, {:budget, key}}, fn ->
+    with_lock({:budget, key}, fn ->
       now = DateTime.utc_now()
       window = current_or_new_window(key, now, window_duration)
 
@@ -369,6 +382,11 @@ defmodule Gemini.RateLimiter.State do
       _ref -> :ets.delete_all_objects(@ets_table)
     end
 
+    case :ets.whereis(@lock_table) do
+      :undefined -> :ok
+      _ref -> :ets.delete_all_objects(@lock_table)
+    end
+
     :ok
   end
 
@@ -458,4 +476,48 @@ defmodule Gemini.RateLimiter.State do
     rate_limiter_config = Application.get_env(:gemini_ex, :rate_limiter, [])
     Keyword.get(rate_limiter_config, :default_retry_delay_ms, 60_000)
   end
+
+  # ETS-based locking to replace :global.trans
+  defp with_lock(lock_key, fun) do
+    acquire_lock(lock_key)
+
+    try do
+      fun.()
+    after
+      release_lock(lock_key)
+    end
+  end
+
+  defp acquire_lock(lock_key) do
+    ensure_table_exists()
+
+    case :ets.insert_new(@lock_table, {lock_key, self()}) do
+      true ->
+        :ok
+
+      false ->
+        cleanup_dead_lock_holder(lock_key)
+        Process.sleep(5)
+        acquire_lock(lock_key)
+    end
+  end
+
+  defp release_lock(lock_key) do
+    :ets.delete(@lock_table, lock_key)
+    :ok
+  end
+
+  defp cleanup_dead_lock_holder(lock_key) do
+    case :ets.lookup(@lock_table, lock_key) do
+      [{^lock_key, pid}] ->
+        unless Process.alive?(pid) do
+          # Use delete_object to atomically delete only if PID still matches
+          # This prevents TOCTOU race where another process acquired the lock
+          :ets.delete_object(@lock_table, {lock_key, pid})
+        end
+
+      _ ->
+        :ok
+    end
+  end
 end

mix.exs

@@ -1,7 +1,7 @@
 defmodule Gemini.MixProject do
   use Mix.Project
 
-  @version "0.7.1"
+  @version "0.7.2"
   @source_url "https://github.com/nshkrdotcom/gemini_ex"
 
   def project do

repro_concurrency_gate.exs

@@ -0,0 +1,57 @@
+alias Gemini.RateLimiter.{Manager, ConcurrencyGate}
+
+defmodule ConcurrencyGateRepro do
+  def run(iterations \\ 500, tasks_per_round \\ 8, sleep_ms \\ 10) do
+    ConcurrencyGate.init()
+    Manager.reset_all()
+
+    for i <- 1..iterations do
+      counter = :atomics.new(1, [])
+      max_seen = :atomics.new(1, [])
+      model = "repro-gate-#{i}-#{System.unique_integer([:positive])}"
+
+      request_fn = fn ->
+        current = :atomics.add_get(counter, 1, 1)
+        bump_max(max_seen, current)
+        Process.sleep(sleep_ms)
+        :atomics.add(counter, 1, -1)
+        {:ok, :done}
+      end
+
+      tasks =
+        for _ <- 1..tasks_per_round do
+          Task.async(fn ->
+            Manager.execute(request_fn, model, max_concurrency_per_model: 1)
+          end)
+        end
+
+      results = Task.await_many(tasks, 5_000)
+
+      unless Enum.all?(results, &match?({:ok, _}, &1)),
+        do: raise("unexpected result: #{inspect(results)}")
+
+      peak = :atomics.get(max_seen, 1)
+
+      if peak > 1 do
+        IO.puts("FAIL iteration=#{i} peak=#{peak} tasks=#{tasks_per_round} sleep_ms=#{sleep_ms}")
+        System.halt(1)
+      end
+    end
+
+    IO.puts(
+      "PASS no overlaps after #{iterations} iterations (tasks=#{tasks_per_round}, sleep_ms=#{sleep_ms})"
+    )
+  end
+
+  defp bump_max(ref, value) do
+    current = :atomics.get(ref, 1)
+
+    if value > current do
+      :atomics.compare_exchange(ref, 1, current, value)
+    end
+
+    :ok
+  end
+end
+
+ConcurrencyGateRepro.run()