[#7] Adapter-specific transaction support with ETS-based locking

Author: cabol

.tool-versions

@@ -1,2 +1,2 @@
 elixir 1.19.4-otp-28
-erlang 28.2
+erlang 28.3

CHANGELOG.md

@@ -1,5 +1,25 @@
 # Changelog
 
+All notable changes to this project will be documented in this file.
+
+This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [v3.0.0-rc.3](https://github.com/elixir-nebulex/nebulex_local/tree/v3.0.0-rc.3) (2025-12-28)
+> [Full Changelog](https://github.com/elixir-nebulex/nebulex_local/compare/v3.0.0-rc.2...v3.0.0-rc.3)
+
+### Enhancements
+
+- [Nebulex.Adapters.Local] Implemented adapter-specific transaction support using
+  `Nebulex.Locks`, a lightweight ETS-based locking mechanism optimized for
+  single-node scenarios. This replaces the previous reliance on `:global` for
+  distributed locking, providing significantly better performance for local
+  cache transactions while maintaining the same public API. The locks manager
+  can be customized via the new `:lock_opts` configuration option (e.g.,
+  `:cleanup_interval`, `:cleanup_batch_size`). This change aligns with the
+  removal of the default transaction implementation from Nebulex core, allowing
+  adapters to provide implementations tailored to their specific needs.
+  [#7](https://github.com/elixir-nebulex/nebulex_local/issues/7).
+
 ## [v3.0.0-rc.2](https://github.com/elixir-nebulex/nebulex_local/tree/v3.0.0-rc.2) (2025-12-07)
 > [Full Changelog](https://github.com/elixir-nebulex/nebulex_local/compare/v3.0.0-rc.1...v3.0.0-rc.2)
 

lib/nebulex/adapters/local.ex

@@ -27,8 +27,10 @@ defmodule Nebulex.Adapters.Local do
       (see `Nebulex.Adapters.Local.Generation`).
     * Sharding - For intensive workloads, the Cache may also be partitioned
       (by using `:shards` backend and specifying the `:partitions` option).
-    * Support for transactions via Erlang global name registration facility.
-      See `Nebulex.Adapter.Transaction`.
+    * Support for transactions via `Nebulex.Locks`, a lightweight ETS-based
+      locking mechanism optimized for single-node scenarios. Provides atomic
+      lock acquisition, deadlock prevention, and automatic stale lock cleanup.
+      See `Nebulex.Locks` and `Nebulex.Adapter.Transaction`.
     * Support for stats.
     * Automatic retry logic for handling race conditions during garbage
       collection (see [Concurrency and resilience](#module-concurrency-and-resilience)).
@@ -745,11 +747,19 @@ defmodule Nebulex.Adapters.Local do
 
   ## Transaction API
 
-  This adapter inherits the default implementation provided by
-  `Nebulex.Adapter.Transaction`. Therefore, the `transaction` command accepts
-  the following options:
+  This adapter implements the `Nebulex.Adapter.Transaction` behaviour using
+  `Nebulex.Locks`, a lightweight ETS-based locking mechanism optimized for
+  single-node scenarios. This implementation provides significantly better
+  performance compared to distributed locking mechanisms (e.g., `:global`)
+  while maintaining the same transaction API.
 
-  #{Nebulex.Adapter.Transaction.Options.options_docs()}
+  The `transaction` command accepts the following options:
+
+  #{Nebulex.Locks.Options.options_docs()}
+
+  The locks manager can be customized via the `:lock_opts` configuration
+  option when starting the cache. See the configuration options above for
+  more details.
 
   ## Extended API (extra functions)
 
@@ -775,9 +785,7 @@ defmodule Nebulex.Adapters.Local do
   @behaviour Nebulex.Adapter
   @behaviour Nebulex.Adapter.KV
   @behaviour Nebulex.Adapter.Queryable
-
-  # Inherit default transaction implementation
-  use Nebulex.Adapter.Transaction
+  @behaviour Nebulex.Adapter.Transaction
 
   # Inherit default info implementation
   use Nebulex.Adapters.Common.Info
@@ -790,6 +798,8 @@ defmodule Nebulex.Adapters.Local do
 
   alias Nebulex.Adapters.Common.Info.Stats
   alias Nebulex.Adapters.Local.{Backend, Generation, Metadata}
+  alias Nebulex.Locks
+  alias Nebulex.Locks.Options, as: LockOptions
   alias Nebulex.Time
 
   ## Types & Internal definitions
@@ -1315,6 +1325,61 @@ defmodule Nebulex.Adapters.Local do
     %{total: max_size, used: mem_size}
   end
 
+  ## Nebulex.Adapter.Transaction
+
+  @impl true
+  def transaction(%{cache: cache, pid: pid} = adapter_meta, fun, opts) do
+    opts = LockOptions.validate!(opts)
+
+    adapter_meta
+    |> do_in_transaction?()
+    |> do_transaction(
+      pid,
+      adapter_meta[:name] || cache,
+      adapter_meta.meta_tab,
+      opts,
+      fun
+    )
+  end
+
+  @impl true
+  def in_transaction?(adapter_meta, _opts) do
+    wrap_ok do_in_transaction?(adapter_meta)
+  end
+
+  defp do_in_transaction?(%{pid: pid}) do
+    !!Process.get({pid, self()})
+  end
+
+  defp do_transaction(true, _pid, _name, _meta_tab, _opts, fun) do
+    {:ok, fun.()}
+  end
+
+  defp do_transaction(false, pid, name, meta_tab, opts, fun) do
+    locks_table = Metadata.fetch!(meta_tab, :locks_table)
+    keys = Keyword.fetch!(opts, :keys)
+    ids = lock_ids(name, keys)
+
+    case Locks.acquire(locks_table, ids, opts) do
+      :ok ->
+        try do
+          _ = Process.put({pid, self()}, keys)
+
+          {:ok, fun.()}
+        after
+          _ = Process.delete({pid, self()})
+
+          Locks.release(locks_table, ids)
+        end
+
+      {:error, :timeout} ->
+        wrap_error Nebulex.Error, reason: :transaction_aborted, cache: name
+    end
+  end
+
+  defp lock_ids(name, []), do: [name]
+  defp lock_ids(name, keys), do: Enum.map(keys, &{name, &1})
+
   ## Helpers
 
   # Inline common instructions

lib/nebulex/adapters/local/backend.ex

@@ -1,15 +1,30 @@
 defmodule Nebulex.Adapters.Local.Backend do
   @moduledoc false
 
+  alias Nebulex.Adapters.Local.Metadata
+
   @doc false
   defmacro __using__(_opts) do
     quote do
       alias Nebulex.Adapters.Local.Generation
 
-      defp generation_spec(opts) do
+      defp generation_spec(opts, extra \\ []) do
+        opts = parse_opts(opts, extra)
+
         Supervisor.child_spec({Generation, opts}, id: Module.concat([__MODULE__, GC]))
       end
 
+      defp locks_spec(opts) do
+        meta_tab = Keyword.fetch!(opts, :adapter_meta).meta_tab
+        {lock_opts, _opts} = Keyword.pop!(opts, :lock_opts)
+
+        Supervisor.child_spec(
+          {Nebulex.Locks,
+           [init_callback: {unquote(__MODULE__), :init_callback, [meta_tab]}] ++ lock_opts},
+          id: Module.concat([__MODULE__, Locks])
+        )
+      end
+
       defp sup_spec(children) do
         %{
           id: Module.concat([__MODULE__, Supervisor]),
@@ -29,13 +44,13 @@ defmodule Nebulex.Adapters.Local.Backend do
 
         backend_opts =
           [
-            type,
             :public,
-            {:keypos, 2},
-            {:read_concurrency, Keyword.fetch!(opts, :read_concurrency)},
-            {:write_concurrency, Keyword.fetch!(opts, :write_concurrency)},
+            type,
             compressed,
-            extra
+            extra,
+            keypos: 2,
+            read_concurrency: Keyword.fetch!(opts, :read_concurrency),
+            write_concurrency: Keyword.fetch!(opts, :write_concurrency)
           ]
           |> List.flatten()
           |> Enum.filter(&(&1 != :named_table))
@@ -66,6 +81,15 @@ defmodule Nebulex.Adapters.Local.Backend do
     get_mod(backend).delete(meta_tab, gen_tab)
   end
 
+  @doc """
+  Helper function for initializing the locks table.
+  """
+  def init_callback(table, meta_tab) do
+    Metadata.put(meta_tab, :locks_table, table)
+  end
+
+  ## Private functions
+
   defp get_mod(:ets), do: Nebulex.Adapters.Local.Backend.ETS
 
   if Code.ensure_loaded?(:shards) do

lib/nebulex/adapters/local/backend/ets.ex

@@ -6,10 +6,10 @@ defmodule Nebulex.Adapters.Local.Backend.ETS do
 
   @doc false
   def child_spec(opts) do
-    opts
-    |> parse_opts()
-    |> generation_spec()
-    |> List.wrap()
+    [
+      locks_spec(opts),
+      generation_spec(opts)
+    ]
     |> sup_spec()
   end
 

lib/nebulex/adapters/local/backend/shards.ex

@@ -40,10 +40,12 @@ if Code.ensure_loaded?(:shards) do
         |> Keyword.fetch!(:adapter_meta)
         |> Map.fetch!(:meta_tab)
 
-      sup_spec([
+      [
         {__MODULE__.DynamicSupervisor, meta_tab},
-        generation_spec(parse_opts(opts, partitions: partitions))
-      ])
+        locks_spec(opts),
+        generation_spec(opts, partitions: partitions)
+      ]
+      |> sup_spec()
     end
 
     @doc false

lib/nebulex/adapters/local/generation.ex

@@ -240,11 +240,11 @@ defmodule Nebulex.Adapters.Local.Generation do
         |> Map.merge(adapter_meta)
       )
 
+    # Create a new generation
+    :ok = new_gen(state)
+
     # Timer ref
-    {:ok, ref} =
-      if state.gc_interval,
-        do: {new_gen(state), start_timer(state.gc_interval)},
-        else: {new_gen(state), nil}
+    ref = if state.gc_interval, do: start_timer(state.gc_interval)
 
     # Update state
     state = %{state | gc_heartbeat_ref: ref}

lib/nebulex/adapters/local/metadata.ex

@@ -1,28 +1,29 @@
 defmodule Nebulex.Adapters.Local.Metadata do
   @moduledoc false
 
-  @type tab :: :ets.tid() | atom
+  @type tab() :: :ets.tid() | atom()
 
-  @spec init :: tab
+  @spec init() :: tab()
   def init do
     :ets.new(__MODULE__, [:public, read_concurrency: true])
   end
 
-  @spec get(tab, term, term) :: term
+  @spec get(tab(), term(), term()) :: term()
   def get(tab, key, default \\ nil) do
     :ets.lookup_element(tab, key, 2)
   rescue
     ArgumentError -> default
   end
 
-  @spec fetch!(tab, term) :: term
+  @spec fetch!(tab(), term()) :: term()
   def fetch!(tab, key) do
     :ets.lookup_element(tab, key, 2)
   end
 
-  @spec put(tab, term, term) :: :ok
+  @spec put(tab(), term(), term()) :: :ok
   def put(tab, key, value) do
     true = :ets.insert(tab, {key, value})
+
     :ok
   end
 end

lib/nebulex/adapters/local/options.ex

@@ -144,6 +144,51 @@ defmodule Nebulex.Adapters.Local.Options do
       This grace period allows ongoing operations to complete before the
       generation is removed.
       """
+    ],
+    lock_opts: [
+      type: :keyword_list,
+      required: false,
+      default: [],
+      doc: """
+      Options to customize the locks manager used for cache transactions.
+
+      The local adapter uses `Nebulex.Locks` for implementing transactions with
+      an ETS-based locking mechanism optimized for single-node scenarios. This
+      option allows you to customize the behavior of the locks manager, such as
+      adjusting the cleanup interval or batch size for stale lock cleanup.
+
+      **Available Options:**
+
+        * `:cleanup_interval` - The interval in milliseconds for periodic
+          cleanup of stale locks (defaults to 5 minutes).
+        * `:cleanup_batch_size` - The number of locks to process per batch
+          during cleanup (defaults to 100).
+
+      Note: The `:name` and `:init_callback` options are managed internally by
+      the adapter and should not be provided. See the "Start Options" section
+      in `Nebulex.Locks` for all available options.
+
+      **Examples:**
+
+          # Use default lock options (recommended for most cases)
+          MyCache.start_link()
+
+          # Customize cleanup for high-throughput scenarios
+          MyCache.start_link(
+            lock_opts: [
+              cleanup_interval: :timer.minutes(1),
+              cleanup_batch_size: 500
+            ]
+          )
+
+          # More frequent cleanup for memory-constrained environments
+          MyCache.start_link(
+            lock_opts: [
+              cleanup_interval: :timer.seconds(30)
+            ]
+          )
+
+      """
     ]
   ]