Arcadic.Changes (Arcadic v0.7.0)

Copy Markdown View Source

Live change-events client for ArcadeDB's /ws WebSocket feed — arcadic's one caller-supervised process.

Start it under your own supervision tree, then subscribe/3 a database. The process presents conn.auth on the handshake, keeps the socket drained continuously, and pushes each change to a single subscriber pid as {:arcadic_change, %Arcadic.Changes.Event{}} (see that struct for the marker contract).

Reliability — best-effort at-most-once

ArcadeDB /ws has no replay or checkpoint. On any reconnect the process delivers a :reconnected marker (before re-subscribing); on buffer overflow it delivers an :overflow marker. Any marker obligates the consumer to reconcile the affected database — the feed is a change hint, not a durable log.

The subscriber may also receive an out-of-band error message {:arcadic_change_error, reason}: :unauthorized (a terminal 401/403 on the handshake — the process then stops, no reconnect spin) or :subscribe_rejected (the server rejected a subscribe/unsubscribe with an error frame; non-terminal — the socket stays open). Both reasons are bare atoms — the server's error text is never forwarded.

Subscriber model

One subscriber pid per process. The first subscribe/3 binds the subscriber and monitors it; later subscribes on other databases route to that same pid (demux on Event.database). A subscribe with a different :subscriber is rejected {:error, :subscriber_conflict}. The subscriber's :DOWN stops the process.

Backpressure

A bounded internal buffer (:max_buffer, default 1000) drops the oldest events on overflow, emits an :overflow marker plus a [:arcadic, :changes, :dropped] telemetry count per drain cycle in which events were dropped (a short burst coalesces into a single marker; a sustained overflow emits one roughly per drain cycle), and keeps reading the socket. The buffer bounds arcadic's memory and guarantees a slow subscriber never wedges the server (the socket is drained continuously); it does not bound the subscriber's own mailbox — a persistently slow consumer must apply its own backpressure (or reconcile on the markers and discard).

Absent optional dependency

The WebSocket client rides the optional mint_web_socket dependency. This module always compiles; if the dependency is absent, start_link/1 returns {:error, :mint_web_socket_not_available} at runtime. start_link/1 also rejects a malformed :conn value-free — see Arcadic.Error for the :invalid_auth / :invalid_url_scheme / :invalid_max_buffer reasons.

Telemetry

[:arcadic, :changes, :start | :stop | :disconnect | :dropped], all with value-free metadata %{operation: :changes} (no database, record, or values).

Summary

Functions

Returns a specification to start this module under a supervisor.

Start the change-events process.

Subscribe client to change events on database.

Unsubscribe client from change events on database.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

start_link(opts)

@spec start_link(keyword()) ::
  GenServer.on_start()
  | {:error,
     :mint_web_socket_not_available
     | :invalid_conn
     | :invalid_auth
     | :invalid_url_scheme
     | :invalid_max_buffer
     | :invalid_establish_timeout}

Start the change-events process.

Options

  • :conn — an Arcadic.Conn (required). Supplies the /ws endpoint (base_url) and handshake credential (auth).
  • :name — optional GenServer name.
  • :max_buffer — bounded delivery buffer (default 1000).

Returns {:error, :mint_web_socket_not_available} when the optional mint_web_socket dependency is absent.

subscribe(client, database, opts \\ [])

@spec subscribe(GenServer.server(), String.t(), keyword()) ::
  :ok | {:error, :subscriber_conflict}

Subscribe client to change events on database.

Options

  • :type — restrict to a document/vertex/edge type (default: all types).
  • :change_types — a subset of [:create, :update, :delete] (default: all).
  • :subscriber — the pid to deliver events to. Defaults to self() evaluated here in the caller's process (never the process running the GenServer), so a supervised Changes still routes to the subscribing caller.

The first subscribe binds the subscriber; a later subscribe with a different subscriber returns {:error, :subscriber_conflict}.

unsubscribe(client, database)

@spec unsubscribe(GenServer.server(), String.t()) :: :ok

Unsubscribe client from change events on database.