A run, step by step

01Install

The engine depends on moonbitlang/async and nothing else. The two subpackages are separate imports, so a reader who wants the graph never downloads a database driver or a tracing stack.

moon add moonbitstack/moonstate

import { "moonbitstack/moonstate" }

02State and fold

Two types: the state S, and the update U a node writes. One function, merge : (S, U) -> S. Both go through JSON, because a checkpoint stores the state, and an executor that ships a task elsewhere sends it as JSON.

A node returns steps, not a new state. Write folds an update, Goto leaves the declared edges behind, Send queues a node with its own input, Interrupt stops the run and waits.

#derive(@json.ToJson, @json.FromJson)
struct Counter {
  n : Int32
  log : Array[String]
}

#derive(@json.ToJson, @json.FromJson)
struct Patch {
  n : Int32?
  log : Array[String]?
}

fn merge(state : Counter, patch : Patch) -> Counter {
  { n: patch.n.unwrap_or(state.n), log: state.log + patch.log.unwrap_or([]) }
}

async fn hello() -> Counter raise {
  let graph = @moonstate.Builder::new(merge~, initial={ n: 0, log: [] })
    .entry("seed")
    .node("seed", (_state, _run) => {
      [@moonstate.Write({ n: Some(1), log: Some(["seed"]) })]
    })
    .node("double", (state, _run) => {
      [@moonstate.Write({ n: Some(state.n * 2), log: None })]
    })
    .edge("seed", "double")
    .compile("hello")
  graph.run({ n: None, log: None })
}

async test "the first graph" {
  let out = hello() catch { error => abort((error : Error).to_string()) }
  assert_eq(out.n, 2)
  assert_eq(out.log, ["seed"])
}

out n = 2 · log = ["seed"]

compile checks the structure before a run can happen: an edge to nowhere, a node nothing reaches, a graph with no end, a node entered at run time that nobody declared with routed. Those arrive as Refused, not as a failure halfway through a run.

03Checkpoints

Compile with a Saver and every superstep is written down before it runs. The next run on the same thread opens the last checkpoint instead of starting over, which is what makes a crash survivable and a finished thread restartable.

async fn durable() -> Array[String] raise {
  let saver = @moonstate.MemorySaver::new()
  let graph = @moonstate.Builder::new(merge~, initial={ n: 0, log: [] })
    .entry("seed")
    .node("seed", (_state, _run) => {
      [@moonstate.Write({ n: Some(1), log: Some(["seed"]) })]
    })
    .node("double", (state, _run) => {
      [@moonstate.Write({ n: Some(state.n * 2), log: None })]
    })
    .edge("seed", "double")
    .compile("durable", saver=saver)
  let _ = graph.run({ n: None, log: None }, thread="job")
  let second = graph.run({ n: None, log: None }, thread="job")
  let history = graph.history("job")
  [second.n.to_string(), history.length().to_string(), history[2].id]
}

async test "a finished thread starts a new turn" {
  let out = durable() catch { error => abort((error : Error).to_string()) }
  assert_eq(out, ["2", "6", "cp-000002"])
}

out ["2", "6", "cp-000002"] — the value, six checkpoints, the third one by id

put_writes records a task's result the moment it finishes, so a task that completed is not run twice after a crash even when the rest of its superstep did not survive. state is the thread now; history is every checkpoint it went through, each naming the one it was written on top of.

A saver is an interface, not a table: MemorySaver is for tests, moonstate/sql is for a database, and src/sql/saver.mbt is a hundred lines to read if you want your own.

04Questions

A question is not a callback and not a thread that blocks. The node returns Interrupt(payload), the run ends with GraphError::Interrupted, and the payload is in the checkpoint. The answer arrives later, from anywhere, and the node that asked runs again and reads it.

async fn approve() -> Int32 raise {
  let graph = @moonstate.Builder::new(merge~, initial={ n: 0, log: [] })
    .entry("approve")
    .node("approve", (_state, run) => match run.answer() {
      Some(answer) => {
        let decided : Int32 = @json.from_json(answer)
        [@moonstate.Write({ n: Some(decided), log: None })]
      }
      None => [@moonstate.Interrupt({ "ask": "approve?" })]
    })
    .compile("approval", saver=@moonstate.MemorySaver::new())
  let stopped = try {
    let _ = graph.run({ n: None, log: None }, thread="review")
    "finished"
  } catch {
    @moonstate.GraphError::Interrupted(value~, ..) => value.stringify()
    _ => abort("the run failed for another reason")
  }
  assert_eq(stopped, "{\"ask\":\"approve?\"}")
  let out = graph.resume_with("review", Json(7))
  out.n
}

async test "a question stops the run and the answer continues it" {
  let out = approve() catch { error => abort((error : Error).to_string()) }
  assert_eq(out, 7)
}

stopped with {"ask":"approve?"} then answered n = 7

05Supersteps and the fan-out

A superstep runs the nodes that are queued, merges what they wrote in task order, then decides what runs next. Send queues one node once per input; a join holds a node until every upstream node has run in the same superstep.

#derive(@json.ToJson, @json.FromJson)
struct Report {
  parts : Array[Array[Int32]]
  total : Int32
}

#derive(@json.ToJson, @json.FromJson)
struct Part {
  value : Int32
}

fn report_merge(state : Report, part : Part) -> Report {
  { parts: state.parts + [[part.value]], total: state.total + part.value }
}

async fn fan_out() -> Report raise {
  let graph = @moonstate.Builder::new(merge=report_merge, initial={ parts: [], total: 0 })
    .entry("fan")
    .node("fan", (state, _run) => {
      let steps : Array[@moonstate.Step[Report, Part]] = []
      for index = 0; index < 3; index = index + 1 {
        steps.push(@moonstate.Send("shard", { parts: state.parts, total: index }))
      }
      steps
    })
    .node("shard", (state, _run) => {
      let steps : Array[@moonstate.Step[Report, Part]] = []
      steps.push(@moonstate.Write({ value: state.total * 10 }))
      steps
    })
    .routed("shard")
    .compile("fan")
  graph.run({ value: 0 })
}

async test "one node, three inputs" {
  let out = fan_out() catch { error => abort((error : Error).to_string()) }
  assert_eq(out.total, 30)
}

out total = 30, four shards folded in the order they were queued

The result does not depend on which task finished first: the engine folds the writes in the order the tasks were queued. That is the whole reason merging is a function rather than a set of field annotations.

06The executor

An Executor decides where a superstep's tasks run. The default puts them all in this process at the same time; an executor that ships them elsewhere finds the node by name on the far side and returns the steps as JSON. A policy of your own is one method.

struct OneAtATime {}

pub impl @moonstate.Executor for OneAtATime with fn each(_self, works) {
  let done : Array[@moonstate.Done] = []
  for work in works {
    let outcome = try {
      Ok((work.run)())
    } catch {
      error => Err(error)
    }
    done.push(match outcome {
      Ok(values) => @moonstate.Done::{ values: Some(values), error: None }
      Err(error) =>
        @moonstate.Done::{ values: None, error: Some((error : Error).to_string()) }
    })
  }
  done
}

// The trait's method is not promoted to a plain method of the type: call it
// through the trait, which is what the engine does.
pub extend OneAtATime with @moonstate.Executor::{each}

async fn one_at_a_time() -> Int32 raise {
  let graph = @moonstate.Builder::new(merge~, initial={ n: 0, log: [] })
    .entry("seed")
    .node("seed", (_state, _run) => [@moonstate.Write({ n: Some(5), log: None })])
    .compile("serial", executor=OneAtATime::{})
  let out = graph.run({ n: None, log: None })
  out.n
}

async test "an executor of one's own" {
  let out = one_at_a_time() catch { error => abort((error : Error).to_string()) }
  assert_eq(out, 5)
}

out n = 5, one task at a time

07A database

moonbitstack/moonstate/sql puts the saver and the store on any backend the database seam speaks. SQLite, PostgreSQL and MySQL run the same code: the placeholders, the upsert spelling and the indexes are written once against the dialect the driver reports.

async fn [T : @moondb.AsyncDriver] on_a_database(
  connection : T,
) -> @moonstate.Graph[Counter, Patch] raise {
  let saver = @sql.saver(connection)
  // Creates the tables and their indexes when they are not there; safe to run on
  // every start.
  saver.setup()
  @moonstate.Builder::new(merge~, initial={ n: 0, log: [] })
    .entry("seed")
    .node("seed", (_state, _run) => {
      [@moonstate.Write({ n: Some(1), log: Some(["seed"]) })]
    })
    .compile("on-a-database", saver=saver)
}

// once, at startup
@sql.saver(connection).setup()
@sql.store(connection).setup()

One connection carries one saver and one store: a driver's connection carries one exchange at a time. A blocking driver — SQLite's calls into C — goes in through @blocking.Blocking::new(connection). A driver that does not report its dialect is told which one to use: @sql.saver(connection, dialect=@moondb.Dialect::Postgres).

08Traces

moonbitstack/moonstate/otel wraps an executor so every superstep is a span and every node a child of it, named with the GenAI semantic conventions. A trace follows a thread from superstep to superstep, so a run is one trace however long it takes, and a node span sits under the superstep that ran it.

fn clock() -> Int64 {
  @env.now().reinterpret_as_int64() * 1_000_000L
}

fn random_bytes(width : Int32) -> Bytes {
  match @env.rand(width) {
    Some(bytes) => bytes
    None => abort("the platform has no entropy source")
  }
}

async fn traced_run() -> Array[String] raise {
  let spans = @trace.Recorder::new()
  let tracer = @trace.Tracer::new(
    rand=random_bytes,
    sink=spans,
    sampler=@trace.parent_based(root=@trace.always),
  )
  let executor = @otel.traced(@moonstate.inline, tracer~, now=clock)
  let graph = @moonstate.Builder::new(merge~, initial={ n: 0, log: [] })
    .entry("seed")
    .node("seed", (_state, _run) => [@moonstate.Write({ n: Some(2), log: None })])
    .compile("traced", executor=executor)
  let _ = graph.run({ n: None, log: None })
  spans.spans().map(span => span.name)
}

async test "a run is a tree of spans" {
  let names = traced_run() catch { error => abort((error : Error).to_string()) }
  assert_eq(names, ["seed", "superstep 0"])
}

span names ["seed", "superstep 0"]

now is required, the way moontel's own layer requires it: a library that reads a clock cannot be replayed. The live check in CI sends these spans to weaver registry live-check, and a name this library invented would fail the job.