EKV

Eventually consistent durable KV store for Elixir with opt-in per-key linearizable CAS, with zero runtime dependencies.

Data survives node restarts, node death, and network partitions. Member nodes replicate directly across all connected Erlang nodes using delta sync via per-shard oplogs. Storage is backed by SQLite (vendored, compiled as a NIF) with zero runtime dependencies.

Installation

def deps do
  [
    {:ekv, "~> 0.2.0"}
  ]
end

EKV uses sqlite as the storage layer. Precompiled NIF binaries are available for common platforms. If a precompiled binary isn't available for your system, it will compile from source (requires a C compiler).

Usage

Add EKV to your supervision tree:

children = [
  {EKV, name: :my_kv, data_dir: "data/ekv/my_kv"}
]

Or start a stateless client that routes to voting members by region preference:

children = [
  {EKV,
   name: :my_kv_client,
   mode: :client,
   region: "ord",
   region_routing: ["iad", "dfw", "lhr"],
   wait_for_route: :timer.seconds(10),
   wait_for_quorum: :timer.seconds(10)}
]

Or start an observer that keeps a local durable replica but routes CAS to voting members:

children = [
  {EKV,
   name: :my_kv_observer,
   mode: :observer,
   data_dir: "data/ekv/my_kv_observer",
   cluster_size: 3,
   region: "lhr",
   region_routing: ["iad", "dfw", "lhr"],
   wait_for_route: :timer.seconds(10),
   wait_for_quorum: :timer.seconds(10)}
]

Then use the API:

# Put / Get / Delete
EKV.put(:my_kv, "user/1", %{name: "Alice", role: :admin})
EKV.get(:my_kv, "user/1")
#=> %{name: "Alice", role: :admin}

EKV.delete(:my_kv, "user/1")
EKV.get(:my_kv, "user/1")
#=> nil

# TTL
EKV.put(:my_kv, "session/abc", token, ttl: :timer.minutes(30))

# Prefix scans
EKV.put(:my_kv, "user/1", %{name: "Alice"})
EKV.put(:my_kv, "user/2", %{name: "Bob"})

EKV.scan(:my_kv, "user/") |> Enum.to_list()
#=> [
#=>   {"user/1", %{name: "Alice"}, {ts, origin_node}},
#=>   {"user/2", %{name: "Bob"}, {ts, origin_node}}
#=> ]

EKV.keys(:my_kv, "user/")
#=> [{"user/1", {ts, origin_node}}, {"user/2", {ts, origin_node}}]

# Subscribe to a key
EKV.subscribe(:my_kv, "room/1")
EKV.put(:my_kv, "room/1", %{title: "Elixir"})

    # => receive
    {:ekv, [%EKV.Event{type: :put, key: "room/1", value: %{title: "Elixir"}}], %{name: :my_kv}}

# Subscribe to a prefix
EKV.subscribe(:my_kv, "room/")

    # => receive
    {:ekv, [
     %EKV.Event{type: :put, key: "room/1", value: %{title: "Elixir"}},
     %EKV.Event{type: :put, key: "room/2", value: %{title: "Phoenix"}}
    ], %{name: :my_kv}}

EKV.unsubscribe(:my_kv, "room/")

Values can be any Erlang term (stored via :erlang.term_to_binary/1). Keys are strings.

Options

Client mode

Client mode keeps the EKV API but does not start SQLite, replication, GC, or blue-green machinery on that node.

• Eventual reads become remote reads against the selected voter.
• wait_for_route can hold startup until a backend route is selected.
• wait_for_quorum can additionally hold startup until that backend reports CAS quorum reachable.
• scan/2 and keys/2 still return Elixir streams, but are backed by paged RPC.
• subscribe/2 works in client mode; client subscribers are delivered cluster-wide.
• Routing, subscriptions, and shutdown coordination use an EKV-instance-specific :pg scope, so multiple EKV instances can share a cluster without mixing control traffic.
• After backend failover, eventual reads may observe an older replica view. Use consistent: true when freshness matters.

Observer mode

Observer mode is for nodes that should keep a full local durable replica and low-latency eventual reads locally, but should not increase the CAS voter set.

• Observers start SQLite, replication, GC, subscriptions, and anti-entropy.
• Eventual reads and eventual writes stay local on the observer.
• CAS reads and writes route to voters selected by region_routing.
• Successful observer CAS calls apply the committed result locally before replying, so immediate local eventual read-your-CAS-writes is preserved.
• Observers do not count toward CAS quorum, and clients do not route to them.
• {:error, :unconfirmed} on an observer still means "do a consistent read before trusting local eventual state"; the ambiguity may be remote CAS outcome or local observer visibility.

How It Works

Storage

Each shard has a single SQLite database (WAL mode) as its sole storage layer — no data is held in memory so your dataset is not bound by available system memory. Normal writes go through the shard GenServer and atomically update current state plus retained replay history in a single NIF call. Replay rows use a deduplicated kv_keyrefs dictionary so kv_oplog does not repeat full key strings on every version, and full sync rebuilds kv without seeding replay history on the receiver. Reads go directly to SQLite via per-scheduler read connections stored in persistent_term.

Data survives restarts automatically since SQLite is the source of truth.

Replication

Every write is broadcast to the counterpart shard on all connected members. Member discovery is self-contained by monitoring connected Erlang nodes going up and down. Client routing, client subscriptions, and shutdown coordination are separate and use an EKV-instance-specific :pg scope.

*Note: Node connection is left up to the user, ie either explicit Node.connect/1/sys.config, or using a library like DNSCluster, or libcluster.

When a node connects (or reconnects), each shard pair exchanges a handshake. Based on high-water marks (HWMs), they decide:

• Delta sync if the oplog still has entries since the member's last known position (efficient for brief disconnects).
• Relayed delta if a member is behind on some third-party origin that is currently down but a live peer still retains that origin stream.
• Full sync if the oplog has been truncated past that point or the member is new (sends all live entries + recent tombstones; expired rows are omitted). Full sync rebuilds kv on the receiver but does not seed kv_oplog.

Connected members also re-run that same handshake periodically by default (anti_entropy_interval) so a member that missed a prior update eventually repairs itself without waiting for a reconnect.

Conflict resolution

Last-Writer-Wins with nanosecond timestamps. Ties are broken deterministically by comparing persisted origin strings, so all nodes converge to the same result without coordination. In current member mode this origin is the stable node_id, not the transient blue/green Erlang node name.

A delete is just an entry with deleted_at set. Same LWW rules apply -- a put with a higher timestamp beats a delete, and vice versa.

Consistency Modes - LWW vs CAS (Compare-And-Swap)

EKV supports two write modes:

• Eventual/LWW mode: default EKV.put/4 and EKV.delete/3 without CAS options.
• CAS mode: EKV.put/4 with if_vsn: or consistent: true, EKV.delete/3 with if_vsn:, and EKV.update/4.

Use consistent mode as key ownership:

• Different keys may use different consistency modes in the same EKV instance.
• A key may start in eventual/LWW mode and later transition to CAS mode (LWW -> CAS is supported).
• Once a key is CAS-managed, eventual writes on that key are rejected (CAS -> LWW is not supported for writes).
• Keys managed via CAS should keep using CAS write APIs.
• Important migration caveat: LWW -> CAS is an operational cutover, not a partition-safe fenced mode switch. A partitioned or stale node that has not yet learned the key is CAS-managed can still accept an eventual write on that key. After heal, that stale LWW write may win by normal LWW timestamp ordering if it is newer than the state the CAS quorum saw. This is a mixed-mode edge case, not a steady-state CAS behavior.
• Recommended cutover for a key moving to CAS:
  • quiesce eventual writers for that key
  • switch writers to CAS on a healthy cluster
  • wait for anti-entropy / partition healing to settle before relying on CAS-only ownership
• Reads for CAS-managed keys can be eventual (EKV.get/2, EKV.lookup/2) for lower latency, or consistent (EKV.get/3, consistent: true) when freshness matters. consistent: true is a barrier/linearizable read.
• EKV.keys/2 returns {key, vsn} tuples so callers can pipeline scans into CAS writes (if_vsn:) without fetching full values.
• CAS write APIs return committed VSNs on success ({:ok, vsn} for put/delete, {:ok, value, vsn} for update) so callers can chain later if_vsn: guards without an extra lookup.
• Eventual writes on CAS-managed keys return {:error, :cas_managed_key}.

CAS write error semantics

CAS writes (put with if_vsn: or consistent: true, delete with if_vsn:, update) can return:

• {:error, :conflict}: write was rejected before a deciding accept phase (for example: version mismatch or pre-accept contention).
• {:error, :unconfirmed}: write entered accept phase, but the caller could not confirm final outcome. The write may already be committed, or it may have lost to a competing ballot and never committed.

On :unconfirmed, resolve with EKV.get(name, key, consistent: true) before taking follow-up actions. On observers, the same recovery rule applies if the remote CAS may have committed but the observer could not confirm local visibility before replying.

You can opt in to internal resolution per call with resolve_unconfirmed: true on CAS write APIs. In that mode, EKV performs one barrier read when an ambiguous accept outcome occurs and returns resolved current-state outcomes ({:ok, ...} or {:error, :conflict}) when possible, or {:error, :unavailable} if the resolution read itself cannot complete.

Mixed-mode note

Transition rule per key:

• LWW -> CAS: allowed.
• CAS -> LWW (eventual put/delete): rejected with {:error, :cas_managed_key}.

Keep lock/ownership keyspaces CAS-write-only after transition.

Garbage collection

A periodic GC timer runs three phases per tick:

1. Observe TTL expiry -- emits :expired events; expired LWW rows become tombstones, expired CAS rows stay lazy/local
2. Purge old tombstones / long-expired CAS rows -- hard-deletes data older than the retention window from SQLite
3. Truncate oplog -- removes oplog entries below the minimum member HWM

Stale database protection

If a node goes away longer than tombstone_ttl and comes back with an old database on disk, other members will have already GC'd the tombstones for entries deleted during the absence. EKV detects this by checking a last_active_at timestamp stored in the database. If the database is too stale, EKV fails startup by default instead of trusting that on-disk state. Operators can then wipe that node's data dir so it rebuilds from members, or explicitly set allow_stale_startup: true when they intend to trust the old on-disk cluster state.

Each shard DB also persists a named schema_version in kv_meta. Fresh databases stamp the current version on first open. Initialized shard DBs with missing or mismatched schema_version fail startup closed so EKV does not silently boot incompatible on-disk state.

Fresh shard DBs also enable SQLite auto_vacuum=INCREMENTAL. This only applies at creation time; EKV does not rewrite existing shard DBs on normal startup just to change SQLite vacuum mode.

Long live-partition protection

A different edge case is when nodes stay up but are partitioned longer than tombstone_ttl. In that window, one side can purge delete tombstones before reconnect.

For shorter outages, oplog retention is anchored independently by member_progress_retention_ttl (default: min(tombstone_ttl, 6 hours)). If a disconnected member rejoins within that window, GC keeps its replay cursor so heal can usually stay on delta instead of falling back to a full sync.

With the default partition_ttl_policy: :quarantine, EKV detects reconnects after a downtime longer than tombstone_ttl and quarantines that member pair instead of syncing potentially unsafe state. Replication stays blocked for that member until an operator rebuilds one side.

Down-since markers are persisted in kv_meta, keyed by node_id when available (fallback: node name), so restart does not clear quarantine history. This also means node-name churn does not bypass quarantine when node_id is stable.

Fallback name-based markers are bounded: EKV prunes very old entries and caps the retained set per shard to avoid unbounded growth over long periods.

Multiple instances

Each EKV instance (identified by :name) is fully independent -- its own SQLite files, shard GenServers, member mesh, and scoped :pg control plane for routing, subscriptions, and shutdown coordination. To isolate replication between groups of nodes, start separate EKV instances with different names on the nodes that should form each group.

# Only nodes in the US region start this:
{EKV, name: :us_data, data_dir: "/data/ekv/us"}

# Only nodes in the EU region start this:
{EKV, name: :eu_data, data_dir: "/data/ekv/eu"}

License

MIT