DrinkMeAlice.Conn (drink_me_alice v0.1.0)

The single supervised connection owner: one Mint/MintWebSocket connection, one outstanding RPC at a time, bounded admission via DrinkMeAlice.Credits before any mailbox send.

Outcome vocabulary (transport facts only)

call/3 returns transport-level facts; DrinkMeAlice.Tx.run/4 (or a caller's own port) maps them to write outcomes:

{:reply, decoded} — a reply for this request id was received and decoded under the bounded profile.
{:lost, :timeout | :closed | :transport | :send_failed | :down} — the request was sent (or may have been); no classifying reply arrived before the deadline. For mutations this is {:unknown, request} — never a claim of failure.
{:refused, :down | :expired | :exhausted | :too_large | :encode | :clock_uncertain} — certified non-send: the request provably never entered a write path (connection down before send, caller deadline already passed while queued, credit bound reached, payload or encoding bound, or — for kind: :mutation requests — clock uncertainty latched at dispatch time). For mutations this is {:unavailable, error}.

Serialization

Requests are handled one at a time — the single socket is FIFO and a serial owner cannot interleave writes. A caller's deadline is carried in the request; a request whose deadline expired before its send is refused (:expired) rather than submitted late, so unavailable always certifies non-submission. A caller that dies while queued is swept by Credits.sweep/1; a caller that dies after send gets its outcome discarded — the durable side is unaffected.

Notifications as hints only

An optional live subscription on one configured table delivers notifications that set a bounded hint_pending flag and fan out to at most 16 subscriber pids as {hint_tag, table, generation}. Notifications never mutate durable reads — callers always read afresh. A notification whose live-id is not the current subscription's is stale (3.0.5's LIVE defect can deliver post-invalidation on a socket we did not replace); it is counted and discarded. On session expiry or auth failure the whole socket is closed and a new connection authenticates/selects/resubscribes — the adapter does not switch identities on a live-subscription socket.

Clock authority (optional)

When clock_query is configured, establishment reads it — a query whose single result is %{"now" => string, "floor" => _, "regressed" => boolean} — and every reply is scanned for a regressed: true result or a clock_regressed decision. Either latches clock_uncertain for this connection generation; a mutation (kind: :mutation) is refused at dispatch — serially in this owner, immediately before its frame is written — while the latch holds. Reads are never gated. clock_authority/1 reads the latch for introspection only; it is not part of any write path. Without clock_query no floor is read and nothing latches.

Configuration

host, port, ns, db, user, pass — required.
signin — :database (default: a database-scoped user, the least-privilege runtime identity), :namespace or :root.
live — nil (default) or a table name to subscribe for hints.
hint_tag — the atom tagging hint messages (default :drink_me_alice_hint).
clock_query — nil (default) or {sql, vars} as above.
poll_ms (default 50), connect_timeout (default 2_000).
name — optional registered name.

Summary

Types

request()

status()

Functions

call(conn, request, opts \\ [])

Send one RPC under the caller's deadline (default 2 s — the ordinary RPC bound). Reserves a credit slot first; see the module doc for the returned transport facts.

clock_authority(conn)

Cached clock authority for introspection: :trusted | :uncertain. Not a write-path gate — mutations are gated at dispatch in this owner.

credits(conn)

The credits table id for Credits.snapshot/1 evidence.

hint_subscribe(conn, pid \\ self())

Subscribe pid (default caller) to work-item hints. Bounded at 16.

hint_take(conn)

Take the coalesced dirty hint: clears the flag, returns times seen.

inject(conn, fault)

Test-only transport fault seam — honest socket-level events, never a bypass of the transaction path

start(config)

start_link(config)

Start the owner. See the module doc for config keys. An invalid configuration is refused before any process starts: {:error, {:invalid_config, key}}.

stats(conn)

Notification/health counters for evidence — no durable claims.

status(conn)

Connection status: :disconnected | :reconciling | :healthy.

Types

request()

@type request() :: %{
  :method => String.t(),
  :params => list() | map(),
  optional(:kind) => :mutation | :read
}

status()

@type status() :: :disconnected | :reconciling | :healthy

Functions

call(conn, request, opts \\ [])

@spec call(GenServer.server(), request(), keyword()) ::
  {:reply, map()} | {:lost, atom() | tuple()} | {:refused, atom()}

Send one RPC under the caller's deadline (default 2 s — the ordinary RPC bound). Reserves a credit slot first; see the module doc for the returned transport facts.

The request payload is installed into the reserved slot atomically (Credits.enqueue/3 — one CAS that publishes payload+state together). The owner drains :enqueued rows on a fixed poll_ms cadence — no caller→owner message exists, so the owner mailbox stays bounded under arbitrarily delayed callers. The owner frees the slot when it consumes the row (replied or deadline-refused): a caller that times out or dies never releases, and a dead holder's :reserved-only slot is reclaimed by Credits.sweep/1. No timing window can orphan a queued payload.

clock_authority(conn)

@spec clock_authority(GenServer.server()) :: :trusted | :uncertain

Cached clock authority for introspection: :trusted | :uncertain. Not a write-path gate — mutations are gated at dispatch in this owner.

credits(conn)

@spec credits(GenServer.server()) :: {:ok, :ets.tid()} | {:error, :no_conn}

The credits table id for Credits.snapshot/1 evidence.

hint_subscribe(conn, pid \\ self())

@spec hint_subscribe(GenServer.server(), pid()) :: :ok | {:error, :full}

Subscribe pid (default caller) to work-item hints. Bounded at 16.

hint_take(conn)

@spec hint_take(GenServer.server()) :: {:hint, non_neg_integer()} | :clean

Take the coalesced dirty hint: clears the flag, returns times seen.

inject(conn, fault)

@spec inject(GenServer.server(), :kill_socket | :drop_after_send | :drop_reply) :: :ok

Test-only transport fault seam — honest socket-level events, never a bypass of the transaction path:

:kill_socket — close the TCP transport now (network-failure shape).
:drop_after_send — after the next request frame is written, close the transport before the reply can arrive (lost-reply shape).
:drop_reply — receive the next request's reply, then close the transport and report {:lost, :reply_dropped} WITHOUT delivering it: a reply arrived but is discarded before classification — it does not prove the write committed — so the caller observes a real lost-reply outcome.

Session-expiry evidence uses REAL server expiry — FOR SESSION credentials issued through Admin helpers — not a transport seam.

start(config)

@spec start(keyword() | map()) :: {:ok, pid()} | {:error, term()}

start_link(config)

@spec start_link(keyword() | map()) :: {:ok, pid()} | {:error, term()}

Start the owner. See the module doc for config keys. An invalid configuration is refused before any process starts: {:error, {:invalid_config, key}}.

stats(conn)

@spec stats(GenServer.server()) :: map()

Notification/health counters for evidence — no durable claims.

status(conn)

@spec status(GenServer.server()) :: status()

Connection status: :disconnected | :reconciling | :healthy.