Shardex (shardex v0.1.0)

Route keys and batches of items to sharded pools.

Define an instance:

defmodule MyApp.Shards do
  use Shardex, otp_app: :my_app
end

# config/runtime.exs
config :my_app, MyApp.Shards,
  shards: [
    shard_1: [
      primary: {Shardex.Adapter.Ecto, repo: MyApp.Repo, config: [database: "s1"]},
      replica: {Shardex.Adapter.Ecto, repo: MyApp.Repo, config: [database: "s1_ro"]}
    ],
    shard_2: {Shardex.Adapter.Ecto, repo: MyApp.Repo, config: [database: "s2"]}
  ]

Add MyApp.Shards to your supervision tree, then:

MyApp.Shards.run(org_id, fn repo -> repo.all(Contact) end)
MyApp.Shards.run_batch(messages, & &1.org_id, fn repo, msgs -> insert_all(repo, msgs) end)

Every function generated by use Shardex delegates to the function of the same name in this module with the instance as first argument.

Summary

Types

instance()

An instance: the use Shardex module or the :name given to start_link/1.

reason()

Reason returned in {:error, reason} by the routing functions.

Functions

__using__(opts)

Defines an instance module. Its functions mirror the ones in Shardex without the instance argument.

activate(instance, shard, opts \\ [])

Puts a shard (or one role with role:) back into rotation, starting pools first. Without role:, every role of the shard becomes active. Idempotent.

add_shard(instance, name, spec, opts \\ [])

Adds a shard at runtime: initializes its adapters, starts its pools, then makes it routable. With status: :drain the pools are started too but the shard is not routed until activate/3. Pools are not started when the instance runs with start_pools: false.

group(instance, items, key_fun, opts \\ [])

Groups items by the shard their key routes to, calling the strategy once with the deduplicated keys. Every item ends up exactly once in either a group or errors. Accepts :role and :fallback like ref/3; items whose shard has no usable role go to errors.

invalidate(instance, key)

Drops a cached routing entry (or :all) when the instance uses Shardex.Strategy.Cached. Returns {:error, :not_cached} otherwise.

lookup(instance, key, opts \\ [])

Returns the shard a key routes to.

maintenance(instance, shard, mode, opts \\ [])

Takes a shard (or one of its roles with role:) out of rotation.

ref(instance, key, opts \\ [])

Returns the ref of the pool a key routes to.

ref!(instance, key, opts \\ [])

Like ref/3 but raises Shardex.Error.

remove_shard(instance, name)

Removes a shard at runtime: unroutes it, stops its pools and drops it from the strategy.

run(instance, key, fun, opts \\ [])

Runs fun against the pool a key routes to, inside the adapter's context (for Ecto dynamic repos, put_dynamic_repo/1 is set and restored).

run!(instance, key, fun, opts \\ [])

Like run/4 but returns the result directly and raises Shardex.Error on routing errors.

run_batch(instance, items, key_fun, fun, opts \\ [])

Groups items (see group/4) and calls fun.(ref, items) once per shard, inside the adapter's context. Returns {:ok, %{shard => result}, errors}.

shards(instance)

Lists shards in configuration order. meta.pids maps each role to its pool pid (or nil).

start_link(opts)

Starts an instance. Usually you use Shardex instead and add your module to your supervision tree.

status(instance, shard)

Returns the status of a shard and of each of its roles.

Types

instance()

@type instance() :: atom()

An instance: the use Shardex module or the :name given to start_link/1.

reason()

@type reason() ::
  :maintenance
  | :unassigned
  | :no_routable_shards
  | {:unknown_shard, term()}
  | {:unknown_role, atom()}
  | {:strategy_error, term()}
  | term()

Reason returned in {:error, reason} by the routing functions.

Functions

using(opts)

(macro)

Defines an instance module. Its functions mirror the ones in Shardex without the instance argument.

Options:

:otp_app (required) - the application whose config holds the instance's options under config otp_app, MyModule, ...

activate(instance, shard, opts \\ [])

@spec activate(instance(), atom(), keyword()) :: :ok | {:error, term()}

Puts a shard (or one role with role:) back into rotation, starting pools first. Without role:, every role of the shard becomes active. Idempotent.

add_shard(instance, name, spec, opts \\ [])

@spec add_shard(instance(), atom(), term(), keyword()) :: :ok | {:error, term()}

Returns {:error, :already_exists}, {:error, {:invalid_shard_spec, message}}, {:error, {:adapter_init_failed, shard, role, reason}}, {:error, {:strategy_error, term}} or a pool start error; on error nothing is added.

Runtime changes are kept in memory only; after a restart the boot config wins. With Shardex.Strategy.Hash most keys remap; with JumpHash ~1/n keys move to the new shard.

group(instance, items, key_fun, opts \\ [])

@spec group(instance(), Enumerable.t(), (term() -> term()), keyword()) ::
  Shardex.Batch.t()

invalidate(instance, key)

@spec invalidate(instance(), term()) :: :ok | {:error, :not_cached}

Drops a cached routing entry (or :all) when the instance uses Shardex.Strategy.Cached. Returns {:error, :not_cached} otherwise.

lookup(instance, key, opts \\ [])

@spec lookup(instance(), term(), keyword()) ::
  {:ok, Shardex.Shard.t()} | {:error, reason()}

Returns the shard a key routes to.

Errors: :unassigned, :maintenance, :no_routable_shards, {:unknown_shard, name}, {:strategy_error, value} or any {:error, term} returned by your strategy.

maintenance(instance, shard, mode, opts \\ [])

@spec maintenance(instance(), atom(), :drain | :stop, keyword()) ::
  :ok | {:error, term()}

Takes a shard (or one of its roles with role:) out of rotation.

:drain - stop routing to it; pools keep running
:stop - stop routing to it, then stop its pools

Idempotent. Hash strategies keep routing over the full shard list, so keys of a shard in maintenance get {:error, :maintenance} rather than moving elsewhere.

ref(instance, key, opts \\ [])

@spec ref(instance(), term(), keyword()) :: {:ok, term()} | {:error, reason()}

Returns the ref of the pool a key routes to.

Options

:role - pool role (default :primary)
:fallback - role or list of roles to try, in order, when :role is missing or in maintenance

ref!(instance, key, opts \\ [])

@spec ref!(instance(), term(), keyword()) :: term()

Like ref/3 but raises Shardex.Error.

remove_shard(instance, name)

@spec remove_shard(instance(), atom()) ::
  :ok
  | {:error, {:unknown_shard, atom()} | :last_shard | {:strategy_error, term()}}

Removes a shard at runtime: unroutes it, stops its pools and drops it from the strategy.

Returns {:error, {:strategy_error, term}}, with nothing changed, when the strategy's on_topology_change/2 raises or does not return {:ok, state}.

run(instance, key, fun, opts \\ [])

@spec run(instance(), term(), (term() -> result), keyword()) ::
  {:ok, result} | {:error, reason()}
when result: term()

Runs fun against the pool a key routes to, inside the adapter's context (for Ecto dynamic repos, put_dynamic_repo/1 is set and restored).

Accepts the same options as ref/3. Exceptions raised by fun propagate; nothing is retried.

run!(instance, key, fun, opts \\ [])

@spec run!(instance(), term(), (term() -> result), keyword()) :: result
when result: term()

Like run/4 but returns the result directly and raises Shardex.Error on routing errors.

run_batch(instance, items, key_fun, fun, opts \\ [])

@spec run_batch(
  instance(),
  Enumerable.t(),
  (term() -> term()),
  (term(), [term()] -> result),
  keyword()
) ::
  {:ok, %{required(atom()) => result}, [{term(), reason()}]}
when result: term()

Groups items (see group/4) and calls fun.(ref, items) once per shard, inside the adapter's context. Returns {:ok, %{shard => result}, errors}.

Options

:role, :fallback - as in ref/3
:on_error - :collect (default) or :raise (raise Shardex.BatchError before running anything if any item cannot be routed)
:max_concurrency - run up to N shard groups in parallel tasks (default 1)
:timeout - per-group timeout in milliseconds when max_concurrency > 1 (default :infinity)

Invalid option values raise ArgumentError before any routing happens.

shards(instance)

@spec shards(instance()) :: [Shardex.Shard.t()]

Lists shards in configuration order. meta.pids maps each role to its pool pid (or nil).

start_link(opts)

@spec start_link(keyword()) :: Supervisor.on_start()

Starts an instance. Usually you use Shardex instead and add your module to your supervision tree.

Options

:name - Required. Instance name, a non-nil atom (set automatically by use Shardex).
:shards - Required. Keyword list of shard_name => pool_spec | [role => pool_spec].
:strategy - Strategy module, {module, opts} or a (key, ctx -> result) function. The default value is Shardex.Strategy.Hash.
:start_pools (boolean/0) - Set to false in tests to skip starting pools. The default value is true.
:start_failure - What to do when a pool fails to start at boot. The default value is :raise.

status(instance, shard)

@spec status(instance(), atom()) ::
  {:ok,
   %{
     status: Shardex.Shard.status(),
     roles: %{required(atom()) => Shardex.Shard.status()}
   }}
  | {:error, {:unknown_shard, atom()}}

Returns the status of a shard and of each of its roles.