defmodule Aerospike.TxnOps do @moduledoc false # Internal ETS-backed transaction state tracker. # # Stores per-transaction tracking data in the `txn_tracking` ETS table: # - state machine (:open, :verified, :committed, :aborted) # - namespace constraint (all ops must target the same namespace) # - server-returned deadline (populated by TxnMonitor.register_key) # - reads map: Key → record version (for verify phase) # - writes set: Key structs (for roll-forward/rollback phase) # - write_in_doubt flag alias Aerospike.Error alias Aerospike.Key alias Aerospike.Tables alias Aerospike.Txn @type tracking :: %{ state: :open | :verified | :committed | :aborted, namespace: String.t() | nil, deadline: integer(), reads: %{Key.t() => term()}, writes: MapSet.t(), write_in_doubt: boolean() } @doc false @spec init_tracking(atom(), Txn.t()) :: true def init_tracking(conn_name, %Txn{} = txn) do entry = {txn.id, empty_tracking()} :ets.insert(Tables.txn_tracking(conn_name), entry) end @doc false @spec track_read(atom(), Txn.t(), Key.t(), term()) :: true def track_read(conn_name, %Txn{} = txn, %Key{} = key, version) do if version != nil do update_tracking(conn_name, txn, fn t -> %{t | reads: Map.put(t.reads, key, version)} end) else true end end @doc false @spec track_write(atom(), Txn.t(), Key.t(), term(), atom()) :: true def track_write(conn_name, %Txn{} = txn, %Key{} = key, version, _result_code) when version != nil do update_tracking(conn_name, txn, fn t -> %{t | reads: Map.put(t.reads, key, version)} end) end def track_write(conn_name, %Txn{} = txn, %Key{} = key, nil, :ok) do update_tracking(conn_name, txn, fn t -> %{t | reads: Map.delete(t.reads, key), writes: MapSet.put(t.writes, key)} end) end def track_write(_conn_name, %Txn{}, %Key{}, nil, _result_code), do: true @doc false @spec track_write_in_doubt(atom(), Txn.t(), Key.t()) :: true def track_write_in_doubt(conn_name, %Txn{} = txn, %Key{} = key) do update_tracking(conn_name, txn, fn t -> %{ t | write_in_doubt: true, reads: Map.delete(t.reads, key), writes: MapSet.put(t.writes, key) } end) end @doc false @spec get_tracking(atom(), Txn.t()) :: {:ok, tracking()} | {:error, :not_found} def get_tracking(conn_name, %Txn{} = txn) do case :ets.lookup(Tables.txn_tracking(conn_name), txn.id) do [{_, data}] -> {:ok, data} [] -> {:error, :not_found} end end @doc false @spec get_reads(atom(), Txn.t()) :: %{Key.t() => term()} def get_reads(conn_name, %Txn{} = txn) do case get_tracking(conn_name, txn) do {:ok, t} -> t.reads _ -> %{} end end @doc false @spec get_writes(atom(), Txn.t()) :: [Key.t()] def get_writes(conn_name, %Txn{} = txn) do case get_tracking(conn_name, txn) do {:ok, t} -> MapSet.to_list(t.writes) _ -> [] end end @doc false @spec set_state(atom(), Txn.t(), :open | :verified | :committed | :aborted) :: true def set_state(conn_name, %Txn{} = txn, state) when state in [:open, :verified, :committed, :aborted] do update_tracking(conn_name, txn, fn t -> %{t | state: state} end) end @doc false @spec set_namespace(atom(), Txn.t(), String.t()) :: :ok | {:error, Error.t()} def set_namespace(conn_name, %Txn{} = txn, namespace) when is_binary(namespace) do case get_tracking(conn_name, txn) do {:ok, %{namespace: nil}} -> update_tracking(conn_name, txn, fn t -> %{t | namespace: namespace} end) :ok {:ok, %{namespace: ^namespace}} -> :ok {:ok, %{namespace: other}} -> {:error, %Error{ code: :parameter_error, message: "transaction namespace mismatch: expected #{inspect(other)}, got #{inspect(namespace)}" }} {:error, :not_found} -> {:error, %Error{code: :parameter_error, message: "transaction not initialized"}} end end @doc false @spec set_deadline(atom(), Txn.t(), integer()) :: true def set_deadline(conn_name, %Txn{} = txn, deadline) when is_integer(deadline) do update_tracking(conn_name, txn, fn t -> %{t | deadline: deadline} end) end @doc false @spec get_deadline(atom(), Txn.t()) :: integer() def get_deadline(conn_name, %Txn{} = txn) do case get_tracking(conn_name, txn) do {:ok, t} -> t.deadline _ -> 0 end end @doc false @spec monitor_exists?(atom(), Txn.t()) :: boolean() def monitor_exists?(conn_name, %Txn{} = txn) do get_deadline(conn_name, txn) != 0 end @doc false @spec close_monitor?(atom(), Txn.t()) :: boolean() def close_monitor?(conn_name, %Txn{} = txn) do case get_tracking(conn_name, txn) do {:ok, t} -> t.deadline != 0 and not t.write_in_doubt _ -> false end end @doc false @spec write_exists?(atom(), Txn.t(), Key.t()) :: boolean() def write_exists?(conn_name, %Txn{} = txn, %Key{} = key) do case get_tracking(conn_name, txn) do {:ok, t} -> MapSet.member?(t.writes, key) _ -> false end end @doc false @spec get_read_version(atom(), Txn.t(), Key.t()) :: term() | nil def get_read_version(conn_name, %Txn{} = txn, %Key{} = key) do case get_tracking(conn_name, txn) do {:ok, t} -> Map.get(t.reads, key) _ -> nil end end @doc false @spec set_write_in_doubt(atom(), Txn.t()) :: true def set_write_in_doubt(conn_name, %Txn{} = txn) do update_tracking(conn_name, txn, fn t -> %{t | write_in_doubt: true} end) end @doc false @spec cleanup(atom(), Txn.t()) :: true def cleanup(conn_name, %Txn{} = txn) do :ets.delete(Tables.txn_tracking(conn_name), txn.id) end @doc false @spec verify_command(atom(), Txn.t()) :: :ok | {:error, Error.t()} def verify_command(conn_name, %Txn{} = txn) do case get_tracking(conn_name, txn) do {:ok, %{state: :open}} -> :ok {:ok, %{state: state}} -> {:error, %Error{ code: :parameter_error, message: "transaction is not open (state: #{state})" }} {:error, :not_found} -> {:error, %Error{code: :parameter_error, message: "transaction not initialized"}} end end defp empty_tracking do %{ state: :open, namespace: nil, deadline: 0, reads: %{}, writes: MapSet.new(), write_in_doubt: false } end defp update_tracking(conn_name, %Txn{} = txn, fun) do case :ets.lookup(Tables.txn_tracking(conn_name), txn.id) do [{_, data}] -> :ets.insert(Tables.txn_tracking(conn_name), {txn.id, fun.(data)}) [] -> true end end end