weft/actor

A strict superset of gleam/otp/actor’s builder, on a receive loop weft owns.

Why a loop of our own

gleam/otp/actor is the right shape and this module does not try to improve on it: new, on_message, start, continue, stop mean exactly what they mean upstream, Started, StartError and ChildSpecification are upstream’s own types rather than copies, and an actor built here drops into a gleam_otp supervisor unchanged.

What upstream cannot give us is a way in. Builder, Initialised and Next are opaque, so there is no wrapping trick that adds a field to them: a handle_continue analog has to be able to put a message somewhere the loop looks before the mailbox, and only the loop can own that place. The same is true of the three smaller gaps that ride along — a terminate/2 analog, hibernation, and a loop timeout — each of which is a decision made between receive and the callback. So weft reimplements the loop and keeps the surface identical, which is the trade that makes migration a change of import line.

The continue queue, and what its order guarantees

gen_server’s {continue, Term} closes a race that every actor with expensive initialisation hits. Sending yourself a message from the initialiser looks equivalent and is not: start returns, the parent hands the subject to a client, and the client’s first request can be ahead of your own message in the mailbox. The actor then serves a query against half-built state, and grows a “not ready yet” case arm that exists only to paper over the timing.

continuing puts the message in a queue the loop drains before it looks at the mailbox at all, and the queue is populated before the acknowledgement that unblocks start. There is no interleaving in which an external message wins, because external messages are not consulted while the queue is non-empty.

The ordering contract, which is observable and which callers will depend on:

Injected messages run ahead of everything in the mailbox.
Within one Next or one Initialised, they run in the order given: continue(s) |> then_handle(A) |> then_handle(B) handles A, then B.
A message injected while handling a queued one runs before the rest of the queue. The whole block goes to the front, depth-first, which is what gen_statem’s next_event does and what makes then_handle usable as a small state machine.
System messages are not starved by any of it. The loop checks the debug plane between every two injected messages, so a sys:suspend/1 arriving in the middle of a continue chain takes effect there rather than after the chain drains.

Example

pub type Message {
  LoadIndex
  Query(term: String, reply: process.Subject(List(String)))
}

pub fn start_search() {
  actor.new_with_initialiser(1000, fn(subject) {
    // Returns immediately: the supervisor is unblocked and the actor is
    // registered and reachable. The slow index load still happens before
    // the first Query is served, with no race.
    actor.initialised(Empty)
    |> actor.returning(subject)
    |> actor.continuing(LoadIndex)
    |> Ok
  })
  |> actor.on_message(handle)
  |> actor.start
}

fn handle(state: Index, message: Message) -> actor.Next(Index, Message) {
  case message {
    LoadIndex -> actor.continue(Loaded(read_index_from_disk()))
    Query(term:, reply:) -> {
      process.send(reply, search(state, term))
      actor.continue(state)
    }
  }
}

Types

Builder

opaque </>

A description of an actor, ready to start or to hand to a supervisor.

Built with new or new_with_initialiser and refined with the setters below.

pub opaque type Builder(state, message, return)

Initialised

opaque </>

The outcome of an actor’s initialiser: the starting state, the selector to receive with, the value to hand back to the parent, and any messages to handle before the mailbox.

Built with initialised and refined with selecting, returning and continuing.

pub opaque type Initialised(state, message, data)

Linkage

</>

Whether start links the new process to the process that started it.

pub type Linkage {
  Linked
  Unlinked
}

Constructors

```
Linked
```
The default, and what OTP does: the starter and the actor share fate down a link, and a supervisor is that starter.
```
Unlinked
```
No link. The actor is started by a process that must neither die with it nor take it down: a guard started from the consumer it serves, a holder that must outlive the host that created it. The starter still learns of a start failure through the acknowledgement, and can monitor the pid it gets back for everything after.

opaque </>

What an actor does after handling a message.

Built with continue, stop or stop_abnormal, and refined with with_selector and then_handle.

pub opaque type Next(state, message)

StartError

</>

Why an actor failed to start: the initialiser timed out, returned an error, or the process died while running it.

gleam/otp/actor’s own type, so a supervisor cannot tell a weft actor from a gleam_otp one.

pub type StartError =
  actor.StartError

StartResult

</>

The result of starting an actor.

gleam/otp/actor’s own alias, for the same interoperability reason as Started.

pub type StartResult(data) =
  Result(actor.Started(data), actor.StartError)

Started

</>

What the parent is given when an actor starts: the actor’s pid and whatever its initialiser chose to return.

This is gleam/otp/actor’s own type, not a copy of it, so a weft actor can be started by anything that expects a gleam_otp one.

pub type Started(data) =
  actor.Started(data)

Values

call

</>

pub fn call(
  subject: process.Subject(message),
  waiting timeout: Int,
  sending make_message: fn(process.Subject(reply)) -> message,
) -> reply

Send a message and wait for the reply.

The caller crashes if no reply arrives within the timeout, rather than carrying on against a process that may be in an unknown state. A re-export of process.call.

Examples

let assert Ok("Robert") = actor.call(subject, waiting: 10, sending: Pop)

continue

</>

pub fn continue(state: state) -> Next(state, message)

Continue, processing any waiting or future messages.

Examples

fn handle(count: Int, _message: Tick) -> actor.Next(Int, Tick) {
  actor.continue(count + 1)
}

continuing

</>

pub fn continuing(
  initialised: Initialised(state, message, return),
  message: message,
) -> Initialised(state, message, return)

Handle message before anything in the mailbox.

This is gen_server’s handle_continue: the initialiser returns at once, so start unblocks, the supervisor moves on and the name is registered, and the expensive setup still runs before the first external message. The guarantee is not statistical — the queue is filled before the acknowledgement that releases start, and the loop never looks at the mailbox while the queue is non-empty.

Several calls run in the order they were written, all of them before the mailbox. A continue handler may inject more with then_handle.

Examples

// `LoadIndex` is handled before any request the parent sends after
// `start` returns.
actor.new_with_initialiser(1000, fn(subject) {
  actor.initialised(Empty)
  |> actor.returning(subject)
  |> actor.continuing(LoadIndex)
  |> Ok
})

hibernate_after

</>

pub fn hibernate_after(
  builder: Builder(state, message, return),
  ms: Int,
) -> Builder(state, message, return)

Hibernate the actor after ms milliseconds with no messages.

Hibernation runs a full garbage collection and drops the process stack, leaving the actor at its minimum footprint until the next message wakes it. It is for the ten-thousand-mostly-idle-actors case, where the aggregate heap matters more than the microseconds a wake-up costs; an actor that receives steadily should not be given it.

The mechanism is a receive timeout rather than a timer message, so nothing is ever queued and there is no stale fire to discard: the actor hibernates from inside the receive it was already blocked in, and erlang:hibernate/3 resumes it into the same loop with the same state when a message arrives.

Examples

actor.new(state)
|> actor.hibernate_after(60_000)
|> actor.on_message(handle)

idle_timeout

</>

pub fn idle_timeout(
  builder: Builder(state, message, return),
  ms: Int,
  message: message,
) -> Builder(state, message, return)

Handle message after ms milliseconds with no messages.

This is gen_server’s loop timeout: the clock is reset every time the actor handles a message, so it fires only after a genuinely quiet stretch, and the message is an ordinary one delivered to the ordinary handler.

The mechanism is a named timer from weft/internal/timer rather than a receive timeout, because an actor may also be hibernating, and a receive cannot have two deadlines. That choice brings a race with it, and the race is closed rather than tolerated: the timer can fire in the instant between a real message arriving and the actor resetting the clock, which would put a timeout message in the mailbox that no longer describes anything true. Every arming carries a generation stamp, and the timer book drops a fire whose stamp is no longer current, so a message already in flight when the clock is reset is discarded before it reaches the handler. The visible behaviour is the one the name promises: traffic cancels the timeout.

The timeout is not armed while the actor is suspended by sys:suspend/1 — a suspended actor is not idle, it is frozen — and is re-armed on resume.

Examples

// A connection that closes itself after five minutes of quiet.
actor.new(connection)
|> actor.idle_timeout(300_000, IdleTooLong)
|> actor.on_message(fn(state, message) {
  case message {
    IdleTooLong -> actor.stop()
    Request(r) -> actor.continue(serve(state, r))
  }
})

initialised

</>

pub fn initialised(
  state: state,
) -> Initialised(state, message, Nil)

Take the post-initialisation state of the actor.

Examples

actor.new_with_initialiser(1000, fn(subject) {
  actor.initialised(0) |> actor.returning(subject) |> Ok
})

named

</>

pub fn named(
  builder: Builder(state, message, return),
  name: process.Name(message),
) -> Builder(state, message, return)

Register the actor under name when it starts, so it can be reached by a named subject rather than by passing a subject around.

If the name is already registered the actor fails to start. When this is used the actor’s default subject is the named one, which is what lets a restarted actor take over from the one it replaced.

Examples

let name = process.new_name("cache")
let assert Ok(_) =
  actor.new(dict.new()) |> actor.named(name) |> actor.start
process.send(process.named_subject(name), Put("k", "v"))

new

</>

pub fn new(
  state: state,
) -> Builder(state, message, process.Subject(message))

Describe an actor with no custom initialisation.

The actor hands the parent a subject to send messages on — a named subject if named was used.

Examples

let assert Ok(started) =
  actor.new([]) |> actor.on_message(handle) |> actor.start
process.send(started.data, Push("Joe"))

new_with_initialiser

</>

pub fn new_with_initialiser(
  timeout: Int,
  initialise: fn(process.Subject(message)) -> Result(
    Initialised(state, message, return),
    String,
  ),
) -> Builder(state, message, return)

Describe an actor with initialisation that runs in the new process before start returns.

timeout is how many milliseconds the initialiser has; overrunning it kills the actor and fails the start with InitTimeout. The actor’s default subject is passed in — return it to the parent with returning, use it some other way, or ignore it.

Work that does not have to happen before start returns should not: continuing runs it after the acknowledgement and still before the first external message, which is what a supervisor’s start timeout wants.

Examples

actor.new_with_initialiser(1000, fn(subject) {
  use table <- result.try(open_table())
  actor.initialised(table) |> actor.returning(subject) |> Ok
})
|> actor.on_message(handle)
|> actor.start

on_message

</>

pub fn on_message(
  builder: Builder(state, message, return),
  handler: fn(state, message) -> Next(state, message),
) -> Builder(state, message, return)

Set the message handler.

The actor handles messages one at a time, in the order it receives them, with injected messages first — see this module’s header.

Examples

actor.new(0) |> actor.on_message(fn(count, _) { actor.continue(count + 1) })

on_shutdown

</>

pub fn on_shutdown(
  builder: Builder(state, message, return),
  handler: fn(state, process.ExitReason) -> Nil,
) -> Builder(state, message, return)

Run handler with the final state as the actor shuts down.

This is a terminate/2 analog and it is best effort. Read the list before relying on it:

It runs when a handler returns stop or stop_abnormal, with that reason.
It runs when a trapped exit signal shuts the actor down, with the reason from the signal — but only if trapping_exits(True) was set. Without trapping, an exit signal kills the process directly and no Gleam code runs.
It never runs when the actor is killed with an untrappable signal (process.kill, or a supervisor’s brutal kill after a shutdown timeout). Nothing runs then; that is what untrappable means.
It never runs if initialisation fails, because there is no state to hand it.
It does not run if the actor’s own handler crashes.

So it is the right place to release something whose loss is an inconvenience, and the wrong place for anything whose loss is a correctness problem. If it must happen, it belongs with a process that monitors this one, not here.

The reason passed may itself be Killed — that is a linked process having been killed, which this actor was told about and is shutting down because of, not this actor being killed.

Examples

actor.new(state)
|> actor.trapping_exits(True)
|> actor.on_shutdown(fn(state, reason) {
  log("closing " <> state.name <> ": " <> string.inspect(reason))
})

returning

</>

pub fn returning(
  initialised: Initialised(state, message, old_return),
  return: return,
) -> Initialised(state, message, return)

Set the value handed back to the parent when the actor has started.

Commonly the subject the actor receives on, so the parent can send to it.

Examples

actor.new_with_initialiser(1000, fn(subject) {
  actor.initialised(state) |> actor.returning(subject) |> Ok
})

selecting

</>

pub fn selecting(
  initialised: Initialised(state, message, return),
  selector: process.Selector(message),
) -> Initialised(state, message, return)

Give the actor a selector to receive messages with.

This replaces the default selector, which selects only the actor’s own subject, so a custom selector must select that subject itself if the actor is still to receive on it. A message that arrives and is not selected for is discarded with a warning.

Deviation from `gleam/otp/actor`

Upstream’s selecting may change the actor’s message type, because upstream’s Initialised mentions that type only in the selector field. Here the injected-message queue mentions it too, so changing the type would mean silently discarding anything continuing had already added. The signature therefore fixes the message type instead. This rejects no program upstream accepts: before selecting, an upstream Initialised always has an as-yet-unbound message type, and unifying it with the selector’s is exactly what the type-changing version does.

Examples

actor.new_with_initialiser(1000, fn(subject) {
  let selector =
    process.new_selector()
    |> process.select(subject)
    |> process.select_map(pubsub_subject, Broadcast)
  actor.initialised(state)
  |> actor.selecting(selector)
  |> actor.returning(subject)
  |> Ok
})

send

</>

pub fn send(
  subject: process.Subject(message),
  message: message,
) -> Nil

Send a message to an actor.

A re-export of process.send, for convenience.

Examples

actor.send(started.data, Push("Joe"))

start

</>

pub fn start(
  builder: Builder(state, message, return),
) -> Result(actor.Started(return), actor.StartError)

Start an actor from a builder.

The new process is linked to the caller, and the caller blocks until the initialiser has finished or the initialisation timeout expires. Messages added with continuing are handled after this returns and before any message sent afterwards.

Examples

let assert Ok(started) =
  actor.new(0) |> actor.on_message(handle) |> actor.start
// -> `started.data` is a subject; `started.pid` is the actor

stop

</>

pub fn stop() -> Next(state, message)

Stop and shut down, handling no further messages.

The exit reason is Normal, so a Permanent child restarts and a Transient one does not. An on_shutdown callback runs first, with the state as it stood.

Examples

fn handle(state: State, message: Message) -> actor.Next(State, Message) {
  case message {
    Shutdown -> actor.stop()
    Work -> actor.continue(state)
  }
}

stop_abnormal

</>

pub fn stop_abnormal(reason: String) -> Next(state, message)

Stop and shut down abnormally, propagating the reason to linked processes.

An on_shutdown callback runs first, with the state as it stood, which is the one chance to release anything the actor still owns before the exit signal goes out.

Examples

fn handle(state: State, message: Message) -> actor.Next(State, Message) {
  case message {
    Corrupted -> actor.stop_abnormal("index checksum mismatch")
    Work -> actor.continue(state)
  }
}

supervised

</>

pub fn supervised(
  builder: Builder(state, message, return),
) -> supervision.ChildSpecification(return)

Describe this actor as a supervisor’s child.

Returns gleam/otp/supervision’s own ChildSpecification, so a weft actor is added to a gleam_otp supervisor exactly as an upstream one is. The default is a permanent worker with a five-second shutdown; refine it with supervision.restart, supervision.timeout and friends.

Examples

supervisor.new(supervisor.OneForOne)
|> supervisor.add(actor.supervised(cache_builder))
|> supervisor.start

then_handle

</>

pub fn then_handle(
  next: Next(state, message),
  message: message,
) -> Next(state, message)

Handle message next, ahead of the mailbox.

This is continuing from a message handler, and the second half of the handle_continue analog: a handler can schedule follow-up work that is guaranteed to run before any client request, without the “not ready yet” arm a self-send would need.

Ordering is the contract in this module’s header, and it is worth repeating because it is what callers depend on. Messages injected by one handler run in the order they were added, and the whole block runs before anything queued earlier — depth-first, as gen_statem’s next_event is. A handler that injects on every call therefore never lets the queue drain, and never reads its mailbox again; that is a live-lock the type system cannot catch.

Applying this to a stop value does nothing: an actor that is stopping handles no more messages.

Examples

// `Reindex` is handled before any client request already in the mailbox.
actor.continue(state) |> actor.then_handle(Reindex)

// `Flush` is handled first, then `Compact`, then the mailbox.
actor.continue(state)
|> actor.then_handle(Flush)
|> actor.then_handle(Compact)

trapping_exits

</>

pub fn trapping_exits(
  builder: Builder(state, message, return),
  trap: Bool,
) -> Builder(state, message, return)

Choose whether the actor traps exits.

A trapping actor receives an exit signal from a linked process as a message instead of dying of it, which is what makes an orderly shutdown possible; it is the prerequisite for on_shutdown firing on anything other than the actor’s own stop.

The policy the loop applies to a trapped exit mirrors what would have happened without trapping, plus the chance to run on_shutdown first: an exit from the parent shuts the actor down whatever the reason — including Normal, as OTP behaviours do, because a child outliving its parent is a leak — and an exit from any other linked process shuts it down only if the reason is abnormal.

Trapping is off by default, and turning it on changes what kills the actor: process.kill still does, process.send_exit no longer does.

Examples

actor.new(state)
|> actor.trapping_exits(True)
|> actor.on_shutdown(fn(state, _reason) { close(state.handle) })

unlinked

</>

pub fn unlinked(
  builder: Builder(state, message, return),
) -> Builder(state, message, return)

Start the actor without linking it to its starter.

Consumers that need this otherwise pay for it with a throwaway process that starts the actor and exits, which leaves the actor linked to a corpse; this is that arrangement made a setting. Only start reads it — a supervisor always links its children, so supervised ignores it.

Examples

builder |> actor.unlinked |> actor.start

with_selector

</>

pub fn with_selector(
  value: Next(state, message),
  selector: process.Selector(message),
) -> Next(state, message)

Replace the selector the actor receives with going forward.

This replaces any selector given by the initialiser or by an earlier Next, rather than adding to it, so a selector that no longer selects the actor’s own subject stops receiving on it.

Applying this to a stop value does nothing: there is no “going forward”.

Examples

actor.continue(state)
|> actor.with_selector(process.new_selector() |> process.select(subject))

Why a loop of our own

The continue queue, and what its order guarantees

Example

Constructors

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Deviation from gleam/otp/actor

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Examples

Deviation from `gleam/otp/actor`