Current section

Files

Jump to
super_cache lib distributed struct.ex
Raw

lib/distributed/struct.ex

defmodule SuperCache.Distributed.Struct do
@moduledoc """
Cluster-aware struct store.
Writes (`add`, `remove`, `remove_all`, `init`) are routed to the primary
node for the partition that owns the struct type. Reads (`get`, `get_all`)
are served from the local node (eventual consistency).
The API is identical to `SuperCache.Struct` — swap the alias to migrate:
# Single-node
alias SuperCache.Struct, as: S
# Cluster-aware
alias SuperCache.Distributed.Struct, as: S
## Partition strategy
Every operation hashes `{:struct_storage, StructName}` to find the
partition. This namespace key is consistent across puts, deletes, and
reads, so all operations for a struct type always hit the same ETS table
regardless of which node handles the request.
## Example
alias SuperCache.Distributed.Struct, as: DS
SuperCache.Cluster.Bootstrap.start!(
key_pos: 0, partition_pos: 0,
cluster: :distributed, replication_factor: 2
)
defmodule Order do
defstruct [:id, :customer, :total, :status]
end
DS.init(%Order{}, :id)
DS.add(%Order{id: "o-1", customer: "Alice", total: 59.99, status: :pending})
DS.add(%Order{id: "o-2", customer: "Bob", total: 12.00, status: :shipped})
DS.get(%Order{id: "o-1"})
# => {:ok, %Order{id: "o-1", customer: "Alice", total: 59.99, status: :pending}}
{:ok, all} = DS.get_all(%Order{})
length(all) # => 2
DS.remove(%Order{id: "o-1"})
DS.remove_all(%Order{})
"""
alias SuperCache.{Storage, Partition}
alias SuperCache.Cluster.{Manager, Replicator}
require Logger
## Public API ─────────────────────────────────────────────────────────────────
@doc """
Register `key` as the lookup field for the struct type of `struct`.
Routed to the primary node. Must be called before any other operation on
this struct type.
## Example
DS.init(%Order{}, :id) # => true
DS.init(%Order{}, :id) # => {:error, "struct already initialised"}
"""
@spec init(map, atom) :: true | {:error, any}
def init(%{__struct__: _} = struct, key \\ :id) when is_atom(key) do
with true <- Map.has_key?(struct, key),
{:error, :key_not_found} <- get_key_field(struct) do
route_write(struct, :local_init, [struct, key])
else
false -> {:error, "key does not exist on struct"}
{:ok, _} -> {:error, "struct already initialised"}
end
end
@doc """
Store `struct`. Routed to the primary node.
Overwrites any existing struct with the same key.
## Example
DS.add(%Order{id: "o-1", customer: "Alice", total: 59.99, status: :pending})
# => {:ok, %Order{...}}
"""
@spec add(map) :: {:ok, map} | {:error, any}
def add(%{__struct__: _} = struct) do
with {:ok, _key} <- get_key_field(struct) do
route_write(struct, :local_add, [struct])
end
end
@doc """
Retrieve a struct by its key field from the **local node**.
Returns `{:ok, struct}` or `{:error, :not_found}`.
## Example
DS.get(%Order{id: "o-1"}) # => {:ok, %Order{...}}
DS.get(%Order{id: "nope"}) # => {:error, :not_found}
"""
@spec get(map) :: {:ok, map} | {:error, :not_found | any}
def get(%{__struct__: struct_name} = struct) do
with {:ok, key} <- get_key_field(struct) do
ns = namespace(struct)
partition = Partition.get_partition(ns)
ets_key = {{:struct_storage, :struct, struct_name}, Map.get(struct, key)}
case Storage.get(ets_key, partition) do
[] -> {:error, :not_found}
[{_, result}] -> {:ok, result}
end
end
end
@doc """
Return all structs of this type from the **local node**.
Returns `{:ok, [struct]}`.
## Example
{:ok, orders} = DS.get_all(%Order{})
Enum.map(orders, & &1.status)
"""
@spec get_all(map) :: {:ok, list} | {:error, any}
def get_all(%{__struct__: struct_name} = struct) do
with {:ok, _key} <- get_key_field(struct) do
ns = namespace(struct)
partition = Partition.get_partition(ns)
results =
Storage.get_by_match_object(
{{{:struct_storage, :struct, struct_name}, :_}, :_},
partition
)
|> Enum.map(fn {_, value} -> value end)
{:ok, results}
end
end
@doc """
Remove the struct matching the key field of `struct`. Routed to the primary.
Returns `{:ok, removed_struct}` or `{:error, :not_found}`.
## Example
DS.remove(%Order{id: "o-1"}) # => {:ok, %Order{...}}
DS.remove(%Order{id: "o-1"}) # => {:error, :not_found}
"""
@spec remove(map) :: {:ok, map} | {:error, any}
def remove(%{__struct__: _} = struct) do
with {:ok, _key} <- get_key_field(struct) do
case get(struct) do
{:error, :not_found} = err -> err
{:ok, _existing} = return ->
case route_write(struct, :local_remove, [struct]) do
:ok -> return
{:ok, _} -> return
other -> other
end
end
end
end
@doc """
Remove all structs of this type. Routed to the primary.
The key field registration is preserved; you can call `add/1` again
without calling `init/2`.
## Example
DS.remove_all(%Order{}) # => {:ok, :removed}
DS.get_all(%Order{}) # => {:ok, []}
"""
@spec remove_all(map) :: {:ok, :removed} | {:error, any}
def remove_all(%{__struct__: struct_name} = struct) do
with {:ok, _key} <- get_key_field(struct) do
Logger.debug(fn -> "super_cache, dist.struct, remove_all #{inspect(struct_name)}" end)
case route_write(struct, :local_remove_all, [struct]) do
:ok -> {:ok, :removed}
{:ok, _} -> {:ok, :removed}
other -> other
end
end
end
## Remote entry points (called via :erpc — do NOT call directly) ──────────────
@doc false
def local_init(%{__struct__: struct_name} = struct, key) do
ns = namespace(struct)
partition = Partition.get_partition(ns)
idx = Partition.get_partition_order(ns)
record = {{:struct_storage, :key, struct_name}, key}
Storage.put(record, partition)
Replicator.replicate(idx, :put, record)
true
end
@doc false
def local_add(%{__struct__: struct_name} = struct) do
with {:ok, key} <- get_key_field(struct) do
ns = namespace(struct)
partition = Partition.get_partition(ns)
idx = Partition.get_partition_order(ns)
key_data = Map.get(struct, key)
ets_key = {{:struct_storage, :struct, struct_name}, key_data}
Logger.debug(fn ->
"super_cache, dist.struct, add #{inspect(struct_name)} key=#{inspect(key_data)}"
end)
Storage.delete(ets_key, partition)
Storage.put({ets_key, struct}, partition)
Replicator.replicate(idx, :delete, ets_key)
Replicator.replicate(idx, :put, {ets_key, struct})
{:ok, struct}
end
end
@doc false
def local_remove(%{__struct__: struct_name} = struct) do
with {:ok, key} <- get_key_field(struct) do
ns = namespace(struct)
partition = Partition.get_partition(ns)
idx = Partition.get_partition_order(ns)
ets_key = {{:struct_storage, :struct, struct_name}, Map.get(struct, key)}
Storage.delete(ets_key, partition)
Replicator.replicate(idx, :delete, ets_key)
:ok
end
end
@doc false
def local_remove_all(%{__struct__: struct_name} = struct) do
ns = namespace(struct)
partition = Partition.get_partition(ns)
idx = Partition.get_partition_order(ns)
pattern = {{{:struct_storage, :struct, struct_name}, :_}, :_}
Storage.delete_match(pattern, partition)
Replicator.replicate(idx, :delete_match, pattern)
{:ok, :removed}
end
## Private ────────────────────────────────────────────────────────────────────
# Consistent namespace key for ALL partition hashing in this module.
defp namespace(%{__struct__: struct_name}), do: {:struct_storage, struct_name}
defp get_key_field(%{__struct__: struct_name} = struct) do
ns = namespace(struct)
partition = Partition.get_partition(ns)
case Storage.get({:struct_storage, :key, struct_name}, partition) do
[] -> {:error, :key_not_found}
[{_, key}] -> {:ok, key}
end
end
defp route_write(struct, fun, args) do
idx = Partition.get_partition_order(namespace(struct))
{primary, _} = Manager.get_replicas(idx)
if primary == node() do
apply(__MODULE__, fun, args)
else
Logger.debug(fn ->
"super_cache, dist.struct, fwd #{fun}#{inspect(primary)}"
end)
:erpc.call(primary, __MODULE__, fun, args, 5_000)
end
end
end