# `ExDataSketch.Server`
[🔗](https://github.com/thanos/ex_data_sketch/blob/main/lib/ex_data_sketch/server.ex#L1)

A supervised, named, concurrently-updatable sketch process with optional
windowing and snapshotting.

`ExDataSketch.Server` is a `GenServer` wrapping a single sketch (or
`ExDataSketch.Window`), so an application can hold a running counter --
"distinct users right now", "distinct users in the last 5 minutes" -- as
an ordinary named process instead of hand-writing state management,
concurrency, and crash recovery.

## Quick Example

    {:ok, _pid} = ExDataSketch.Server.start_link(
      name: :uniques,
      sketch: :hll,
      sketch_opts: [p: 14],
      window: [every: :timer.minutes(1), keep: 5],
      snapshot: [to: {ExDataSketch.Storage.ETS, :sketches, "uniques"}, every: :timer.seconds(30)]
    )

    ExDataSketch.Server.update(:uniques, user_id)
    ExDataSketch.Server.estimate(:uniques)

## Backpressure

`update/2` and `update_many/2` are `GenServer.cast/2` calls: they return
immediately and do not wait for the update to be applied. Under
sustained load faster than the server can process, its mailbox grows
without bound unless `:max_queue` is configured. This is a deliberate
trade for throughput -- callers that need a guarantee the update was
applied before continuing use `update_sync/2` instead, a `GenServer.call/2`
that blocks until processed (and, as an ordinary call, provides natural
backpressure: a caller cannot get more than one `update_sync/2` ahead of
the server).

When `:max_queue` is set, `update/2` and `update_many/2` drop the update
(rather than applying it) whenever the server's own mailbox is at or
above that threshold, emitting `[:ex_data_sketch, :server, :drop]`. This
bounds memory growth from a backed-up mailbox at the cost of the data
loss `update/2` already implies under overload -- `update_sync/2` is never
subject to `:max_queue`.

`:max_queue` is checked when each cast is *processed*, not when it is
sent -- casts are asynchronous, so a sudden burst can enqueue far more
messages than `:max_queue` before the server drains even one of them.
The threshold bounds the server's steady-state backlog (how many queued
updates it will keep applying before it starts dropping instead), not
the peak mailbox size any single burst can reach. If bounding peak
memory during bursts specifically matters, rate-limit or batch on the
producer side rather than relying on `:max_queue` alone.

## Windowing

Pass `:window` (the same options `ExDataSketch.Window.new/3` accepts:
`:every`, `:keep`, `:time_fn`) to have the server hold a
`ExDataSketch.Window` instead of a bare sketch. `estimate/1` then answers
"in the last `keep * every`" instead of "ever". See `guides/windowing.md`
for what that means precisely (a tumbling ring, not an exact sliding
window) and which sketch families can be windowed.

Add `track_all_time: true` to `:window` to also maintain a second,
un-windowed sketch alongside the window (same `:sketch_opts`, so the same
fixed size as any other sketch of that family -- not a growing or
unbounded structure). `estimate(server, window: :all)` reads it;
`snapshot`, if configured, always persists the window, not this second
sketch, which restarts empty after a crash even when enabled (see
`baoulo/plans/0.10.0_phase4_design_review.md` section 4.2 for the
rationale).

A windowed server does not support `merge/2` in this release --
`ExDataSketch.Window` has no primitive to merge an already-built sketch
into its current slot (only raw-item `update/2,3` and `update_many/2`).
Calling `merge/2` on a windowed server raises
`ExDataSketch.Errors.UnsupportedOperationError`. This is tracked as
follow-up work in `baoulo/plans/plan-0.10.0.md` section 9, not an
oversight.

## Snapshotting

Pass `:snapshot` (`:to`, an `{backend_module, ref, key}` triple naming
any `ExDataSketch.Storage` backend; `:every`, a millisecond interval or
`:infinity`) to persist the server's current state periodically and on
graceful shutdown. On start, the server attempts to load from the same
location first (crash recovery); if nothing is found, or loading fails
for any other reason, it starts from a fresh sketch either way --
`[:ex_data_sketch, :server, :restore]` fires either way, with `found` in
its metadata saying which happened.

The periodic snapshot itself is triggered by a timer message that shares
the server's single mailbox with every `update/2`/`update_many/2` cast --
a GenServer processes its mailbox strictly in arrival order, so under a
sustained cast backlog the snapshot timer message can sit queued behind
it. `:every` is therefore a best-effort *minimum* interval between
snapshots, not a hard deadline: a server saturated with updates snapshots
less often than configured until the backlog drains (or `:max_queue`
starts shedding load). This does not affect the graceful-shutdown
snapshot below, which runs from `terminate/2` regardless of backlog.

Worst-case data loss from an ordinary (non-`:kill`) process termination
is bounded by `:every` -- a graceful stop or supervisor-initiated shutdown
additionally snapshots on `terminate/2` before exiting, so that case loses
nothing. This relies on the server trapping exits (it does, unconditionally,
from `init/1` on); if your storage backend can be slow (e.g. a networked
Ecto repo under load), consider overriding the default 5-second shutdown
timeout via `Supervisor.child_spec(ExDataSketch.Server, shutdown: 10_000)`
so the supervisor doesn't escalate to a hard kill before the snapshot
write completes. An untrappable `Process.exit(pid, :kill)` never runs
`terminate/2` at all (this is a BEAM guarantee, not something any
process can opt out of), so in that specific case the periodic `:every`
interval is the only protection -- this is the "kill" scenario the
plan's own test targets.

## Flushing

Pass `:flush` (`:interval`, a millisecond interval or `:infinity`;
`:callback`, an optional `(struct() -> term())`) for the
return-and-reset-on-a-timer pattern `ExDataSketch.Broadway.PeriodicAggregator`
uses: on each interval, the callback (if any) receives the current
sketch, `[:ex_data_sketch, :server, :flush]` fires, and the server resets
to a fresh sketch. `PeriodicAggregator` is a thin wrapper around a
`:flush`-configured `Server` -- see its moduledoc for the relationship.
`:flush` and `:window` are independent options; combining them is not
rejected, but resets the whole window (every slot) on each flush and is
not a use case this module specifically targets.

# `estimate_opts`

```elixir
@type estimate_opts() :: [{:window, :all}]
```

# `flush_opts`

```elixir
@type flush_opts() :: [
  interval: pos_integer() | :infinity,
  callback: (struct() -> term()) | nil
]
```

Periodic return-and-reset configuration, matching `ExDataSketch.Broadway.PeriodicAggregator`.

# `sketch_type`

```elixir
@type sketch_type() :: ExDataSketch.sketch_type() | module()
```

Either a registry atom (`:hll`) or a sketch module (`ExDataSketch.HLL`).

# `snapshot_opts`

```elixir
@type snapshot_opts() :: [
  to: {module(), term(), ExDataSketch.Storage.key()},
  every: pos_integer() | :infinity
]
```

Where and how often to persist state. `:every` may be `:infinity` (snapshot on graceful stop only).

# `start_opts`

```elixir
@type start_opts() :: [
  sketch: sketch_type(),
  sketch_opts: keyword(),
  name: GenServer.name(),
  window: window_opts() | nil,
  snapshot: snapshot_opts() | nil,
  flush: flush_opts() | nil,
  max_queue: non_neg_integer() | :infinity
]
```

# `window_opts`

```elixir
@type window_opts() :: [
  every: pos_integer(),
  keep: pos_integer(),
  time_fn: (-> integer()),
  track_all_time: boolean()
]
```

The same options `ExDataSketch.Window.new/3` accepts, plus `:track_all_time`.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `estimate`

```elixir
@spec estimate(GenServer.server(), estimate_opts()) :: number()
```

Returns the server's current headline estimate.

With no `:window` configured, this is `ExDataSketch.estimate/1` on the
underlying sketch. With `:window` configured, it defaults to the windowed
reading (`ExDataSketch.Window.estimate/1`); pass `window: :all` to read
the un-windowed, all-time accumulator instead, which requires the server
to have been started with `window: [..., track_all_time: true]`.

Raises `ExDataSketch.Errors.UnsupportedOperationError` for `window: :all`
when `track_all_time: true` was not set, or for any `estimate/1,2` call
against a family with no single-value reading (see
`ExDataSketch.estimate/1`).

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
    iex> ExDataSketch.Server.estimate(pid) > 0.0
    true

    iex> {:ok, pid} = ExDataSketch.Server.start_link(
    ...>   sketch: :hll, sketch_opts: [p: 10],
    ...>   window: [every: 60_000, keep: 5, track_all_time: true]
    ...> )
    iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
    iex> ExDataSketch.Server.estimate(pid) > 0.0 and ExDataSketch.Server.estimate(pid, window: :all) > 0.0
    true

# `flush`

```elixir
@spec flush(GenServer.server()) :: struct()
```

Returns the server's current state and resets it to a fresh, empty one,
on demand (independent of any configured `:flush` interval).

Unlike the automatic timer-driven flush (see the module documentation's
"Flushing" section), a manual call here does not invoke the `:flush`
callback -- only the timer does. Both emit
`[:ex_data_sketch, :server, :flush]`.

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
    iex> flushed = ExDataSketch.Server.flush(pid)
    iex> ExDataSketch.HLL.estimate(flushed) > 0.0
    true
    iex> ExDataSketch.Server.estimate(pid)
    0.0

# `merge`

```elixir
@spec merge(
  GenServer.server(),
  struct()
) :: :ok
```

Merges an already-built partial sketch into the server's current state.

Raises `ExDataSketch.Errors.UnsupportedOperationError` if the server is
windowed -- see the module documentation's "Windowing" section.

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> partial = ExDataSketch.HLL.new(p: 10) |> ExDataSketch.HLL.update("a")
    iex> :ok = ExDataSketch.Server.merge(pid, partial)
    iex> ExDataSketch.Server.estimate(pid) > 0.0
    true

# `reset`

```elixir
@spec reset(GenServer.server()) :: :ok
```

Resets the server to a fresh, empty state (the window and, if configured,
the all-time accumulator, when windowed).

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
    iex> :ok = ExDataSketch.Server.reset(pid)
    iex> ExDataSketch.Server.estimate(pid)
    0.0

# `sketch`

```elixir
@spec sketch(GenServer.server()) :: struct()
```

Returns the server's current raw state: a `%ExDataSketch.Window{}` if
`:window` was configured, otherwise the bare sketch struct.

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> match?(%ExDataSketch.HLL{}, ExDataSketch.Server.sketch(pid))
    true

# `start_link`

```elixir
@spec start_link(start_opts()) :: GenServer.on_start()
```

Starts a server process.

## Options

- `:sketch` -- required. A registry atom (`:hll`, looked up via
  `ExDataSketch.sketches/0`) or a sketch module directly. Must resolve to
  a mergeable family (see `ExDataSketch.Window.new/3`) when `:window` is
  also given.
- `:sketch_opts` -- options forwarded to the sketch module's `new/1`
  (default: `[]`).
- `:name` -- optional, forwarded to `GenServer.start_link/3`'s `:name`
  (an atom, `{:global, term}`, or `{:via, module, term}`).
- `:window` -- optional. See the module documentation's "Windowing"
  section.
- `:snapshot` -- optional. See the module documentation's "Snapshotting"
  section.
- `:flush` -- optional. See the module documentation's "Flushing" section.
- `:max_queue` -- optional non-negative integer or `:infinity` (default:
  `:infinity`). See the module documentation's "Backpressure" section.

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> is_pid(pid)
    true

# `update`

```elixir
@spec update(GenServer.server(), term()) :: :ok
```

Updates the server with a single item.

Asynchronous (`GenServer.cast/2`) -- returns immediately, does not wait
for the update to be applied, and can be dropped under `:max_queue`
pressure. See the module documentation's "Backpressure" section, and
`update_sync/2` for a guaranteed alternative.

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> ExDataSketch.Server.update(pid, "user_1")
    :ok

# `update_many`

```elixir
@spec update_many(GenServer.server(), Enumerable.t()) :: :ok
```

Updates the server with every item in an enumerable.

Asynchronous, like `update/2` -- see the module documentation's
"Backpressure" section.

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> ExDataSketch.Server.update_many(pid, ["a", "b", "c"])
    :ok

# `update_sync`

```elixir
@spec update_sync(GenServer.server(), term()) :: :ok
```

Updates the server with a single item, blocking until it has been applied.

Never dropped by `:max_queue` -- callers that need a guarantee the
update was applied use this instead of `update/2`.

## Examples

    iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
    iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
    iex> ExDataSketch.Server.estimate(pid) > 0.0
    true

---

*Consult [api-reference.md](api-reference.md) for complete listing*