#moonstate

#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"])
}

#Install

[dependencies]
"moonbitstack/moonstate" = "0.1.0"

import { "moonbitstack/moonstate" }

#The state and the update

#Supersteps

• an edge a -> b queues b after a;
• a join ["a", "b"] -> c is a gate: c runs only when both wrote in the same superstep;
• a branch is called with the state as merged at the end of the superstep and returns the nodes to queue;
• Goto inside a node replaces the edges that node declared;
• Send queues a node once per input, each with its own state.

#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)
}

#Checkpoints

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"])
}

• put_writes records a task's result as soon as 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(thread) is the thread as it stands; history(thread) is every checkpoint it went through, oldest first, each one naming the checkpoint it was written on top of.
• update(thread, patch, as_node=...) writes into the latest state as if a node had written it, and queues what would have run after that node. A run afterwards continues from there, which is how a finished thread is taken back to an earlier decision.

#Checkpoints in a database

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)
}

• One connection, one saver. The saver serialises its own statements, but a driver's connection carries one exchange at a time: a saver and a store that share a connection take turns, and two savers on one connection do not.
• The dialect. A driver that numbers its parameters reports Dialect::Postgres and the saver asks it. A driver that does not report its dialect builds the statements for SQLite's ? and is refused by the server at the first statement; pass dialect=@moondb.Dialect::Postgres to @sql.saver in that case.
• A driver that blocks. SQLite's driver calls into C and implements the synchronous Driver; @sql.blocking(connection) is the bridge, and it is a bridge for any blocking driver, not a special case for SQLite.

#Interrupts

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)
}

#Where the tasks run

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)
}

#Traces

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"])
}

#Functional form

let fetch = @moonstate.task("fetch", async fn(url : String) -> String { ... })
let score = @moonstate.task("score", async fn(text : String) -> Int { ... })

let workflow = @moonstate.entrypoint("rank", async fn(url : String, ctx : @moonstate.Context) -> Int {
  let text = fetch.call(ctx, url)
  score.call(ctx, text)
}, saver=saver)

#Around a run

#Documentation

#API

#Conformance

• A write conflict. Two tasks writing the same field in one superstep is an error there and a fold in task order here. Merging is a function of (S, U) in this library, so a conflict is always resolvable and always deterministic.
• A join across supersteps. A node fed by two others runs once per trigger there; here join holds it until both upstream nodes ran in the same superstep, which is what makes a diamond run its join once.