relay/runtime

The OTP actor that owns one connection’s relay/reducer state and runs its handlers, for authors of custom transports.

start(server, config(), sink) starts the runtime; the transport then submits each inbound frame with send_frame on a fresh relay/reducer.ExchangeId, and the runtime calls sink with OutputWrite and OutputClose for that exchange. A sink that returns Error(Nil) closes only its exchange. exchange_closed reports a peer disconnect, which cancels that exchange’s invocation. notify, register_tool, unregister_tool and end_streams change the live server. The stdio and HTTP transports run on this module.

Each handler runs in its own unlinked process with a timeout. Every cancellation (a peer disconnect, notifications/cancelled, the invocation timeout, close, stop, or the exit of the process that started the runtime) first fires the handler’s relay/tool.cancelled selector, then gives it the cancellation grace period to stop work it started elsewhere and return; Relay kills it only when the grace ends. The runtime stays alive until every cancelled handler has returned or been killed. A crash becomes the JSON-RPC internal error on its exchange only.

Every request has one correlation, which its events, its handler (relay/tool.correlation) and its exchange.closed event carry: the one the transport passes to send_frame, else the one the client sent in the request’s _meta under io.github.gleam-dream/correlation (1 to 128 visible ASCII characters; anything else is ignored), else a fresh sinal/correlation.unique(). A carried value is untrusted telemetry: Relay never uses it to authorize anything, and two clients can send the same one.

Setting	Default	Setter
live exchanges	100	`with_max_live_exchanges`
frame size	1 MiB	`with_max_frame_bytes`
JSON nesting depth	64	`with_max_json_depth`
invocation timeout	30 s	`with_invocation_timeout`
cancellation grace	5 s	`with_cancellation_grace`
tombstone retention	60 s, at most 10,000	`with_tombstone_retention`, `with_max_tombstones`

import gleam/option.{None}
import relay/reducer
import relay/runtime
import relay/server

pub fn run(frame: BitArray) {
  let assert Ok(rt) =
    runtime.start(server.new([]), runtime.config(), fn(output) {
      case output {
        runtime.OutputWrite(_exchange, _bytes) -> Ok(Nil)
        runtime.OutputClose(_exchange) -> Ok(Nil)
      }
    })
  let _ = runtime.send_frame(rt, reducer.new_exchange_id(), Nil, frame, None)
  runtime.stop(rt)
}

Types

Config

opaque </>

Runtime limits. Build it with config() and the with_* setters; start validates it.

pub opaque type Config

ConfigField

</>

The setting validate refused.

pub type ConfigField {
  MaxLiveExchanges
  MaxFrameBytes
  MaxJsonDepth
  InvocationTimeout
  CancellationGrace
  TombstoneRetention
  MaxTombstones
}

Constructors

```
MaxLiveExchanges
```
```
MaxFrameBytes
```
```
MaxJsonDepth
```
```
InvocationTimeout
```
```
CancellationGrace
```
```
TombstoneRetention
```
```
MaxTombstones
```

FrameError

</>

Why send_frame refused a frame. The transport answers the peer.

pub type FrameError {
  FrameTooLarge(size: Int, limit: Int)
  FrameTooDeep(limit: Int)
  TooManyLiveExchanges(current: Int, limit: Int)
  RuntimeStopped
}

Constructors

```
FrameTooLarge(size: Int, limit: Int)
```
```
FrameTooDeep(limit: Int)
```

TooManyLiveExchanges(current: Int, limit: Int)

```
RuntimeStopped
```

Message

opaque </>

The messages a runtime actor receives; used to name a supervised one.

pub opaque type Message(context)

Output

</>

What the runtime asks the transport to do on an exchange.

pub type Output {
  OutputWrite(exchange: reducer.ExchangeId, bytes: BitArray)
  OutputClose(exchange: reducer.ExchangeId)
}

Constructors

OutputWrite(exchange: reducer.ExchangeId, bytes: BitArray)

OutputClose(exchange: reducer.ExchangeId)

Runtime

opaque </>

A running runtime.

pub opaque type Runtime(context)

StartError

</>

Why start failed.

pub type StartError {
  InvalidConfig(field: ConfigField)
  ActorStartFailed(actor.StartError)
}

Constructors

```
InvalidConfig(field: ConfigField)
```
A setting is out of range: counts, sizes and durations must be positive; the grace period may be zero.
```
ActorStartFailed(actor.StartError)
```

Values

close

</>

pub fn close(runtime: Runtime(context)) -> Nil

Closes the connection: cancels every invocation and refuses new frames. Each cancelled handler keeps its grace period. Returns at once.

config

</>

pub fn config() -> Config

The default limits; see the module table.

describe_start_error

</>

pub fn describe_start_error(error: StartError) -> String

A one-line description of a start error.

end_streams

</>

pub fn end_streams(runtime: Runtime(context)) -> Nil

Ends every open subscriptions/listen stream with its result.

exchange_closed

</>

pub fn exchange_closed(
  runtime: Runtime(context),
  exchange: reducer.ExchangeId,
) -> Nil

Reports that the peer closed one exchange; its invocation is cancelled and nothing more is written to it.

invocation_timeout

</>

pub fn invocation_timeout(config: Config) -> duration.Duration

The invocation timeout of this configuration.

max_frame_bytes

</>

pub fn max_frame_bytes(config: Config) -> Int

The largest frame this configuration accepts.

named

</>

pub fn named(
  name: process.Name(Message(context)),
) -> Runtime(context)

The runtime registered under name by supervised.

notify

</>

pub fn notify(
  runtime: Runtime(context),
  notification: subscriptions.Notification,
) -> Nil

Tells every listening stream that asked for this notification.

register_tool

</>

pub fn register_tool(
  runtime: Runtime(context),
  tool: tool.Tool(context),
) -> Nil

Registers a tool and tells listening streams the tool list changed. A duplicate name is ignored.

send_frame

</>

pub fn send_frame(
  runtime: Runtime(context),
  exchange: reducer.ExchangeId,
  context: context,
  bytes: BitArray,
  correlation: option.Option(correlation.Correlation),
) -> Result(Nil, FrameError)

Submits one frame on a fresh exchange, with the context and the correlation the transport built for it. With None, the runtime uses the correlation the frame carries in _meta, or a fresh one. Returns once the runtime has admitted or refused the frame; responses arrive through the sink.

start

</>

pub fn start(
  server: server.Server(context),
  config: Config,
  sink: fn(Output) -> Result(Nil, Nil),
) -> Result(Runtime(context), StartError)

Starts a runtime linked to the caller.

stop

</>

pub fn stop(runtime: Runtime(context)) -> Nil

Closes the connection, then stops the actor once every cancelled handler has returned or been killed at the end of its grace period. Returns when the runtime has stopped, waiting at most the grace period plus 5 s. To stop without waiting, call close and then stop from another process.

supervised

</>

pub fn supervised(
  server: server.Server(context),
  config: Config,
  sink: fn(Output) -> Result(Nil, Nil),
  name: process.Name(Message(context)),
) -> supervision.ChildSpecification(Runtime(context))

A child specification for a runtime registered under name; find it with named. A restart starts a fresh connection state.

unregister_tool

</>

pub fn unregister_tool(
  runtime: Runtime(context),
  name: String,
) -> Nil

Removes a tool and tells listening streams when it existed.

validate

</>

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

Checks the settings without starting anything.

with_cancellation_grace

</>

pub fn with_cancellation_grace(
  config: Config,
  grace: duration.Duration,
) -> Config

How long a cancelled or timed-out handler may keep running after its relay/tool.cancelled selector fires, before Relay kills it.

with_invocation_timeout

</>

pub fn with_invocation_timeout(
  config: Config,
  timeout: duration.Duration,
) -> Config

How long a handler may run before the client receives an internal error.

with_label

</>

pub fn with_label(config: Config, label: String) -> Config

The listener label in this runtime’s telemetry.

with_max_frame_bytes

</>

pub fn with_max_frame_bytes(config: Config, bytes: Int) -> Config

The largest frame send_frame accepts.

with_max_json_depth

</>

pub fn with_max_json_depth(config: Config, depth: Int) -> Config

The deepest object and array nesting a frame may have, checked before parsing.

with_max_live_exchanges

</>

pub fn with_max_live_exchanges(
  config: Config,
  count: Int,
) -> Config

How many exchanges may be open at once; a frame beyond it fails with TooManyLiveExchanges.

with_max_tombstones

</>

pub fn with_max_tombstones(config: Config, count: Int) -> Config

How many finished or cancelled invocations are remembered at most.

with_tombstone_retention

</>

pub fn with_tombstone_retention(
  config: Config,
  retention: duration.Duration,
) -> Config

How long a finished or cancelled invocation is remembered, so a late start or result for it is dropped.