feat(yaml): finite-state-machine entities via an fsms: block

crates/instrumentation-build/example/build.rs

@@ -23,6 +23,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let opts = Options {
         event_derives: &["Debug"],
         record_derives: &["Debug"],
+        // To just print the events in this example, we'll be using the callback
+        // exporter. This exporter takes a type-erased event, so in order to
+        // simplify downcasting back to a statically-typed event, this features
+        // enables the generation of the "AnyEvent" helper type (see main.rs).
+        // This is typically left false when using "real" exporters.
+        any_event: true,
         ..Default::default()
     };
     let GenerateInfo { path, warnings } = generate(&parsed.schema, &opts)?;

crates/instrumentation-build/example/model.yaml

@@ -43,3 +43,20 @@ entities:
         once:
           upstream: { ref: Server, data: Route }
       closed: once
+
+fsms:
+  Query:
+    doc: A query on the server, modeled as a finite-state machine.
+    states:
+      submitted:
+        initial: true
+        attributes:
+          text: string
+          connection: { scope-ref: Connection }
+        to: [running]
+      running:
+        attributes: { rows: u64 }
+        to: [running, ready]
+      ready:
+        attributes: { ok: bool }
+        to: [exit]

crates/instrumentation-build/example/src/main.rs

@@ -3,38 +3,40 @@
 
 use quent_instrumentation::{EventCallback, ExporterOptions};
 
-use crate::demo::{ConnectionHandle, ConnectionObserver, DemoContext, Event, Uuid};
+use crate::demo::{ConnectionHandle, ConnectionObserver, DemoContext, Uuid};
 
 #[allow(unused)]
 mod demo {
     include!(concat!(env!("OUT_DIR"), "/demo.rs"));
 }
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // The context owns the exporter and exposes one observer per entity type.
     let context: DemoContext = demo::DemoContext::try_new(Some(debug_printing_exporter()))?;
 
-    // Boot a server; its instance id is the target the connection references.
+    // `observer.handle()` creates a fresh entity instance to events emit for.
     let mut server = context.server_observer().handle();
     server.booted()?;
 
     let observer: ConnectionObserver = context.connection_observer();
-
-    // The handle (may) hold per-instance state that enforces once-cardinality,
-    // hence it is mut so it can update its state after producing a once-event.
+    // Once-cardinality events take `&mut self` and may fire only once, tracked
+    // by the handle, hence it is mut:
     let mut conn: ConnectionHandle = observer.handle();
 
+    // One method per entity event:
     conn.opened(
         demo::Endpoint {
             host: "localhost".to_owned(),
             port: 8080,
         },
         Uuid::nil(),
+        // A handle can deal out a reference to the entity it represents:
         server.as_entity_ref(),
     )?;
     conn.data(1234, None)?;
 
-    // `extra` is the schema's `dynamic` field: a runtime-keyed bag of typed
-    // attributes, so callers attach whatever key/values they have on hand.
+    // A `dynamic` schema field maps to `DynamicAttributes`, which are
+    // dynamically-typed key-value pairs:
     let mut extra = demo::DynamicAttributes::new();
     extra.add_string("peer_agent", "curl/8.4");
     extra.add_u64("chunk_index", 3);
@@ -47,12 +49,22 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         }),
     )?;
 
-    // `ref` field carrying data: the server reference plus details of the edge.
+    // `as_entity_ref_with` produces an entity ref that also carries data:
     conn.routed(server.as_entity_ref_with(demo::Route { hops: 3 }))?;
 
+    // An FSM entity's events are transitions into its states.
+    // Their cardinality is derived from the topology at build time.
+    //
+    // FSMs will get typestate pattern handles in the future, also see
+    // https://github.com/rapidsai/quent/issues/416
+    let mut query = context.query_observer().handle();
+    query.submitted("select 1".to_owned(), conn.as_entity_ref())?;
+    query.running(10)?;
+    query.ready(true)?;
+
     conn.closed()?;
 
-    // Emitting a once-event a second time fails.
+    // A once-event returns an error if emitted again.
     assert!(conn.closed_emitted());
     assert!(conn.closed().is_err());
 
@@ -62,15 +74,8 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 /// Return an exporter that debug-prints each emitted event's payload.
 fn debug_printing_exporter() -> ExporterOptions {
     ExporterOptions::Callback(EventCallback::new(|recorded| {
-        if let Some(event) = recorded
-            .event
-            .downcast_ref::<Event<demo::ConnectionEvent>>()
-        {
-            println!("[{} @ {}] {:?}", event.id, event.timestamp, event.data);
-        } else if let Some(event) = recorded.event.downcast_ref::<Event<demo::ServerEvent>>() {
-            println!("[{} @ {}] {:?}", event.id, event.timestamp, event.data);
-        } else {
-            unreachable!()
+        if let Some(event) = demo::AnyEvent::from_any(recorded.event.as_ref()) {
+            println!("{event:?}");
         }
     }))
 }

crates/yaml/Cargo.toml

@@ -7,6 +7,7 @@ publish.workspace = true
 [dependencies]
 indexmap = { workspace = true, features = ["serde"] }
 quent-constraints = { path = "../constraints" }
+quent-fsm = { path = "../fsm" }
 quent-ref-target = { path = "../ref-target" }
 quent-ref-tree = { path = "../ref-tree" }
 quent-schema = { path = "../schema" }

crates/yaml/src/ast.rs

@@ -38,6 +38,37 @@ pub(crate) struct Model {
     pub(crate) records: IndexMap<String, Record>,
     #[serde(default)]
     pub(crate) entities: IndexMap<String, Entity>,
+    #[serde(default)]
+    pub(crate) fsms: IndexMap<String, FsmSpec>,
+}
+
+/// An FSM entity: annotations plus its states.
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct FsmSpec {
+    #[serde(default)]
+    pub(crate) doc: Option<String>,
+    #[serde(default)]
+    pub(crate) constraints: AnnotationMap,
+    #[serde(default)]
+    pub(crate) metadata: AnnotationMap,
+    pub(crate) states: IndexMap<String, StateSpec>,
+}
+
+/// One state of an FSM.
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct StateSpec {
+    #[serde(default)]
+    pub(crate) initial: bool,
+    #[serde(default)]
+    pub(crate) attributes: IndexMap<String, Field>,
+    // States the FSM can transition to.
+    //
+    // "exit" is a reserved special name to mark a state as final before
+    // dissapearing from existence through the exit transition.
+    #[serde(default)]
+    pub(crate) to: Vec<String>,
 }
 
 /// A record: named fields plus annotations.

crates/yaml/src/lib.rs

@@ -10,8 +10,9 @@
 use std::path::Path;
 
 use quent_constraints::validate;
-use quent_ref_target::{RefTargetConstraint, RefTargetError};
-use quent_ref_tree::{RefTreeConstraint, RefTreeError};
+use quent_fsm::FsmConstraint;
+use quent_ref_target::RefTargetConstraint;
+use quent_ref_tree::RefTreeConstraint;
 use quent_schema::Schema;
 use serde_saphyr::{MessageFormatter, UserMessageFormatter};
 
@@ -66,7 +67,7 @@ pub fn parse_from_str(src: impl AsRef<str>, source: Option<&str>) -> Result<Pars
         return Err(Error::Invalid(sink));
     }
 
-    let report = validate::<(RefTargetConstraint, RefTreeConstraint)>(&schema);
+    let report = validate::<(RefTargetConstraint, RefTreeConstraint, FsmConstraint)>(&schema);
     if let Err(e) = report.base_constraints {
         for record in e.recursive_records {
             sink.error(
@@ -82,12 +83,15 @@ pub fn parse_from_str(src: impl AsRef<str>, source: Option<&str>) -> Result<Pars
             sink.error("", format!("unresolved reference: {reference}"), None);
         }
     }
-    let (ref_target, ref_tree) = report.results;
+    let (ref_target, ref_tree, fsm) = report.results;
     if let Err(e) = ref_target {
-        ref_target_diagnostics(e, &mut sink);
+        sink.error("", e.to_string(), None);
     }
     if let Err(e) = ref_tree {
-        ref_tree_diagnostics(e, &mut sink);
+        sink.error("", e.to_string(), None);
+    }
+    if let Err(e) = fsm {
+        sink.error("", e.to_string(), None);
     }
     if sink.has_errors() {
         return Err(Error::Invalid(sink));
@@ -116,27 +120,3 @@ pub fn parse_from_file(path: impl AsRef<Path>) -> Result<Parsed, Error> {
     let src = std::fs::read_to_string(path)?;
     parse_from_str(&src, Some(&path.display().to_string()))
 }
-
-/// Report one diagnostic per ref-target violation, flattening `Multiple`.
-fn ref_target_diagnostics(error: RefTargetError, sink: &mut Diagnostics) {
-    match error {
-        RefTargetError::Multiple(errors) => {
-            errors
-                .into_iter()
-                .for_each(|error| ref_target_diagnostics(error, sink));
-        }
-        error => sink.error("", error.to_string(), None),
-    }
-}
-
-/// Report one diagnostic per ref-tree violation, flattening `Multiple`.
-fn ref_tree_diagnostics(error: RefTreeError, sink: &mut Diagnostics) {
-    match error {
-        RefTreeError::Multiple(errors) => {
-            errors
-                .into_iter()
-                .for_each(|error| ref_tree_diagnostics(error, sink));
-        }
-        error => sink.error("", error.to_string(), None),
-    }
-}