Current section

Files

Jump to
super_cache lib api super_cache.ex
Raw

lib/api/super_cache.ex

# =============================================================================
# lib/super_cache.ex
#
# Unified public API. In :local mode writes go directly to ETS.
# In :distributed mode writes are routed through Cluster.Router.
# =============================================================================
defmodule SuperCache do
@moduledoc """
High-throughput, in-memory cache backed by ETS.
Works in both **local** (single-node) and **distributed** (cluster) modes.
The mode is selected at startup via the `:cluster` option and is
transparent to callers — the same API covers both.
## Setup
# Local (single-node, default)
SuperCache.start!(key_pos: 0, partition_pos: 0)
# Distributed cluster
SuperCache.start!(
key_pos: 0,
partition_pos: 0,
cluster: :distributed,
replication_factor: 2,
replication_mode: :async,
num_partition: 16
)
# Enable debug logging
# in config.exs:
# config :super_cache, debug_log: true
## Options
| Option | Type | Default | Description |
|------------------|---------|-----------------|--------------------------------------|
| `:key_pos` | integer | `0` | Tuple index used as the ETS key |
| `:partition_pos` | integer | `0` | Tuple index used to select partition |
| `:cluster` | atom | `:local` | `:local` or `:distributed` |
| `:num_partition` | integer | scheduler count | Number of ETS partitions |
| `:table_type` | atom | `:set` | ETS table type |
## Read modes (meaningful in distributed mode)
| Mode | Consistency | Latency |
|------------|--------------------|----------------------|
| `:local` | Eventual (default) | Zero extra latency |
| `:primary` | Strong per key | +1 RTT if non-primary|
| `:quorum` | Majority vote | +1 RTT (parallel) |
## Replication modes (distributed mode only)
| Mode | Guarantee | Extra latency |
|-----------|--------------------|-------------------|
| `:async` | Eventual (default) | None |
| `:sync` | At-least-once | +1 RTT per write |
| `:strong` | Three-phase commit | +3 RTTs per write |
## Basic usage
SuperCache.put!({:user, 1, "Alice"})
SuperCache.get!({:user, 1, nil})
# => [{:user, 1, "Alice"}]
# Read modes (distributed)
SuperCache.get!({:user, 1, nil}, read_mode: :primary)
SuperCache.get!({:user, 1, nil}, read_mode: :quorum)
SuperCache.delete!({:user, 1, nil})
SuperCache.delete_all()
## Error handling
`!`-suffix functions raise on error. Non-bang variants return
`{:error, exception}` instead and never raise.
"""
require Logger
require SuperCache.Log
alias SuperCache.{Config, Partition, Storage, Buffer}
alias SuperCache.Cluster.Router
## ── Lifecycle ────────────────────────────────────────────────────────────────
@doc "Start SuperCache with default options (local, key_pos: 0, partition_pos: 0)."
@spec start!() :: :ok
def start!(), do: SuperCache.Bootstrap.start!()
@doc "Start SuperCache with the given keyword options. Raises on invalid config."
@spec start!(keyword) :: :ok
def start!(opts), do: SuperCache.Bootstrap.start!(opts)
@doc "Start SuperCache. Returns `:ok | {:error, exception}` instead of raising."
@spec start() :: :ok | {:error, Exception.t()}
def start(), do: safe(fn -> SuperCache.Bootstrap.start!() end)
@doc "Start SuperCache with opts. Returns `:ok | {:error, exception}` instead of raising."
@spec start(keyword) :: :ok | {:error, Exception.t()}
def start(opts), do: safe(fn -> SuperCache.Bootstrap.start!(opts) end)
@doc "Returns `true` when SuperCache is running and ready."
@spec started?() :: boolean
def started?(), do: Config.get_config(:started, false)
@doc "Stop SuperCache and free all ETS memory."
@spec stop() :: :ok
def stop(), do: SuperCache.Bootstrap.stop()
## ── Write ────────────────────────────────────────────────────────────────────
@doc """
Store a tuple in the cache.
In distributed mode, routes the write to the partition's primary node and
replicates according to `:replication_mode`. Raises on error.
"""
@spec put!(tuple) :: true
def put!(data) when is_tuple(data) do
if Config.distributed?() do
Router.route_put!(data)
else
partition = data |> Config.get_partition!() |> Partition.get_partition()
SuperCache.Log.debug(fn ->
"super_cache, put key=#{inspect(Config.get_key!(data))} partition=#{inspect(partition)}"
end)
Storage.put(data, partition)
end
end
@doc "Store a tuple. Returns `true | {:error, exception}` instead of raising."
@spec put(tuple) :: true | {:error, Exception.t()}
def put(data), do: safe(fn -> put!(data) end)
@doc """
Enqueue a tuple for a buffered (lazy) write.
Falls back to `put!/1` with a warning when `:strong` replication is active.
"""
@spec lazy_put(tuple) :: :ok
def lazy_put(data) when is_tuple(data) do
if Config.distributed?() and SuperCache.Cluster.Manager.replication_mode() == :strong do
Logger.warning(
"super_cache, lazy_put called in :strong replication mode — falling back to put!"
)
put!(data)
:ok
else
Buffer.enqueue(data)
end
end
@doc """
Store multiple tuples in a single batch operation.
In distributed mode, groups data by partition and sends each group
in a single `:erpc` call, dramatically reducing network overhead.
## Example
SuperCache.put_batch!([
{:user, 1, "Alice"},
{:user, 2, "Bob"},
{:session, "tok1", :active}
])
"""
@spec put_batch!([tuple]) :: :ok
def put_batch!([]), do: :ok
def put_batch!(data_list) when is_list(data_list) do
if Config.distributed?() do
Router.route_put_batch!(data_list)
else
# Route each record to its correct partition based on its own key.
# Group by partition to minimize ETS table switches.
data_list
|> Enum.group_by(fn data ->
data |> Config.get_partition!() |> Partition.get_partition()
end)
|> Enum.each(fn {partition, records} ->
Enum.each(records, fn data ->
Storage.put(data, partition)
end)
end)
end
:ok
end
## ── Read ─────────────────────────────────────────────────────────────────────
@doc """
Retrieve all records whose key matches the key element of `data`.
## Options
- `:read_mode``:local` (default), `:primary`, or `:quorum`.
## Examples
SuperCache.get!({:user, 1, nil})
SuperCache.get!({:user, 1, nil}, read_mode: :primary)
SuperCache.get!({:user, 1, nil}, read_mode: :quorum)
"""
@spec get!(tuple, keyword) :: [tuple]
def get!(data, opts \\ []) when is_tuple(data) do
if Config.distributed?() do
Router.route_get!(data, opts)
else
key = Config.get_key!(data)
partition = data |> Config.get_partition!() |> Partition.get_partition()
SuperCache.Log.debug(fn ->
"super_cache, get key=#{inspect(key)} partition=#{inspect(partition)}"
end)
Storage.get(key, partition)
end
end
@doc "Retrieve records. Returns `[tuple] | {:error, exception}` instead of raising."
@spec get(tuple, keyword) :: [tuple] | {:error, Exception.t()}
def get(data, opts \\ []), do: safe(fn -> get!(data, opts) end)
@doc """
Retrieve records by explicit `key` and `partition_data`.
## Options
- `:read_mode``:local` (default), `:primary`, or `:quorum`.
"""
@spec get_by_key_partition!(any, any, keyword) :: [tuple]
def get_by_key_partition!(key, partition_data, opts \\ []) do
if Config.distributed?() do
Router.route_get_by_key_partition!(key, partition_data, opts)
else
Storage.get(key, Partition.get_partition(partition_data))
end
end
@doc "Non-raising variant of `get_by_key_partition!/3`."
@spec get_by_key_partition(any, any, keyword) :: [tuple] | {:error, Exception.t()}
def get_by_key_partition(key, partition_data, opts \\ []) do
safe(fn -> get_by_key_partition!(key, partition_data, opts) end)
end
@doc "Equivalent to `get_by_key_partition!(key, key, opts)` — for tables where key_pos == partition_pos."
@spec get_same_key_partition!(any, keyword) :: [tuple]
def get_same_key_partition!(key, opts \\ []), do: get_by_key_partition!(key, key, opts)
@doc "Non-raising variant of `get_same_key_partition!/2`."
@spec get_same_key_partition(any, keyword) :: [tuple] | {:error, Exception.t()}
def get_same_key_partition(key, opts \\ []),
do: safe(fn -> get_same_key_partition!(key, opts) end)
@doc """
Retrieve records matching an ETS match pattern (binding lists).
Pass `:_` as `partition_data` to scan all partitions.
## Options
- `:read_mode``:local` (default), `:primary`, or `:quorum`.
"""
@spec get_by_match!(any, tuple, keyword) :: [[any]]
def get_by_match!(partition_data, pattern, opts \\ []) when is_tuple(pattern) do
if Config.distributed?() do
Router.route_get_by_match!(partition_data, pattern, opts)
else
reduce_partitions(partition_data, [], fn p, acc ->
Storage.get_by_match(pattern, p) ++ acc
end)
end
end
@doc "Scan all partitions. Equivalent to `get_by_match!(:_, pattern, [])`."
@spec get_by_match!(tuple) :: [[any]]
def get_by_match!(pattern) when is_tuple(pattern), do: get_by_match!(:_, pattern, [])
@doc "Non-raising variant of `get_by_match!/2`."
@spec get_by_match(any, tuple, keyword) :: [[any]] | {:error, Exception.t()}
def get_by_match(partition_data, pattern, opts \\ []) when is_tuple(pattern) do
safe(fn -> get_by_match!(partition_data, pattern, opts) end)
end
@doc """
Retrieve full records matching an ETS match-object pattern.
Pass `:_` as `partition_data` to scan all partitions.
## Options
- `:read_mode``:local` (default), `:primary`, or `:quorum`.
"""
@spec get_by_match_object!(any, tuple, keyword) :: [tuple]
def get_by_match_object!(partition_data, pattern, opts \\ []) when is_tuple(pattern) do
if Config.distributed?() do
Router.route_get_by_match_object!(partition_data, pattern, opts)
else
reduce_partitions(partition_data, [], fn p, acc ->
Storage.get_by_match_object(pattern, p) ++ acc
end)
end
end
@doc "Scan all partitions. Equivalent to `get_by_match_object!(:_, pattern, [])`."
@spec get_by_match_object!(tuple) :: [tuple]
def get_by_match_object!(pattern) when is_tuple(pattern),
do: get_by_match_object!(:_, pattern, [])
@doc "Non-raising variant of `get_by_match_object!/2`."
@spec get_by_match_object(any, tuple, keyword) :: [tuple] | {:error, Exception.t()}
def get_by_match_object(partition_data, pattern, opts \\ []) when is_tuple(pattern) do
safe(fn -> get_by_match_object!(partition_data, pattern, opts) end)
end
## ── Scan ─────────────────────────────────────────────────────────────────────
@doc """
Fold over every record in a partition (or all when `:_`).
Always reads from the local ETS table regardless of replication mode.
"""
@spec scan!(any, (any, any -> any), any) :: any
def scan!(partition_data, fun, acc) when is_function(fun, 2) do
if Config.distributed?() do
Router.route_scan!(partition_data, fun, acc)
else
reduce_partitions(partition_data, acc, fn p, a -> Storage.scan(fun, a, p) end)
end
end
@doc "Equivalent to `scan!(:_, fun, acc)`."
@spec scan!((any, any -> any), any) :: any
def scan!(fun, acc) when is_function(fun, 2), do: scan!(:_, fun, acc)
@doc "Non-raising variant of `scan!/3`."
@spec scan(any, (any, any -> any), any) :: any | {:error, Exception.t()}
def scan(partition_data, fun, acc) when is_function(fun, 2) do
safe(fn -> scan!(partition_data, fun, acc) end)
end
## ── Delete ───────────────────────────────────────────────────────────────────
@doc """
Delete the record matching the key in `data`.
Routes to the primary in distributed mode.
"""
@spec delete!(tuple) :: :ok
def delete!(data) when is_tuple(data) do
if Config.distributed?() do
Router.route_delete!(data)
else
key = Config.get_key!(data)
partition = data |> Config.get_partition!() |> Partition.get_partition()
Storage.delete(key, partition)
:ok
end
end
@doc "Non-raising variant of `delete!/1`."
@spec delete(tuple) :: :ok | {:error, Exception.t()}
def delete(data), do: safe(fn -> delete!(data) end)
@doc """
Delete all records from every partition.
In distributed mode, issues one routed delete per partition to its primary.
"""
@spec delete_all() :: :ok
def delete_all() do
if Config.distributed?() do
Router.route_delete_all()
else
Partition.get_all_partition() |> List.flatten() |> Enum.each(&Storage.delete_all/1)
end
end
@doc """
Delete records matching `pattern` in `partition_data` (or all partitions for `:_`).
"""
@spec delete_by_match!(any, tuple) :: :ok
def delete_by_match!(partition_data, pattern) when is_tuple(pattern) do
if Config.distributed?() do
Router.route_delete_match!(partition_data, pattern)
else
resolve_partitions(partition_data) |> Enum.each(&Storage.delete_match(pattern, &1))
end
end
@doc "Delete from all partitions. Equivalent to `delete_by_match!(:_, pattern)`."
@spec delete_by_match!(tuple) :: :ok
def delete_by_match!(pattern) when is_tuple(pattern), do: delete_by_match!(:_, pattern)
@doc "Non-raising variant of `delete_by_match!/2`."
@spec delete_by_match(any, tuple) :: :ok | {:error, Exception.t()}
def delete_by_match(partition_data, pattern) when is_tuple(pattern) do
safe(fn -> delete_by_match!(partition_data, pattern) end)
end
@doc "Delete by explicit `key` and `partition_data`."
@spec delete_by_key_partition!(any, any) :: :ok
def delete_by_key_partition!(key, partition_data) do
if Config.distributed?() do
Router.route_delete_by_key_partition!(key, partition_data)
else
Storage.delete(key, Partition.get_partition(partition_data))
:ok
end
end
@doc "Non-raising variant of `delete_by_key_partition!/2`."
@spec delete_by_key_partition(any, any) :: :ok | {:error, Exception.t()}
def delete_by_key_partition(key, partition_data) do
safe(fn -> delete_by_key_partition!(key, partition_data) end)
end
@doc "Equivalent to `delete_by_key_partition!(key, key)` — for tables where key_pos == partition_pos."
@spec delete_same_key_partition!(any) :: :ok
def delete_same_key_partition!(key), do: delete_by_key_partition!(key, key)
@doc "Non-raising variant of `delete_same_key_partition!/1`."
@spec delete_same_key_partition(any) :: :ok | {:error, Exception.t()}
def delete_same_key_partition(key), do: safe(fn -> delete_same_key_partition!(key) end)
## ── Stats ────────────────────────────────────────────────────────────────────
# ── Direct partition access (zero routing overhead) ──────────────────────────
@doc """
Store a tuple directly into a specific partition table.
Bypasses all routing and partition resolution. Use when you already know
the target partition table atom (e.g., from a previous `get_partition/1`
call or from `Partition.get_all_partition/0`).
## Examples
partition = SuperCache.Partition.get_partition(:my_key)
SuperCache.put_partition!({:user, 1, "Alice"}, partition)
"""
@spec put_partition!(tuple, atom | :ets.tid()) :: true
def put_partition!(data, partition) when is_tuple(data) do
Storage.put(data, partition)
end
@doc """
Retrieve records directly from a specific partition table.
Bypasses all routing and partition resolution. The `key` is looked up
directly in the given `partition` table.
## Examples
partition = SuperCache.Partition.get_partition(:my_key)
SuperCache.get_partition!(:user, partition)
# => [{:user, 1, "Alice"}]
"""
@spec get_partition!(any, atom | :ets.tid()) :: [tuple]
def get_partition!(key, partition) do
Storage.get(key, partition)
end
@doc """
Delete a record directly from a specific partition table.
Bypasses all routing and partition resolution.
## Examples
partition = SuperCache.Partition.get_partition(:my_key)
SuperCache.delete_partition!(:user, partition)
# => :ok
"""
@spec delete_partition!(any, atom | :ets.tid()) :: :ok
def delete_partition!(key, partition) do
Storage.delete(key, partition)
:ok
end
@doc """
Store a tuple directly into a partition by its integer index.
This is the fastest write path — bypasses routing, partition resolution,
and even the partition table name lookup. The index is used directly to
fetch the cached partition table atom from `persistent_term`.
## Examples
# Get partition index once
idx = SuperCache.Partition.get_partition_order(:my_key)
# Use it for fast repeated access
SuperCache.put_partition_by_idx!({:user, 1, "Alice"}, idx)
"""
@spec put_partition_by_idx!(tuple, non_neg_integer) :: true
def put_partition_by_idx!(data, partition_idx)
when is_tuple(data) and is_integer(partition_idx) do
partition = Partition.get_partition_by_idx(partition_idx)
Storage.put(data, partition)
end
@doc """
Retrieve records directly from a partition by its integer index.
This is the fastest read path — bypasses routing, partition resolution,
and partition table name lookup.
## Examples
idx = SuperCache.Partition.get_partition_order(:my_key)
SuperCache.get_partition_by_idx!(:user, idx)
# => [{:user, 1, "Alice"}]
"""
@spec get_partition_by_idx!(any, non_neg_integer) :: [tuple]
def get_partition_by_idx!(key, partition_idx) when is_integer(partition_idx) do
partition = Partition.get_partition_by_idx(partition_idx)
Storage.get(key, partition)
end
@doc """
Delete a record directly from a partition by its integer index.
This is the fastest delete path — bypasses routing and partition resolution.
## Examples
idx = SuperCache.Partition.get_partition_order(:my_key)
SuperCache.delete_partition_by_idx!(:user, idx)
# => :ok
"""
@spec delete_partition_by_idx!(any, non_neg_integer) :: :ok
def delete_partition_by_idx!(key, partition_idx) when is_integer(partition_idx) do
partition = Partition.get_partition_by_idx(partition_idx)
Storage.delete(key, partition)
:ok
end
## ── Stats ────────────────────────────────────────────────────────────────────
@doc """
Return local ETS record counts per partition plus `:total`.
"""
@spec stats() :: keyword
def stats() do
entries = Partition.get_all_partition() |> List.flatten() |> Enum.map(&Storage.stats/1)
total = Enum.reduce(entries, 0, fn {_, n}, acc -> acc + n end)
entries ++ [total: total]
end
@doc """
Return an aggregated cluster-wide statistics map.
In local mode wraps `stats/0` in the same map format for a uniform interface.
In distributed mode gathers per-node counts via `:erpc`.
"""
@spec cluster_stats() :: map
def cluster_stats() do
if Config.distributed?() do
SuperCache.Cluster.Stats.cluster() |> Map.merge(gather_node_stats())
else
base = stats()
total = Keyword.get(base, :total, 0)
%{
nodes: [node()],
node_count: 1,
replication_factor: 1,
replication_mode: :none,
num_partitions: Partition.get_num_partition(),
total_records: total,
node_stats: %{node() => [partition_count: length(base) - 1, record_count: total]},
unreachable_nodes: []
}
end
end
## ── Private ──────────────────────────────────────────────────────────────────
@doc """
Returns `true` when SuperCache is running in distributed mode.
Delegates to `Config.distributed?/0` for zero-cost persistent_term reads.
"""
@spec distributed?() :: boolean
def distributed?, do: Config.distributed?()
defp resolve_partitions(:_), do: List.flatten(Partition.get_all_partition())
defp resolve_partitions(data), do: [Partition.get_partition(data)]
defp reduce_partitions(partition_data, acc, fun) do
partition_data |> resolve_partitions() |> Enum.reduce(acc, fun)
end
defp gather_node_stats() do
live = SuperCache.Cluster.Manager.live_nodes()
{node_stats, unreachable} =
live
|> Task.async_stream(
fn n ->
try do
s = :erpc.call(n, __MODULE__, :stats, [], 5_000)
total = Keyword.get(s, :total, 0)
{n, [partition_count: length(s) - 1, record_count: total]}
catch
_, reason -> {n, {:unreachable, reason}}
end
end,
timeout: 8_000,
on_timeout: :kill_task
)
|> Enum.reduce({%{}, []}, fn
{:ok, {n, {:unreachable, _}}}, {s, bad} -> {s, [n | bad]}
{:ok, {n, info}}, {s, bad} -> {Map.put(s, n, info), bad}
{:exit, _}, {s, bad} -> {s, bad}
end)
%{node_stats: node_stats, unreachable_nodes: unreachable}
end
defp safe(fun) do
fun.()
rescue
err ->
Logger.error(Exception.format(:error, err, __STACKTRACE__))
{:error, err}
end
end