weft/state_machine
A typed gen_statem: a state ADT, a data value, postponed events, and
the three timeout kinds, on a receive loop weft owns.
Why Gleam ships half of gen_statem’s surface
Erlang’s gen_statem has two callback modes. state_functions
dispatches on a state atom by naming a function per state;
handle_event_function hands every event to one callback and leaves the
dispatch to you. The split exists because a state is an atom and Erlang
cannot check that every state answered every event.
Gleam does not need it. A state ADT plus case state, message gives the
ergonomics of state_functions and exhaustiveness: add a variant and
every unhandled pair is a compile error. So this module ships the single
callback only, and one whole axis of gen_statem disappears with nothing
lost.
Why a loop of its own, rather than weft/actor
The two features that earn this module — postpone replay and the state
timeout — are not things a handler can do to itself. Replay has to put
events ahead of the mailbox in their original arrival order at the
moment the state changes, and a state timeout has to be cancelled by a
change of state the handler never mentions. Both are decisions the loop
makes between receive and the callback, so they belong to a loop, and
building them into weft/actor would mean the actor growing a state
concept it deliberately does not have. What is shared instead is the
machinery underneath: weft/internal/sys answers the debug plane and
weft/internal/timer keeps the timer book, exactly as they do for the
actor.
State and data are separate on purpose
The state a machine is in and the data it carries are separate
type parameters. The split is load-bearing rather than cosmetic: a state
timeout is cancelled by a change of state and is indifferent to
data, and postponed events are replayed by a change of state and
never by a change of data. Fold the two together and neither rule can
be stated.
The timeout taxonomy
Three kinds, distinguished only by what cancels them. All three are the same mechanism underneath — a generation-stamped entry in the timer book — so a fire that beat its own cancellation into the mailbox is recognised and dropped rather than handled.
| Kind | Armed by | Cancelled by |
|---|---|---|
| State | with_state_timeout | a transition to a different state |
| Event | with_event_timeout | the next event of any kind |
| Named | with_named_timeout | cancel_timeout under the same name |
Each also dies by firing, since all three are one-shot.
The state timeout’s rule is gen_statem’s, down to the awkward corner:
transition(to: s, data: d) where s is the state the machine is
already in is not a state change. It cancels no state timeout, it
replays no postponed event, and it runs no enter callback. keep never
cancels a state timeout either. Only a move to a state that compares
unequal — structurally, by Gleam’s == — counts.
The event timeout measures quiet, so anything that breaks the quiet
cancels it: a message from the mailbox, a message injected with
then_handle, a replayed postponed event, and any other timeout firing.
A handler that wants the deadline to continue must arm it again, which
is gen_statem’s rule and not an oversight.
A named timeout survives everything until it fires or is cancelled by name. It is the one for a deadline that belongs to a piece of work rather than to a state.
Postpone, and the replay contract
postpone re-queues the event being handled. It is redelivered on the
next change of state, ahead of the mailbox, in the order the events
originally arrived, exactly once each — and an event may be postponed
again in the new state, as often as it takes. This is the action that
deletes hand-rolled pending queues, along with the ordering bug and the
“not ready yet” case arm that come with them.
The full ordering after a transition to a different state, which is observable and which callers will depend on:
- The enter callback runs, immediately and in the calling process. It is a callback, not a queued event.
- Messages the enter callback injected with
then_handle, in the order written. - Messages the event handler injected with
then_handle, in the order written. - Postponed events, in original arrival order.
- The mailbox.
Two and three are one rule rather than two: an injected block always
goes to the front of the pending work, depth-first, exactly as
weft/actor’s then_handle and gen_statem’s next_event do. The enter
callback ran last, so its block is in front. A handler that injects on
every call never lets the queue drain and never reads its mailbox again;
that is a live-lock the type system cannot catch.
System messages are not starved by any of it. The debug plane is checked
between every two pending messages, so a sys:suspend/1 arriving in the
middle of a replay takes effect there rather than after it drains.
Enter callbacks
on_enter runs on every transition where the state actually changed,
and once for the initial state — gen_statem makes the initial state
enter call too, and so do we. Since a same-state transition is not a
state change, from == to means exactly one thing: this is the initial
call. That is a useful thing for a callback to be able to test, and it
is the reason the initial call passes the state twice rather than
inventing an Option.
An enter callback may transition again, which runs another enter call, and may stop the machine. It may not postpone, because there is no event in hand to postpone: an enter call is caused by a state change, not by an event.
That last rule is enforced by the type system rather than documented and
ignored at runtime. Step carries a fourth type parameter that says
whether the value came from a place where an event is in hand;
Next and Enter are the two aliases of it, on_event takes a handler
returning the first and on_enter a handler returning the second, and
postpone only accepts the first. Writing postpone in an enter
callback is a compile error naming Postponable and Unpostponable.
The alternative was a second family of constructors and actions for
enter callbacks — enter_keep, enter_then_handle, and so on — which
doubles the surface to be documented and learned for the sake of one
forbidden action. A marker parameter buys the same guarantee with one
set of functions, and the cost is confined to two type names a reader
meets in an error message they will only ever see once.
Example
The connection manager, which is the canonical gen_statem example because every part of this module shows up in it:
pub type Link {
Connecting
Ready
Backoff
}
pub type Message {
Attempt
Established
Request(body: String)
}
fn handle(state: Link, data: Session, message: Message) {
case state, message {
// A request that arrives before the link is up is not an error and
// not a special case: it waits in the machine and is redelivered the
// instant we reach Ready, in the order it arrived.
Connecting, Request(..) | Backoff, Request(..) ->
sm.keep(data) |> sm.postpone
Connecting, Attempt ->
case dial(data) {
Ok(session) -> sm.transition(to: Ready, data: session)
// The retry is a state timeout, so it dies with the state: a
// stale Attempt can never arrive after something else has
// already moved us out of Backoff.
Error(_) ->
sm.transition(to: Backoff, data:)
|> sm.with_state_timeout(after: 1000, sending: Attempt)
}
Ready, Request(body:) -> sm.keep(send(data, body))
Backoff, Attempt -> sm.transition(to: Connecting, data:)
Connecting, Established | Ready, Established | Ready, Attempt |
Backoff, Established -> sm.keep(data)
}
}
// Entering Connecting starts the attempt, so every path into the state
// gets the effect — including the paths added next year.
fn entered(_from: Link, to: Link, data: Session) {
case to {
Connecting -> sm.keep(data) |> sm.then_handle(Attempt)
Ready | Backoff -> sm.keep(data)
}
}
pub fn start_link() {
sm.new(Connecting, new_session())
|> sm.on_event(handle)
|> sm.on_enter(entered)
|> sm.start
}
Types
A description of a state machine, 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, data, message, return)
What an enter callback returns: a step with no event to postpone.
pub type Enter(state, data, message) =
Step(state, data, message, Unpostponable)
The outcome of a machine’s initialiser: the starting state and data, the selector to receive with, the value to hand back to the parent, and any events to handle before the mailbox.
Built with initialised and refined with selecting, returning and
continuing.
pub opaque type Initialised(state, data, message, return)
Whether start links the new process to the process that started it.
pub type Linkage {
Linked
Unlinked
}
Constructors
-
LinkedThe default, and what OTP does: the starter and the machine share fate down a link, and a supervisor is that starter.
-
UnlinkedNo link. The machine 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.
What an event handler returns: a step that may postpone the event it was given.
pub type Next(state, data, message) =
Step(state, data, message, Postponable)
The marker on a step built where an event is in hand, and so may be
postponed. Next is Step carrying it.
It has no values. Its whole job is to be a name the compiler can refuse to unify, and a name a reader can look up when it appears in the error that refusal produces.
pub type Postponable
Why a machine 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 state
machine from a gleam_otp actor.
pub type StartError =
actor.StartError
The result of starting a state machine.
gleam/otp/actor’s own alias, for the same interoperability reason as
Started.
pub type StartResult(data) =
Result(actor.Started(data), actor.StartError)
What the parent is given when a machine starts: the machine’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 state
machine can be started by anything that expects a gleam_otp actor.
pub type Started(data) =
actor.Started(data)
What the machine does after a callback returns.
Built with transition, keep, stop or stop_abnormal, and refined
with postpone, the three with_*_timeout actions, cancel_timeout and
then_handle. Actions apply in the order they are written.
The fourth type parameter records where the step may be used. Callers
normally write Next or Enter rather than naming Step at all; it is
public because those two are aliases of it, and opaque because a step’s
fields are the loop’s business.
pub opaque type Step(state, data, message, postponing)
The marker on a step built where no event is in hand, and so may not be
postponed. Enter is Step carrying it.
An enter call is caused by a state change rather than by an event, so
there is nothing for postpone to re-queue. See this module’s header for
why that is a type error rather than a runtime no-op.
pub type Unpostponable
Values
pub fn call(
subject: process.Subject(message),
waiting timeout: Int,
sending make_message: fn(process.Subject(reply)) -> message,
) -> reply
Send an event and wait for the reply.
The caller crashes if no reply arrives within the timeout, rather than
carrying on against a machine that may be in an unknown state. A
re-export of process.call.
Beware of calling a machine that might postpone the call: the reply subject waits in the postponed queue until the state changes, and if it never does, the caller crashes on the timeout.
Examples
let assert Ok(body) = sm.call(subject, waiting: 1000, sending: Fetch)
pub fn cancel_timeout(
step: Step(state, data, message, postponing),
name name: String,
) -> Step(state, data, message, postponing)
Cancel the named timeout armed under name.
Cancelling a name nothing is armed under is a no-op, which is what lets a handler cancel unconditionally rather than tracking whether it armed anything. A fire already in flight when this runs is dropped by the timer book rather than handled.
Applying this to a stop value does nothing.
Examples
sm.transition(to: Ready, data:) |> sm.cancel_timeout(name: "handshake")
// -> the handshake deadline is gone, even if it just fired
pub fn continuing(
initialised: Initialised(state, data, message, return),
message: message,
) -> Initialised(state, data, message, return)
Handle message before anything in the mailbox.
This is gen_server’s handle_continue and gen_statem’s initial
next_event in one: the initialiser returns at once, so start
unblocks and the supervisor moves on, and the message is still handled
before the first external one. 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.
The initial enter callback runs before these, and anything it injects goes in front of them, by the depth-first rule in this module’s header.
Examples
sm.new_with_initialiser(1000, fn(subject) {
sm.initialised(Connecting, session)
|> sm.returning(subject)
|> sm.continuing(Attempt)
|> Ok
})
pub fn initialised(
state: state,
data: data,
) -> Initialised(state, data, message, Nil)
Take the post-initialisation state and data of the machine.
Examples
sm.new_with_initialiser(1000, fn(subject) {
sm.initialised(Connecting, new_session()) |> sm.returning(subject) |> Ok
})
pub fn keep(data: data) -> Step(state, data, message, postponing)
Stay in the current state with new data.
Nothing is cancelled and nothing is replayed: keep is not a state
change however much the data moved. A state timeout armed earlier keeps
running.
Examples
sm.keep(Session(..data, sent: data.sent + 1))
// -> same state, new data, state timeout still ticking
pub fn named(
builder: Builder(state, data, message, return),
name: process.Name(message),
) -> Builder(state, data, message, return)
Register the machine 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 machine fails to start. When this is used the machine’s default subject is the named one, which is what lets a restarted machine take over from the one it replaced.
Examples
let name = process.new_name("link")
let assert Ok(_) =
sm.new(Connecting, session) |> sm.named(name) |> sm.start
sm.send(process.named_subject(name), Request("GET /"))
pub fn new(
state: state,
data: data,
) -> Builder(state, data, message, process.Subject(message))
Describe a machine starting in state with data, and no custom
initialisation.
The machine hands the parent a subject to send messages on — a named
subject if named was used.
Examples
let assert Ok(started) =
sm.new(Connecting, new_session())
|> sm.on_event(handle)
|> sm.start
sm.send(started.data, Request("GET /"))
pub fn new_with_initialiser(
timeout: Int,
initialise: fn(process.Subject(message)) -> Result(
Initialised(state, data, message, return),
String,
),
) -> Builder(state, data, message, return)
Describe a machine with initialisation that runs in the new process
before start returns.
timeout is how many milliseconds the initialiser has; overrunning it
kills the machine and fails the start with InitTimeout. The machine’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, or an enter callback on the initial state, runs it after
the acknowledgement and still before the first external message.
Examples
sm.new_with_initialiser(1000, fn(subject) {
use socket <- result.try(open())
sm.initialised(Connecting, Session(socket:))
|> sm.returning(subject)
|> Ok
})
|> sm.on_event(handle)
|> sm.start
pub fn on_enter(
builder: Builder(state, data, message, return),
handler: fn(state, state, data) -> Step(
state,
data,
message,
Unpostponable,
),
) -> Builder(state, data, message, return)
Set the enter callback, run on every real state change and once for the initial state.
The arguments are the state left, the state entered, and the data as the transition left it. On the initial call the two states are the same value, which cannot happen otherwise: a same-state transition is not a state change and runs no enter call at all.
The callback returns an Enter, which is a Next without postpone —
there is no event in hand to re-queue. It may transition again, which
runs another enter call, and it may stop the machine.
Examples
// Every path into Connecting starts an attempt; entering Backoff arms
// the retry that will leave it.
sm.new(Connecting, session)
|> sm.on_enter(fn(_from, to, data) {
case to {
Connecting -> sm.keep(data) |> sm.then_handle(Attempt)
Backoff ->
sm.keep(data) |> sm.with_state_timeout(after: 1000, sending: Attempt)
Ready -> sm.keep(data)
}
})
pub fn on_event(
builder: Builder(state, data, message, return),
handler: fn(state, data, message) -> Step(
state,
data,
message,
Postponable,
),
) -> Builder(state, data, message, return)
Set the event handler.
It is called with the current state, the current data and one event, and
returns a Next. Writing it as case state, message is the point of the
module: add a state variant and the compiler names every pair that has
not been thought about.
Examples
sm.new(Connecting, session)
|> sm.on_event(fn(state, data, message) {
case state, message {
Connecting, Request(..) -> sm.keep(data) |> sm.postpone
Connecting, Established -> sm.transition(to: Ready, data:)
Ready, Request(body:) -> sm.keep(send(data, body))
Ready, Established -> sm.keep(data)
}
})
pub fn postpone(
next: Step(state, data, message, Postponable),
) -> Step(state, data, message, Postponable)
Re-queue the event being handled, to be redelivered on the next change of state.
This is the action that deletes hand-rolled pending queues. Postponed events are replayed ahead of the mailbox, in the order they originally arrived, exactly once each; an event may be postponed again in the new state, and takes its place at the back of the queue as it does. Postponing on a step that is itself a state change is meaningful and useful: the event is redelivered immediately, in the state being moved to.
The type refuses this in an enter callback, which has no event in hand.
Applying it to a stop value does nothing: a machine that is stopping
redelivers nothing.
Nothing bounds the postponed queue but the machine’s own logic. A state that postpones an event it can never leave will hold it forever, which is the one way this action is worse than the pending list it replaces — there the leak is at least visible.
Examples
// A request that arrives before the link is up waits in the machine.
Connecting, Request(..) -> sm.keep(data) |> sm.postpone
pub fn returning(
initialised: Initialised(state, data, message, old_return),
return: return,
) -> Initialised(state, data, message, return)
Set the value handed back to the parent when the machine has started.
Commonly the subject the machine receives on, so the parent can send to it.
Examples
sm.new_with_initialiser(1000, fn(subject) {
sm.initialised(Connecting, session) |> sm.returning(subject) |> Ok
})
pub fn selecting(
initialised: Initialised(state, data, message, return),
selector: process.Selector(message),
) -> Initialised(state, data, message, return)
Give the machine a selector to receive messages with.
This replaces the default selector, which selects only the machine’s own subject, so a custom selector must select that subject itself if the machine is still to receive on it. A message that arrives and is not selected for is discarded with a warning.
The message type is fixed rather than changed by this call, for the same
reason as in weft/actor: the injected queue mentions the message type
too, so a type-changing selecting would have to silently discard
anything continuing had already added.
Examples
sm.new_with_initialiser(1000, fn(subject) {
let selector =
process.new_selector()
|> process.select(subject)
|> process.select_map(peer_events, FromPeer)
sm.initialised(Connecting, session)
|> sm.selecting(selector)
|> sm.returning(subject)
|> Ok
})
pub fn send(
subject: process.Subject(message),
message: message,
) -> Nil
Send an event to a state machine.
A re-export of process.send, for convenience.
Examples
sm.send(started.data, Request("GET /"))
pub fn start(
builder: Builder(state, data, message, return),
) -> Result(actor.Started(return), actor.StartError)
Start a state machine 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. The
initial enter callback and any messages added with continuing are
handled after this returns and before any message sent afterwards.
Examples
let assert Ok(started) =
sm.new(Connecting, session) |> sm.on_event(handle) |> sm.start
// -> `started.data` is a subject; `started.pid` is the machine
pub fn stop() -> Step(state, data, message, postponing)
Stop and shut down, handling no further events.
The exit reason is Normal, so a Permanent child restarts and a
Transient one does not.
Examples
case message {
Shutdown -> sm.stop()
Request(body:) -> sm.keep(send(data, body))
}
pub fn stop_abnormal(
reason: String,
) -> Step(state, data, message, postponing)
Stop and shut down abnormally, propagating the reason to linked processes.
Examples
sm.stop_abnormal("the peer sent a frame we cannot parse")
// -> linked processes are told; a Permanent child is restarted
pub fn supervised(
builder: Builder(state, data, message, return),
) -> supervision.ChildSpecification(return)
Describe this machine as a supervisor’s child.
Returns gleam/otp/supervision’s own ChildSpecification, so a weft
state machine is added to a gleam_otp supervisor exactly as an upstream
actor 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(sm.supervised(link_builder))
|> supervisor.start
pub fn then_handle(
step: Step(state, data, message, postponing),
message: message,
) -> Step(state, data, message, postponing)
Handle message next, ahead of the mailbox.
This is gen_statem’s next_event: the supported way for a machine to
drive itself, and the reason an enter callback can start the work its
state exists to do. The injected message goes through the ordinary event
handler in the ordinary way, and — like any event — cancels the event
timeout.
Ordering is the contract in this module’s header. Messages injected by one callback run in the order they were written, and the whole block runs in front of anything queued earlier, depth-first. A callback that injects on every call never lets the queue drain and never reads its mailbox again.
Applying this to a stop value does nothing: a machine that is stopping
handles no more events.
Examples
// Every path into Connecting starts the attempt, including the ones
// added later.
Connecting -> sm.keep(data) |> sm.then_handle(Attempt)
// `Flush` is handled first, then `Compact`, then anything else pending.
sm.keep(data) |> sm.then_handle(Flush) |> sm.then_handle(Compact)
pub fn transition(
to state: state,
data data: data,
) -> Step(state, data, message, postponing)
Move to state, carrying data.
Moving to a state that compares unequal to the current one is a state change: it cancels the state timeout, replays postponed events ahead of the mailbox, and runs the enter callback. Moving to the state the machine is already in does none of those things, matching gen_statem — see this module’s header.
Examples
sm.transition(to: Ready, data: session)
// -> a state change, if the machine was not already Ready
// The retry deadline is armed by the same step that enters Backoff, and
// survives it: the state timeout is cancelled before the step's own
// actions run.
sm.transition(to: Backoff, data:)
|> sm.with_state_timeout(after: 1000, sending: Attempt)
pub fn trapping_exits(
builder: Builder(state, data, message, return),
trap: Bool,
) -> Builder(state, data, message, return)
Choose whether the machine traps exits.
A trapping machine receives an exit signal from a linked process as a
message instead of dying of it. The policy the loop then applies mirrors
what would have happened without trapping: an exit from the parent shuts
the machine 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.
The one thing trapping buys here is promptness rather than cleanup: a suspended machine still watches for exits, so a supervisor terminating one does not have to wait out the whole shutdown timeout and then kill it.
Trapping is off by default, and turning it on changes what kills the
machine: process.kill still does, process.send_exit no longer does.
Examples
sm.new(Connecting, session) |> sm.trapping_exits(True)
pub fn unlinked(
builder: Builder(state, data, message, return),
) -> Builder(state, data, message, return)
Start the machine without linking it to its starter.
Consumers that need this otherwise pay for it with a throwaway
process that starts the machine and exits, which leaves the machine
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 |> sm.unlinked |> sm.start
pub fn with_event_timeout(
step: Step(state, data, message, postponing),
after ms: Int,
sending message: message,
) -> Step(state, data, message, postponing)
Send the machine message after ms milliseconds with no events at all.
This one measures quiet, so the next event of any kind cancels it —
including a message injected with then_handle, a replayed postponed
event, and another timeout firing. A handler that wants the deadline to
continue arms it again, which is gen_statem’s rule.
Arming replaces any event timeout already armed. Applying this to a
stop value does nothing.
Examples
// Give up on a half-open connection that has gone quiet.
sm.keep(data) |> sm.with_event_timeout(after: 30_000, sending: PeerSilent)
pub fn with_named_timeout(
step: Step(state, data, message, postponing),
name name: String,
after ms: Int,
sending message: message,
) -> Step(state, data, message, postponing)
Send the machine message after ms milliseconds, whatever else
happens.
A named timeout survives state changes and events alike; only firing or
cancel_timeout under the same name ends it. It is the timeout for a
deadline that belongs to a piece of work rather than to a state.
Arming replaces any timeout already armed under the same name. Applying
this to a stop value does nothing.
Examples
sm.keep(data)
|> sm.with_named_timeout(name: "handshake", after: 5000, sending: TooSlow)
// -> TooSlow in five seconds, in whatever state the machine has reached
pub fn with_selector(
step: Step(state, data, message, postponing),
selector: process.Selector(message),
) -> Step(state, data, message, postponing)
Replace the selector the machine receives with going forward.
This replaces the selector given at initialisation or by an earlier
step rather than adding to it, so a selector that no longer selects the
machine’s own subject stops receiving on it. It is the way a machine
widens its mailbox to a channel that did not exist when it started — a
subject the handler itself just created, a monitor it just installed —
which otherwise forces the whole first phase into the initialiser.
Applying this to stop does nothing: there is no “going forward”.
Examples
let inner = process.new_subject()
sm.transition(to: Forwarding, data: Request(..data, inner:))
|> sm.with_selector(
process.new_selector()
|> process.select(control)
|> process.select_map(inner, InnerEvent),
)
pub fn with_state_timeout(
step: Step(state, data, message, postponing),
after ms: Int,
sending message: message,
) -> Step(state, data, message, postponing)
Send the machine message after ms milliseconds, unless it leaves this
state first.
The deadline belongs to the state, which is what makes it safe: a fire
that raced the transition out of the state is recognised as stale by the
timer book and dropped, so a retry can never be handled in a state that
did not ask for it. A transition to the same state does not cancel it,
and neither does keep.
Arming replaces any state timeout already armed. Applying this to a
stop value does nothing.
Examples
sm.transition(to: Backoff, data:)
|> sm.with_state_timeout(after: 1000, sending: Attempt)
// -> Attempt in a second, unless Backoff is left first