beryl_protocol

A deterministic channel protocol. Hosts supply time, decoded messages and application decisions, then execute returned actions in list order.

Admission and cleanup are barriers: feed their completion back before later external input. DrainBeforeCleanup lets committed worker results reply and push until the host reports cleanup complete. Raw callbacks use InvalidateImmediately and finish unanswered joins with CallbackFinished.

Connection epochs are trusted host-issued unique strings, not socket IDs or client references. Never reuse an epoch for another connection.

Types

`Action`

pub type Action {
  DeliverJoin(
    channel: Channel,
    operation: Operation,
    topic: String,
    payload: dynamic.Dynamic
  )
  AdmitJoin(
    channel: Channel,
    operation: Operation,
    topic: String
  )
  Joined(
    channel: Channel,
    topic: String
  )
  DeliverMessage(
    channel: Channel,
    topic: String,
    event: String,
    payload: dynamic.Dynamic,
    operation: option.Option(Operation)
  )
  CloseChannel(
    channel: Channel,
    operation: Operation,
    topic: String,
    reason: CloseReason
  )
  Send(Outbound)
  CloseConnection(reason: CloseReason)
  Deadline(deadline_ms: option.Option(Int))
}

Execute in order. Never bypass an admission or cleanup barrier on failure.

`Channel`

pub type Channel

One topic instance, independent of all wire references.

`CloseMode`

pub type CloseMode {
  InvalidateImmediately
  DrainBeforeCleanup
}

When the host invalidates a closing channel's pending replies and pushes.

`CloseReason`

pub type CloseReason {
  Normal
  Shutdown
  HeartbeatTimeout
  Fault(String)
}

Logical channel/connection closure reason.

`Config`

pub type Config

Validated resource limits. Payload byte admission remains the host's job.

`Connection`

pub type Connection

Trusted connection identity.

`Decoded`

pub type Decoded {
  Join(
    topic: String,
    payload: dynamic.Dynamic,
    join_ref: option.Option(String),
    ref: option.Option(String)
  )
  Message(
    topic: String,
    event: String,
    payload: dynamic.Dynamic,
    join_ref: option.Option(String),
    ref: option.Option(String)
  )
  Leave(
    topic: String,
    join_ref: option.Option(String),
    ref: option.Option(String)
  )
  Heartbeat(ref: option.Option(String))
}

Codec-normalized inbound frames. Framing and rate limiting stay in the host.

`Error`

pub type Error {
  InvalidConfig
  InvalidIdentity
  InvalidClock
  CounterExhausted
  InvalidSnapshot
  UnsupportedVersion(Int)
  SnapshotTooLarge
}

A rejected transition commits no state or actions.

`Input`

pub type Input {
  Receive(Decoded)
  AcceptJoin(
    operation: Operation,
    payload: json.Json,
    close_mode: CloseMode
  )
  RejectJoin(
    operation: Operation,
    payload: json.Json
  )
  CallbackFinished(operation: Operation)
  AdmissionFinished(
    operation: Operation,
    accepted: Bool
  )
  Reply(
    operation: Operation,
    status: Status,
    payload: json.Json
  )
  Discard(operation: Operation)
  Push(
    channel: Channel,
    event: String,
    payload: json.Json
  )
  BeginClose(
    channel: Channel,
    reason: CloseReason
  )
  DrainFinished(
    operation: Operation,
    reason: CloseReason
  )
  CleanupFinished(operation: Operation)
  CheckDeadline
  Disconnect(reason: CloseReason)
}

A transition input. Stale capabilities and completions are successful no-ops.

Constructors

`DrainFinished`

DrainFinished(
  operation: Operation,
  reason: CloseReason
)

Finish committed work and retain its final close outcome for cleanup.

`Operation`

pub type Operation

Single-use join, reply, admission or cleanup capability.

`Outbound`

pub type Outbound {
  ReplyMessage(
    join_ref: option.Option(String),
    ref: option.Option(String),
    topic: String,
    status: Status,
    payload: json.Json
  )
  PushMessage(
    topic: String,
    event: String,
    payload: json.Json
  )
  TerminalMessage(
    join_ref: option.Option(String),
    topic: String,
    reason: CloseReason
  )
  HeartbeatReply(ref: option.Option(String))
}

Logical outbound messages, translated through the host's active codec.

`State`

pub type State

Per-connection protocol state. Contains no host resources or app models.

`Status`

pub type Status {
  StatusOk
  StatusError
}

Reply status before framing.

`ValidationError`

pub type ValidationError {
  EmptyTopic
  InvalidFormat(String)
}

Shared validation errors, translated by beryl's existing public helpers.

Functions

`can_join`

pub fn can_join(
  State,
  String
) -> Bool

Whether a new join can fit, accounting for same-topic replacement.

`can_push`

pub fn can_push(
  State,
  Channel
) -> Bool

Whether pushes/committed worker replies can still target this instance.

`channel`

pub fn channel(
  State,
  String
) -> Result(Channel, Nil)

Current established channel, excluding channels whose close has begun.

`close_reason`

pub fn close_reason(State) -> option.Option(CloseReason)

The connection's teardown reason, once teardown begins.

`config`

pub fn config() -> Config

Default limits: 256 topic bytes, 64 event bytes, 1000 channels, 1024 pending application operations and a 60-second heartbeat timeout.

`connection`

pub fn connection(State) -> Connection

The connection identity, for host-owned indexes.

`describe_error`

pub fn describe_error(Error) -> String

Describe a rejected transition without application data.

`event_allowed`

pub fn event_allowed(
  String,
  Int
) -> Bool

Shared inbound event check for hosts with existing configuration types.

`export`

pub fn export(
  State,
  Int
) -> Result(String, Error)

Export bounded version-1 JSON, without callbacks, application payloads or clock-origin timestamps. Save a trusted wall-clock checkpoint beside this string in the host attachment envelope.

`instance_ref`

pub fn instance_ref(
  State,
  String
) -> option.Option(String)

Wire reference for an instance, including one awaiting host cleanup.

`join_wire_key`

pub fn join_wire_key(
  State,
  Operation
) -> Result(#(String, option.Option(String), option.Option(String)), Nil)

Wire metadata for a host's join-handle wrapper, never for capability lookup.

`joined_ref`

pub fn joined_ref(
  State,
  String
) -> option.Option(String)

Current channel's wire join reference. Missing references are not identities.

`new`

pub fn new(
  Config,
  String,
  Int
) -> Result(State, Error)

Start a connection. The host must issue a new unique epoch on reconnect.

`next_deadline`

pub fn next_deadline(State) -> option.Option(Int)

The next required timer instant. Expiry is strictly after the timeout.

`pending_cleanup`

pub fn pending_cleanup(
  State,
  String
) -> Result(Operation, Nil)

Retrieve the current generation-scoped cleanup barrier.

`pending_count`

pub fn pending_count(State) -> Int

Number of retained join/reply operations; host byte limits are independent.

`pending_join`

pub fn pending_join(
  State,
  String
) -> Result(#(Channel, Operation), Nil)

Retrieve the outstanding authorization handle for a host callback.

`push_channel`

pub fn push_channel(
  State,
  String
) -> Result(Channel, Nil)

A push-capable instance, including a draining worker.

`reply_pending`

pub fn reply_pending(
  State,
  Operation
) -> Bool

Whether a reply operation is still live.

`reply_wire_key`

pub fn reply_wire_key(
  State,
  Operation
) -> Result(#(String, option.Option(String), option.Option(String)), Nil)

Wire metadata for a host's backwards-compatible public handle wrapper. These fields are not the operation's identity.

`reserved_topic`

pub fn reserved_topic(String) -> Bool

Whether the topic is reserved for beryl's host machinery.

`resolve_topic`

pub fn resolve_topic(
  State,
  String
) -> String

Resolve an opt-in topicless event without guessing between channels.

`restore`

pub fn restore(
  String,
  Int,
  Int
) -> Result(#(State, List(Action)), Error)

Restore using elapsed wall-clock milliseconds validated by the host. A backward or unavailable wall clock must close/reconnect, not resume.

Established channels without lost work retain their generation. Channels with unrecoverable authorization, replies or cleanup are invalidated and receive phx_error, requiring fresh authorization on rejoin. Counters never rewind. Rehydrate host authority before delivering another command.

`stale_join_ref`

pub fn stale_join_ref(
  State,
  String,
  option.Option(String)
) -> Bool

Preserve Phoenix's omitted-reference behavior: only two differing supplied references prove that a frame belongs to an old channel instance.

`step`

pub fn step(
  State,
  Input,
  Int
) -> Result(#(State, List(Action)), Error)

Apply one input without clock reads, IO, global state, or host callbacks.

`topic_allowed`

pub fn topic_allowed(
  String,
  Int
) -> Bool

Shared inbound topic check for hosts with existing configuration types.

`topics`

pub fn topics(State) -> List(String)

Sorted established topics, used for host delivery-index reconstruction.

`valid_event`

pub fn valid_event(
  Config,
  String
) -> Bool

Apply inbound application-event limits, including reserved phx_ names.

`valid_topic`

pub fn valid_topic(
  Config,
  String
) -> Bool

Apply inbound topic byte and syntax limits.

`validate_event`

pub fn validate_event(String) -> Result(String, ValidationError)

Validate event syntax without adding protocol-only restrictions.

`validate_pattern`

pub fn validate_pattern(String) -> Result(String, ValidationError)

Validate pattern syntax without adding topic-only restrictions.

`validate_topic`

pub fn validate_topic(String) -> Result(String, ValidationError)

Validate the syntax shared with beryl's public topic helper.

`with_limits`

pub fn with_limits(
  Config,
  max_topic_bytes: Int,
  max_event_bytes: Int,
  max_channels: Int,
  max_pending: Int,
  heartbeat_timeout_ms: Int
) -> Result(Config, Error)

Set finite limits. Pending operations include joins and deferred replies; cleanup barriers are bounded separately by the channel limit.

`with_topicless_events`

pub fn with_topicless_events(
  Config,
  Bool
) -> Config

Enable topicless application events only when exactly one channel is joined.