replicated

Replicated mode for brazier/store: every node of the cluster keeps a full replica of the store and the replicas converge.

Open a replicated store with store.open(name, store.Replicated(config)) (or store.open_list). This module holds its configuration, its error types and the calls into the replica process (brazier_store_replica); applications use the brazier/store API, which is the same in both modes.

How it works (see docs/CLUSTERING.md):

Each node runs one replica process per replicated store. It holds an add-wins last-writer-wins map (brazier/store/crdt, a port of delta_crdt_ex’s AWLWWMap) and mirrors it into ETS, so reads never leave the node or call the replica.
A write is a call to the local replica. It is applied there, then its delta is broadcast to the other replicas over brazier/pubsub.
Every sync_interval_ms each replica broadcasts a digest of its state. A replica whose state differs pulls the sender’s full state, which repairs every delta lost to a partition or a crash. A replica that starts (a node joining, rejoining or a replica restarting) asks every peer for its full state at once.
Writes are ordered by a hybrid logical clock (brazier/store/hlc): concurrent writes of one key resolve to the greatest (timestamp, node, dot) on every node, and a delete never removes a value written concurrently on another node (add wins).
Every limit is checked on the writer (the write returns an error) and on every receiver (the merge is refused and logged).

Types

Config

</>

Replicated store settings. Build one from default_config() with a record update, e.g. Config(..replicated.default_config(), max_entries: 50_000).

pub type Config {
  Config(
    sync_interval_ms: Int,
    max_entries: Int,
    max_value_bytes: Int,
    max_delta_bytes: Int,
    max_sync_bytes: Int,
  )
}

Constructors

```
Config(
  sync_interval_ms: Int,
  max_entries: Int,
  max_value_bytes: Int,
  max_delta_bytes: Int,
  max_sync_bytes: Int,
)
```
Arguments

sync_interval_ms

How often each replica broadcasts its digest (the anti-entropy period): the longest a lost delta stays lost once nodes can talk.

max_entries

The most entries a replica holds: keys of a key-value store, values of a list store (counting concurrent values of one key until the next write replaces them).

max_value_bytes

The largest value (with its key), as erlang:external_size/1.

max_delta_bytes

The largest delta a write may produce or a replica may merge.

max_sync_bytes

The largest full state a replica sends or merges.

Kind

</>

The kind of store a replica holds.

pub type Kind {
  KeyValue
  List
}

Constructors

```
KeyValue
```
```
List
```

Server

</>

The registered replica process of one store on this node.

pub type Server

StartError

</>

Why a replicated store could not be opened.

pub type StartError {
  InvalidName(name: String, reason: String)
  InvalidConfig(reason: String)
  StartFailed(reason: String)
}

Constructors

InvalidName(name: String, reason: String)

The name is not 1-200 bytes of [A-Za-z0-9_-].

```
InvalidConfig(reason: String)
```
The configuration is out of range, or the store is already open in local mode or with a different configuration.
```
StartFailed(reason: String)
```
The replica process failed to start.

WriteError

</>

Why a write to a store failed. A node-local store never fails a write except for an invalid per-key topic.

pub type WriteError {
  TooManyEntries(limit: Int)
  ValueTooLarge(bytes: Int, limit: Int)
  DeltaTooLarge(bytes: Int, limit: Int)
  Unavailable(reason: String)
  InvalidTopic(topic: String)
}

Constructors

```
TooManyEntries(limit: Int)
```
The write would grow the replica past max_entries.
```
ValueTooLarge(bytes: Int, limit: Int)
```
A value is larger than max_value_bytes.
```
DeltaTooLarge(bytes: Int, limit: Int)
```
The write’s delta is larger than max_delta_bytes.
```
Unavailable(reason: String)
```
The local replica did not answer (not running, restarting or stuck).
```
InvalidTopic(topic: String)
```
The per-key topic prefix <> key is not a valid PubSub topic.

Values

append

</>

pub fn append(
  server: Server,
  key: String,
  values: List(value),
  topics: List(String),
) -> Result(Nil, WriteError)

Add values to key (list stores); topics are notified on the nodes that merge them, whenever they merge them.

context_size

</>

pub fn context_size(server: Server) -> Result(Int, WriteError)

The number of elements of the replica’s causal context (clock entries, out-of-order dots and floors). A replica restart adds one clock entry for its new incarnation; once it is in sync the node’s floor drops the earlier incarnations’ entries, so this stays bounded however often the replicas restart (see brazier/store/crdt, “Floors”).

count

</>

pub fn count(table: @internal EtsTable) -> Result(Int, String)

The number of keys in a key-value store’s mirror.

default_config

</>

pub fn default_config() -> Config

Defaults: a digest every second, 100_000 entries, 64 KiB values, 1 MiB deltas and 64 MiB full syncs.

describe_start_error

</>

pub fn describe_start_error(err: StartError) -> String

A human-readable description of a start error.

describe_write_error

</>

pub fn describe_write_error(err: WriteError) -> String

A human-readable description of a write error.

digest

</>

pub fn digest(server: Server) -> Result(BitArray, WriteError)

A hash of the replica’s whole state: two replicas have converged when their digests are equal.

get

</>

pub fn get(
  table: @internal EtsTable,
  key: String,
) -> Result(option.Option(value), String)

The value of key in a key-value store’s mirror.

get_all

</>

pub fn get_all(
  table: @internal EtsTable,
  key: String,
) -> Result(List(value), String)

Every value of key in a list store’s mirror, in timestamp order.

max_name_bytes

</>

pub const max_name_bytes: Int

The longest replicated store name, in bytes (its PubSub topic brazier_store:list:<name> must stay within 255 bytes).

put

</>

pub fn put(
  server: Server,
  key: String,
  value: value,
) -> Result(Nil, WriteError)

Set key to value (key-value stores).

remove

</>

pub fn remove(
  server: Server,
  key: String,
  topics: List(String),
) -> Result(Nil, WriteError)

Remove every value of key this replica has seen; topics are notified on the nodes that merge the removal.

size

</>

pub fn size(server: Server) -> Result(Int, WriteError)

The number of entries the replica holds.

start

</>

pub fn start(
  kind: Kind,
  name: String,
  config: Config,
) -> Result(#(Server, @internal EtsTable), StartError)

Start this node’s replica of a store (or return the running one, when it was started with the same configuration), with the ETS table its values are mirrored into.

stop

</>

pub fn stop(server: Server) -> Result(Nil, String)

Stop this node’s replica of a store and its supervisor. Its ETS mirror goes with it.

synced

</>

pub fn synced(server: Server) -> Result(Bool, WriteError)

Whether the replica is in sync: every peer replica (the replicas of this store on verified cluster members) last sent a digest equal to this replica’s digest. Ok(True) with no peer. A local write, a merge or a new peer makes it Ok(False) until the next digests (one sync_interval_ms apart) agree; a peer that has not yet sent a digest since it joined counts as not in sync. Every value written anywhere before the last digests were sent is then on this node.

validate_config

</>

pub fn validate_config(
  config: Config,
) -> Result(Config, StartError)

Check a configuration’s ranges.

Constructors

Arguments

Constructors

Constructors

Constructors