Packages
The typed, attribute-queryable in-memory store for ETS and Mnesia, with Ecto changeset support, record expiry (TTL), crash resilience, and atomic take-once reads.
Current section
Files
Jump to
Current section
Files
lib/store.ex
defmodule ActiveMemory.Store do
@moduledoc """
# The Store
## Store API
- `Store.all/1` Get all records stored, optionally ordered and paged (see [Ordering and paging](#module-ordering-and-paging))
- `Store.count/1` Count the records stored, without reading them (see [Counting](#module-counting))
- `Store.delete/1` Delete the record provided, matched in full (see [Deleting a record](#module-deleting-a-record))
- `Store.delete_all/0` Delete all records stored
- `Store.exists?/2` Whether any record matches an attributes search or `match` query
- `Store.get/1` Get the record with the given primary key, or `{:error, :not_found}`
- `Store.get!/1` Like `get/1` but raises `ActiveMemory.NotFoundError`
- `Store.get_by/1` Get the single record matching an attributes search
- `Store.get_by!/1` Like `get_by/1` but raises `ActiveMemory.NotFoundError`
- `Store.one/1` Get one record matching either an attributes search or `match` query. Raises `ActiveMemory.MultipleResultsError` when several match
- `Store.one!/1` Like `one/1` but raises `ActiveMemory.NotFoundError`
- `Store.reload/1` Re-read a record from the table by its primary key
- `Store.reload!/1` Like `reload/1` but raises `ActiveMemory.NotFoundError`
- `Store.select/2` Get all records matching either an attributes search or `match` query, optionally ordered and paged
- `Store.withdraw/1` **Atomically** get one record matching either an attributes search or `match` query, delete the record and return it — exactly one concurrent caller wins, making it safe for take-once workloads
- `Store.write/1` Write a record into the memory table, from a struct or an `Ecto.Changeset`. An invalid changeset is returned as `{:error, changeset}` with its `action` set to `:insert`, exactly like `c:Ecto.Repo.insert/2`
## Reading a single record
`get/1` reads by primary key — the table's first field, which is what ETS and
Mnesia key a record on. That is `:uuid` on a table using
`auto_generate_uuid: true`, an Ecto schema's declared primary key, or simply the
first field declared.
```elixir
{:ok, person} = MyApp.People.Store.get(uuid)
person = MyApp.People.Store.get!(uuid) # raises ActiveMemory.NotFoundError
{:ok, person} = MyApp.People.Store.get_by(%{email: "kara@galactica.com"})
```
A query that is meant to find one record but matches several raises
`ActiveMemory.MultipleResultsError` from `one/1`, `one!/1`, `get_by/1` and
`get_by!/1`, as `c:Ecto.Repo.one/2` does. Use `select/2` when many records are
expected.
Because reads and writes match a record in full, a struct held across a change
goes stale; `reload/1` gets the current copy.
## Counting
`count/1` asks the table for its size (`:ets.info/2`, `:mnesia.table_info/2`), so
it is O(1) and never copies records out — unlike `length(all())`.
It is the size of the backend table, not a count under application level rules. On
a table with a `ttl` it includes records that have expired but have not been swept
yet, so it can exceed what the reads return; pass `sweep: true` to delete those
first and get a count that agrees with the reads. On a replicated Mnesia table it
is the size of the replica this node reads from, so nodes whose replicas have
diverged report different counts.
```elixir
MyApp.Tokens.Store.count() # O(1), may include expired records
MyApp.Tokens.Store.count(sweep: true) # sweeps first, then counts
```
`exists?/2` has to match a query against fields, so it costs a scan rather than
being O(1). It accepts `sweep: true` as well, though the answer never depends on
it, since reads already ignore an expired record.
## Ordering and paging
`all/1` and `select/2` take `:order_by`, `:limit` and `:offset`:
```elixir
MyApp.People.Store.all(order_by: :last, limit: 20)
MyApp.People.Store.all(order_by: [{:desc, :age}, :last], offset: 20, limit: 20)
MyApp.People.Store.select(%{cylon?: true}, order_by: :last)
```
Neither ETS nor Mnesia can order a result, so this sorts after reading —
`O(n log n)` over the matched records, not an index backed sort. `:limit` and
`:offset` are **convenience pagination, not indexed pagination**: every matched
record is read and sorted before the offset is thrown away, so
`offset: 10_000, limit: 10` pays for all 10,010. Without an `:order_by` the order
is whatever the table returns, which for a `:set` table is unspecified.
Values are compared with their own `compare/2` when they have one, so `Decimal`,
`DateTime`, `NaiveDateTime`, `Date` and `Time` fields sort correctly instead of by
Erlang term order, which compares those structs field by field.
## Deleting a record
`delete/1` removes an **exact** record match: the struct you pass is compared
field for field against what is stored (`:ets.delete_object/2`,
`:mnesia.delete_object/3`). Pass a struct that has diverged from the stored copy
— a stale read, or one you modified in memory — and nothing is removed, yet the
call still returns `:ok`, the same answer `delete/1` gives for a record that was
never there.
This is deliberate. It is the only correct behavior for a `:bag` table, where
several records share a key, and on a `:set` table it means a delete never
clobbers a newer version of a record written since you read it.
When you hold an identifier rather than a record you know is current, reach for
`withdraw/1` instead. It matches on a query, so staleness cannot affect it, it is
atomic, and it tells you whether anything was actually removed:
```elixir
case MyApp.People.Store.withdraw(%{uuid: uuid}) do
{:ok, person} -> # removed, and here is the record that was stored
{:error, :not_found} -> # nothing matched
end
```
## Concurrency
A `Store` is a `GenServer`, but the data functions above (`all/0`, `one/1`,
`select/1`, `write/1`, `delete/1`, `withdraw/1`, `delete_all/0`) are **not**
routed through that process and are **not** serialized by it. They are ordinary
module functions that run in the **caller's** process, delegating straight to the
table's adapter (and therefore to `:ets`/`:mnesia`). Concurrency is governed by
ETS/Mnesia themselves, so many processes read and write in parallel — the single
`GenServer` is not a bottleneck. Only lifecycle and metadata operations (`init`,
`state/0`, `reload_seeds/0`) actually use the `GenServer`.
These functions live on the `GenServer` module purely for **organization**: the
`Store` is the single place responsible for how the application talks to its
table, following the Single Responsibility Principle. See the
[S.T.O.N.E principles](https://www.hpt-consulting.org/blog/stone-principles) for
the broader design philosophy.
## Expiry (TTL)
When the `Store`'s `Table` declares a `ttl` (see `ActiveMemory.Table`), records
expire automatically. Expiry is enforced in two ways: reads (`one/1`, `select/1`,
`all/0`, `withdraw/1`) never return an expired record, and the `Store` periodically
sweeps expired records to reclaim memory. The sweep cadence defaults to one minute
and can be set with the `sweep_interval` option (milliseconds):
```elixir
defmodule MyApp.Tokens.Store do
use ActiveMemory.Store,
table: MyApp.Tokens.Token,
sweep_interval: :timer.seconds(30)
end
```
The sweep only runs when the table has a `ttl`; otherwise it is never scheduled.
## Seeding
When starting a `Store` there is an option to provide a valid seed file and have the `Store` auto load seeds contained in the file.
```elixir
defmodule MyApp.People.Store do
use ActiveMemory.Store,
table: MyApp.People.Person,
seed_file: Path.expand("person_seeds.exs", __DIR__)
end
```
## Before `init`
All stores are `GenServers` and have `init` functions. While those are abstracted you can still specify methods to run during the `init` phase of the GenServer startup. Use the `before_init` keyword and add the methods as tuples with the arguments.
```elixir
defmodule MyApp.People.Store do
use ActiveMemory.Store,
table: MyApp.People.Person,
before_init: [{:run_me, ["arg1", "arg2", ...]}, {:run_me_too, []}]
end
```
> #### `before_init` and table recovery {: .warning}
>
> For ETS stores, the table is preserved across a store crash/restart by
> `ActiveMemory.TableHeir`. On such a recovery seed files are *not* re-run, but
> `before_init` methods **always** run, including on recovery. If a
> `before_init` method writes records with unique or generated keys (for
> example a `uuid`), running it again on recovery can create duplicates.
>
> How to handle this is left to the implementer. One option is to make any
> `before_init` write follow a "find or create" pattern — check with `one/1`
> before calling `write/1` — so the method is idempotent across restarts:
>
> ```elixir
> def run_me(args) do
> record = build_record(args)
>
> case one(%{key: record.key}) do
> {:ok, existing} -> {:ok, existing}
> {:error, :not_found} -> write(record)
> end
> end
> ```
## Initial State
All stores are `GenServers` and thus have a state. The default state is a map as such:
```elixir
%{started_at: "date time when first started", table_name: MyApp.People.Person}
```
This default state can be overwritten with a new state structure or values by supplying a method and arguments as a tuple to the keyword `initial_state`. The method must return `{:ok, new_state}`.
```elixir
defmodule MyApp.People.Store do
use ActiveMemory.Store,
table: MyApp.People.Person,
initial_state: {:initial_state_method, ["arg1", "arg2", ...]}
end
```
"""
@default_sweep_interval :timer.seconds(60)
@doc false
defmacro __before_compile__(_env) do
# The catch-all must compile after any handle_info clauses the using module
# defines. Injected at the `use` site it would sit above them and swallow
# every custom message (timer ticks, monitors) without a trace.
quote generated: true do
def handle_info(_message, state), do: {:noreply, state}
end
end
defmacro __using__(opts) do
quote do
import unquote(__MODULE__)
use GenServer
alias ActiveMemory.Operations
opts = unquote(Macro.expand(opts, __CALLER__))
@table Keyword.get(opts, :table)
@before_init Keyword.get(opts, :before_init, :default)
@initial_state Keyword.get(opts, :initial_state, :default)
@seed_file Keyword.get(opts, :seed_file, nil)
@sweep_interval Keyword.get(opts, :sweep_interval, unquote(@default_sweep_interval))
def start_link(options \\ []) do
GenServer.start_link(__MODULE__, options, name: __MODULE__)
end
@impl true
def init(_) do
with {:ok, table_status} <- create_table(),
{:ok, :seed_success} <- __maybe_run_seeds__(table_status),
{:ok, _result} <- Operations.before_init(@before_init, __MODULE__),
{:ok, initial_state} <- __initial_state__() do
__schedule_sweep__()
{:ok, initial_state}
end
end
@spec all(keyword()) :: list(map())
def all(opts \\ []), do: Operations.all(@table, opts)
@spec count(keyword()) :: non_neg_integer()
def count(opts \\ []), do: Operations.count(@table, opts)
def create_table, do: Operations.create_table(@table)
@spec delete(any()) :: :ok | {:error, any()}
def delete(struct), do: Operations.delete(struct, @table)
@spec delete_all() :: :ok | {:error, any()}
def delete_all, do: Operations.delete_all(@table)
@spec exists?(map() | tuple(), keyword()) :: boolean()
def exists?(query, opts \\ []), do: Operations.exists?(query, @table, opts)
@spec get(any()) :: {:ok, map()} | {:error, any()}
def get(key), do: Operations.get(key, @table)
@spec get!(any()) :: map()
def get!(key), do: Operations.get!(key, @table)
@spec get_by(map()) :: {:ok, map()} | {:error, any()}
def get_by(query), do: Operations.get_by(query, @table)
@spec get_by!(map()) :: map()
def get_by!(query), do: Operations.get_by!(query, @table)
@spec one(map() | list(any())) :: {:ok, map()} | {:error, any()}
def one(query), do: Operations.one(query, @table)
@spec one!(map() | list(any())) :: map()
def one!(query), do: Operations.one!(query, @table)
@spec reload(map()) :: {:ok, map()} | {:error, any()}
def reload(struct), do: Operations.reload(struct, @table)
@spec reload!(map()) :: map()
def reload!(struct), do: Operations.reload!(struct, @table)
def reload_seeds do
GenServer.call(__MODULE__, :reload_seeds)
end
@spec select(map() | list(any()), keyword()) :: {:ok, list(map())} | {:error, any()}
def select(query, opts \\ []), do: Operations.select(query, @table, opts)
def state do
GenServer.call(__MODULE__, :state)
end
@spec withdraw(map() | list(any())) :: {:ok, map()} | {:error, any()}
def withdraw(query), do: Operations.withdraw(query, @table)
@spec write(map() | Ecto.Changeset.t()) ::
{:ok, map()} | {:error, Ecto.Changeset.t() | any()}
def write(struct_or_changeset), do: Operations.write(struct_or_changeset, @table)
@impl true
def handle_call(:reload_seeds, _from, state) do
{:reply, Operations.seed(@seed_file, @table), state}
end
@impl true
def handle_call(:state, _from, state), do: {:reply, state, state}
# Sent by `ActiveMemory.TableHeir` when it hands a recovered ETS table back
# to this store on restart. Ownership transfers with the message; nothing
# further is required here.
@impl true
def handle_info({:"ETS-TRANSFER", _table_ref, _from, _data}, state) do
{:noreply, state}
end
# Periodically reclaims memory from expired records when the table has a `ttl`.
def handle_info(:sweep, state) do
Operations.delete_expired(@table, System.system_time(:millisecond))
__schedule_sweep__()
{:noreply, state}
end
# The unknown-message catch-all is injected via __before_compile__ so it
# lands after any handle_info clauses the using module defines.
@before_compile ActiveMemory.Store
# A recovered table already holds its data, so seeding is skipped to avoid
# duplicating or clobbering the surviving records.
defp __maybe_run_seeds__(:recovered), do: {:ok, :seed_success}
defp __maybe_run_seeds__(:created), do: Operations.seed(@seed_file, @table)
# Schedules the next expiry sweep only when the table actually uses a `ttl`.
# The `ttl` is read at runtime by `Operations`, never here: inspecting the
# table module while this one compiles would make the table a compile time
# dependency, which breaks tooling that loads this file on its own.
defp __schedule_sweep__, do: Operations.schedule_sweep(@table, @sweep_interval)
# Only the clause matching the compile-time option is generated so the
# Elixir 1.19+ type checker never sees an unreachable clause.
if @initial_state == :default do
defp __initial_state__ do
{:ok,
%{
started_at: DateTime.utc_now(),
table_name: @table
}}
end
else
defp __initial_state__ do
{method, args} = @initial_state
:erlang.apply(__MODULE__, method, args)
end
end
end
end
end