Bendik Aagaard LynghaugandClaude Opus 5
f18d1a5181
Five things a building authority needed and portal could not say,
found by building one and replaying its cases through the engine.
The one that mattered: a submitter could only ever act on a record
nobody had touched. `access.rs` pinned a self_transition's from-state
to `open`, though the runtime had always passed the real state and
asked. So the applicant written to for a missing drawing could not
send it, and nobody could withdraw once anything had happened.
`self_transition` now takes `from: [state, ...]` and compiles one rule
per state; omitted it still means the initial state alone, which is
what every unsubscribe link already meant.
Then the four beside it. A record can belong to another record
(`type: record`, `of:`), with `where: { field, equals }` on a listing
as the other half, so an application's page lists the objections that
name it. A public view can be narrowed to the states worth showing,
so a register that is public by law is public about what was decided
and not about what was withdrawn. A state can grant a group, so an
approval that means membership needs no second step. And a state can
be moved by a clock (`after: { days, to }`, src/deadlines.rs) - held
to the same declared edge as a desk button, firing the same mail,
publishing the same event.
Also: the README is a README again, and everything that was crowding
it is in docs/ - architecture, content reference, state machines,
access, operations.
95 tests, up from 82.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
606 lines
21 KiB
Rust
606 lines
21 KiB
Rust
//! Generic pure-replay + CAS-write machinery for event-sourced
|
|
//! aggregates - the state graph itself (what states exist, what
|
|
//! event-type string represents each, what transitions are legal) is
|
|
//! loaded at runtime from `questions/aggregates.yaml`, not compiled
|
|
//! Rust. Content declares the graph; this module only knows how to
|
|
//! replay/append against whatever graph it's handed
|
|
//! (`content::load_aggregates_from_gitea`, hot-swapped in
|
|
//! `AppState.aggregates` the same way `AppState.questions` is).
|
|
//!
|
|
//! `AggregateSchema.bucket` doubles as both the KV bucket name
|
|
//! (`answers.rs`'s read-model projection) and the JetStream
|
|
//! event-subject segment (`events::store`'s `events.<bucket>.<id>`) -
|
|
//! one name, not two, since content only ever had one name for this to
|
|
//! begin with.
|
|
#![cfg(feature = "ssr")]
|
|
|
|
use std::collections::HashMap;
|
|
|
|
use crate::events::store::{append_event, load_events, Event};
|
|
|
|
/// One aggregate type's full state graph, already validated (see
|
|
/// `parse_aggregates_yaml`) - `initial` is guaranteed to be a key of
|
|
/// `event_for_state`/`state_for_event`, every `transitions` key and
|
|
/// target is guaranteed to be a declared state, and every state's
|
|
/// event-type string is guaranteed unique within this schema.
|
|
#[derive(Clone, Debug)]
|
|
pub struct AggregateSchema {
|
|
pub bucket: String,
|
|
pub initial: String,
|
|
pub event_for_state: HashMap<String, String>,
|
|
pub state_for_event: HashMap<String, String>,
|
|
pub transitions: HashMap<String, Vec<String>>,
|
|
/// Names what consumes this bucket's answers when no page in the
|
|
/// content repo reads it (e.g. "n8n newsletter compose").
|
|
/// Free-text - it exists so `validate_questions`' attended-bucket
|
|
/// rule has an explicit, auditable opt-out instead of a silent
|
|
/// one, and so the next reader knows where the answers go.
|
|
pub attended_by: Option<String>,
|
|
/// The mail a record's arrival in a state sends (see `mail.rs`):
|
|
/// declared on the state in aggregates.yaml, next to the event
|
|
/// that enters it, so content decides when a person hears
|
|
/// something and the mailer never has to know a site.
|
|
pub mail_for_state: HashMap<String, MailSpec>,
|
|
/// What a state grants the person whose record entered it: a group
|
|
/// they become a member of. A firm approved to file is a member of
|
|
/// the approved firms from that moment, and the decision that made
|
|
/// it so is the same one the desk already pressed - nobody should
|
|
/// have to remember to invite them afterwards.
|
|
pub grants_for_state: HashMap<String, String>,
|
|
/// The move a clock makes rather than a person. A hearing that gets
|
|
/// no answer in six weeks has to go on without one, and nothing in
|
|
/// a building is going to press that button.
|
|
pub after_for_state: HashMap<String, AfterSpec>,
|
|
}
|
|
|
|
/// A mail to send when a record enters a state. `to` is `submitter`
|
|
/// (the record's `email` field, which the lint then requires the
|
|
/// forms to collect) or one fixed address. Subject and body are
|
|
/// templates: `{field}` for any field of the record, plus `{site}`
|
|
/// (the instance's public URL), `{chain}` (the record id), `{email}`,
|
|
/// `{bucket}` and `{state}`. The body is markdown; the mailer sends it
|
|
/// as plain text with an HTML alternative.
|
|
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
pub struct MailSpec {
|
|
pub to: String,
|
|
pub subject: String,
|
|
pub body: String,
|
|
}
|
|
|
|
impl MailSpec {
|
|
pub fn to_submitter(&self) -> bool {
|
|
self.to == "submitter"
|
|
}
|
|
}
|
|
|
|
/// A move made by a clock: `after: { days: 42, to: no_answer }` on the
|
|
/// state a record is waiting in. The wait is measured from when the
|
|
/// record entered that state, and `to` must be an edge the same schema
|
|
/// already declares - a deadline may only do what a person could have
|
|
/// done. Days, because every deadline a public body keeps is in days
|
|
/// and a shorter unit would invite this to be used as a job queue.
|
|
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
|
|
#[serde(deny_unknown_fields)]
|
|
pub struct AfterSpec {
|
|
pub days: u32,
|
|
pub to: String,
|
|
}
|
|
|
|
impl AggregateSchema {
|
|
pub fn has_state(&self, state: &str) -> bool {
|
|
self.event_for_state.contains_key(state)
|
|
}
|
|
|
|
pub fn mail_for(&self, state: &str) -> Option<&MailSpec> {
|
|
self.mail_for_state.get(state)
|
|
}
|
|
|
|
pub fn grants_for(&self, state: &str) -> Option<&str> {
|
|
self.grants_for_state.get(state).map(String::as_str)
|
|
}
|
|
|
|
pub fn after_for(&self, state: &str) -> Option<&AfterSpec> {
|
|
self.after_for_state.get(state)
|
|
}
|
|
|
|
pub fn allowed(&self, state: &str) -> &[String] {
|
|
self.transitions
|
|
.get(state)
|
|
.map(|v| v.as_slice())
|
|
.unwrap_or(&[])
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, serde::Deserialize)]
|
|
struct RawState {
|
|
event: String,
|
|
#[serde(default)]
|
|
mail: Option<MailSpec>,
|
|
#[serde(default)]
|
|
grants: Option<String>,
|
|
#[serde(default)]
|
|
after: Option<AfterSpec>,
|
|
}
|
|
|
|
#[derive(Debug, serde::Deserialize)]
|
|
struct RawAggregateSchema {
|
|
bucket: String,
|
|
initial: String,
|
|
states: HashMap<String, RawState>,
|
|
#[serde(default)]
|
|
transitions: HashMap<String, Vec<String>>,
|
|
#[serde(default)]
|
|
attended_by: Option<String>,
|
|
}
|
|
|
|
#[derive(Debug, serde::Deserialize)]
|
|
struct AggregatesFile {
|
|
aggregates: Vec<RawAggregateSchema>,
|
|
}
|
|
|
|
/// Parses and validates `aggregates.yaml`'s full contents. A malformed
|
|
/// entry (an `initial`/`transitions` key or target that isn't a
|
|
/// declared state, a duplicate `bucket`, or two states sharing one
|
|
/// event-type string) is rejected here with a clear message - the same
|
|
/// "bad content never silently corrupts a read/write" policy
|
|
/// `content::validate_questions` already applies to pages.
|
|
pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result<HashMap<String, AggregateSchema>> {
|
|
let file: AggregatesFile =
|
|
serde_yaml::from_str(raw).map_err(|e| anyhow::anyhow!("parsing aggregates.yaml: {e}"))?;
|
|
|
|
let mut out = HashMap::new();
|
|
for raw_schema in file.aggregates {
|
|
let bucket = raw_schema.bucket.clone();
|
|
if out.contains_key(&bucket) {
|
|
anyhow::bail!("aggregates.yaml: bucket {bucket:?} declared more than once");
|
|
}
|
|
if !raw_schema.states.contains_key(&raw_schema.initial) {
|
|
anyhow::bail!(
|
|
"aggregates.yaml: bucket {bucket:?}: initial state {:?} is not declared in states",
|
|
raw_schema.initial
|
|
);
|
|
}
|
|
|
|
let mut event_for_state = HashMap::new();
|
|
let mut state_for_event = HashMap::new();
|
|
for (state, raw_state) in &raw_schema.states {
|
|
if let Some(existing) = state_for_event.insert(raw_state.event.clone(), state.clone())
|
|
{
|
|
anyhow::bail!(
|
|
"aggregates.yaml: bucket {bucket:?}: event {:?} used by both state {:?} and {:?}",
|
|
raw_state.event, existing, state
|
|
);
|
|
}
|
|
event_for_state.insert(state.clone(), raw_state.event.clone());
|
|
}
|
|
|
|
for (from, targets) in &raw_schema.transitions {
|
|
if !raw_schema.states.contains_key(from) {
|
|
anyhow::bail!(
|
|
"aggregates.yaml: bucket {bucket:?}: transitions declared from undeclared state {from:?}"
|
|
);
|
|
}
|
|
for to in targets {
|
|
if !raw_schema.states.contains_key(to) {
|
|
anyhow::bail!(
|
|
"aggregates.yaml: bucket {bucket:?}: transition {from:?} -> {to:?} targets an undeclared state"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut mail_for_state = HashMap::new();
|
|
let mut grants_for_state = HashMap::new();
|
|
let mut after_for_state = HashMap::new();
|
|
for (state, raw_state) in raw_schema.states {
|
|
if let Some(group) = raw_state.grants {
|
|
if group.trim().is_empty() {
|
|
anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: grants needs a group");
|
|
}
|
|
grants_for_state.insert(state.clone(), group);
|
|
}
|
|
if let Some(after) = raw_state.after {
|
|
if after.days == 0 {
|
|
anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: after.days must be at least one");
|
|
}
|
|
// A clock may only make a move a person could have
|
|
// made, so it is held to the same graph.
|
|
if !raw_schema.transitions.get(&state).is_some_and(|tos| tos.contains(&after.to)) {
|
|
anyhow::bail!(
|
|
"aggregates.yaml: bucket {bucket:?}: state {state:?}: after.to {:?} is not a declared transition from it",
|
|
after.to
|
|
);
|
|
}
|
|
after_for_state.insert(state.clone(), after);
|
|
}
|
|
if let Some(mail) = raw_state.mail {
|
|
if mail.subject.trim().is_empty() || mail.body.trim().is_empty() {
|
|
anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: mail needs a subject and a body");
|
|
}
|
|
if !mail.to_submitter() && !(mail.to.contains('@') && !mail.to.contains(char::is_whitespace)) {
|
|
anyhow::bail!(
|
|
"aggregates.yaml: bucket {bucket:?}: state {state:?}: mail.to must be `submitter` or one address, not {:?}",
|
|
mail.to
|
|
);
|
|
}
|
|
mail_for_state.insert(state, mail);
|
|
}
|
|
}
|
|
|
|
out.insert(
|
|
bucket.clone(),
|
|
AggregateSchema {
|
|
bucket,
|
|
initial: raw_schema.initial,
|
|
event_for_state,
|
|
state_for_event,
|
|
transitions: raw_schema.transitions,
|
|
attended_by: raw_schema.attended_by,
|
|
mail_for_state,
|
|
grants_for_state,
|
|
after_for_state,
|
|
},
|
|
);
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
/// One aggregate's current view, replayed from its event history.
|
|
#[derive(Clone, Debug)]
|
|
pub struct Aggregate {
|
|
pub id: String,
|
|
pub state: String,
|
|
pub payload: serde_json::Value,
|
|
pub sequence: u64,
|
|
}
|
|
|
|
/// Pure, no IO, never panics. An event whose type isn't declared in
|
|
/// `schema`, or that doesn't fit the transition table from wherever
|
|
/// replay has gotten to so far, is silently skipped - not an error,
|
|
/// not even the *first* recognized event has anything to be "from" yet
|
|
/// so it's always accepted. This lenience (not enforcing the
|
|
/// transition table during replay, only at write time - see
|
|
/// `transition` below) is deliberate: a corrupted or out-of-order
|
|
/// event never crashes a read, it's just ignored.
|
|
pub fn replay(schema: &AggregateSchema, id: &str, events: &[Event]) -> Option<Aggregate> {
|
|
let mut current: Option<Aggregate> = None;
|
|
for event in events {
|
|
let Some(next) = schema.state_for_event.get(&event.event_type) else {
|
|
continue;
|
|
};
|
|
let accept = match ¤t {
|
|
None => true,
|
|
Some(agg) => *next == agg.state || schema.allowed(&agg.state).iter().any(|s| s == next),
|
|
};
|
|
if !accept {
|
|
continue;
|
|
}
|
|
current = Some(Aggregate {
|
|
id: id.to_string(),
|
|
state: next.clone(),
|
|
payload: event.payload.clone(),
|
|
sequence: event.sequence,
|
|
});
|
|
}
|
|
current
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub enum TransitionError {
|
|
/// No creation event found for this id - nothing to transition.
|
|
UnknownAggregate,
|
|
DisallowedTransition { from: String, to: String },
|
|
/// A concurrent writer advanced this aggregate past the sequence
|
|
/// this call read (see `events::store::append_event`'s CAS).
|
|
Conflict,
|
|
Store(String),
|
|
}
|
|
|
|
impl std::fmt::Display for TransitionError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
Self::UnknownAggregate => write!(f, "unknown item"),
|
|
Self::DisallowedTransition { from, to } => {
|
|
write!(f, "cannot go from {from} to {to}")
|
|
}
|
|
Self::Conflict => write!(f, "item was just changed by someone else, try again"),
|
|
Self::Store(e) => write!(f, "storage error: {e}"),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn classify_publish_err(e: async_nats::jetstream::context::PublishError) -> TransitionError {
|
|
use async_nats::jetstream::context::PublishErrorKind;
|
|
match e.kind() {
|
|
PublishErrorKind::WrongLastSequence | PublishErrorKind::WrongLastMessageId => {
|
|
TransitionError::Conflict
|
|
}
|
|
_ => TransitionError::Store(e.to_string()),
|
|
}
|
|
}
|
|
|
|
/// Appends the creation event for a brand-new aggregate. `expected_seq:
|
|
/// Some(0)` (no message yet on this id's subject) doubles as a
|
|
/// create-only-once guard - a double-submit race on the same id fails
|
|
/// the same way a disallowed transition does, rather than silently
|
|
/// overwriting.
|
|
pub async fn create(
|
|
js: &async_nats::jetstream::Context,
|
|
schema: &AggregateSchema,
|
|
id: &str,
|
|
payload: serde_json::Value,
|
|
occurred_at_ms: i64,
|
|
) -> Result<Aggregate, TransitionError> {
|
|
let event_type = schema
|
|
.event_for_state
|
|
.get(&schema.initial)
|
|
.expect("schema.initial is validated to have an event mapping");
|
|
let ack = append_event(
|
|
js,
|
|
&schema.bucket,
|
|
id,
|
|
event_type,
|
|
payload.clone(),
|
|
occurred_at_ms,
|
|
Some(0),
|
|
)
|
|
.await
|
|
.map_err(classify_publish_err)?;
|
|
Ok(Aggregate {
|
|
id: id.to_string(),
|
|
state: schema.initial.clone(),
|
|
payload,
|
|
sequence: ack.sequence,
|
|
})
|
|
}
|
|
|
|
/// Reconstructs a minimal one-event history for an aggregate whose
|
|
/// event log is empty but whose KV projection still knows its current
|
|
/// state - what makes wiping the EVENTS stream a recoverable
|
|
/// operation rather than one that strands every pre-wipe record in an
|
|
/// untransitionable limbo. `expected_seq: Some(0)` guards two racing
|
|
/// reseeds the same way `create` guards a double submit.
|
|
pub async fn reseed(
|
|
js: &async_nats::jetstream::Context,
|
|
schema: &AggregateSchema,
|
|
id: &str,
|
|
state: &str,
|
|
occurred_at_ms: i64,
|
|
) -> Result<(), TransitionError> {
|
|
let event_type = schema.event_for_state.get(state).ok_or_else(|| {
|
|
TransitionError::Store(format!(
|
|
"cannot reseed {id}: state {state:?} is not declared for bucket {:?}",
|
|
schema.bucket
|
|
))
|
|
})?;
|
|
append_event(
|
|
js,
|
|
&schema.bucket,
|
|
id,
|
|
event_type,
|
|
serde_json::json!({ "reseeded_from_projection": true }),
|
|
occurred_at_ms,
|
|
Some(0),
|
|
)
|
|
.await
|
|
.map_err(classify_publish_err)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Loads and replays `id`'s current state, then - if `target` is
|
|
/// actually reachable from it - appends the corresponding event with
|
|
/// the aggregate's just-replayed sequence as the CAS guard.
|
|
pub async fn transition(
|
|
js: &async_nats::jetstream::Context,
|
|
schema: &AggregateSchema,
|
|
id: &str,
|
|
target: &str,
|
|
payload: serde_json::Value,
|
|
occurred_at_ms: i64,
|
|
) -> Result<Aggregate, TransitionError> {
|
|
let events = load_events(js, &schema.bucket, id)
|
|
.await
|
|
.map_err(|e| TransitionError::Store(e.to_string()))?;
|
|
let current = replay(schema, id, &events).ok_or(TransitionError::UnknownAggregate)?;
|
|
if !schema.allowed(¤t.state).iter().any(|s| s == target) {
|
|
return Err(TransitionError::DisallowedTransition {
|
|
from: current.state.clone(),
|
|
to: target.to_string(),
|
|
});
|
|
}
|
|
// Validated by parse_aggregates_yaml: every transitions target is a
|
|
// declared state, so this is always Some given the check above.
|
|
let event_type = schema
|
|
.event_for_state
|
|
.get(target)
|
|
.expect("transition target is validated to be a declared state");
|
|
let ack = append_event(
|
|
js,
|
|
&schema.bucket,
|
|
id,
|
|
event_type,
|
|
payload.clone(),
|
|
occurred_at_ms,
|
|
Some(current.sequence),
|
|
)
|
|
.await
|
|
.map_err(classify_publish_err)?;
|
|
Ok(Aggregate {
|
|
id: id.to_string(),
|
|
state: target.to_string(),
|
|
payload,
|
|
sequence: ack.sequence,
|
|
})
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn test_schema() -> AggregateSchema {
|
|
parse_aggregates_yaml(
|
|
r#"
|
|
aggregates:
|
|
- bucket: test
|
|
initial: open
|
|
states:
|
|
open: { event: opened }
|
|
a: { event: a }
|
|
b: { event: b }
|
|
transitions:
|
|
open: [a, b]
|
|
"#,
|
|
)
|
|
.unwrap()
|
|
.remove("test")
|
|
.unwrap()
|
|
}
|
|
|
|
#[test]
|
|
fn a_state_may_carry_the_mail_its_arrival_sends() {
|
|
let schemas = parse_aggregates_yaml(
|
|
r#"
|
|
aggregates:
|
|
- bucket: reports
|
|
initial: open
|
|
states:
|
|
open: { event: received, mail: { to: submitter, subject: "Got it, {name}", body: "We read it. {site}/decide?chain={chain}" } }
|
|
fixed: { event: fixed, mail: { to: board@example.no, subject: "Fixed: {where}", body: "Done." } }
|
|
dropped: { event: dropped }
|
|
transitions:
|
|
open: [fixed, dropped]
|
|
"#,
|
|
)
|
|
.unwrap();
|
|
let s = &schemas["reports"];
|
|
assert!(s.mail_for("open").unwrap().to_submitter());
|
|
assert_eq!(s.mail_for("fixed").unwrap().to, "board@example.no");
|
|
assert!(s.mail_for("dropped").is_none());
|
|
|
|
for bad in [
|
|
"mail: { to: submitter, subject: \"\", body: b }",
|
|
"mail: { to: \"the board\", subject: s, body: b }",
|
|
"mail: { to: submitter, subject: s, body: b, cc: x }",
|
|
] {
|
|
let yaml = format!("aggregates:\n - bucket: r\n initial: open\n states:\n open: {{ event: received, {bad} }}\n");
|
|
assert!(parse_aggregates_yaml(&yaml).is_err(), "{bad}");
|
|
}
|
|
}
|
|
|
|
fn event(seq: u64, event_type: &str) -> Event {
|
|
Event {
|
|
id: uuid::Uuid::new_v4(),
|
|
aggregate_type: "test".to_string(),
|
|
aggregate_id: "x".to_string(),
|
|
event_type: event_type.to_string(),
|
|
payload: serde_json::json!({}),
|
|
occurred_at_ms: 0,
|
|
sequence: seq,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn empty_history_is_no_aggregate() {
|
|
let schema = test_schema();
|
|
assert!(replay(&schema, "x", &[]).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn valid_path_replays() {
|
|
let schema = test_schema();
|
|
let events = vec![event(1, "opened"), event(2, "a")];
|
|
let agg = replay(&schema, "x", &events).unwrap();
|
|
assert_eq!(agg.state, "a");
|
|
assert_eq!(agg.sequence, 2);
|
|
}
|
|
|
|
#[test]
|
|
fn disallowed_edge_is_skipped_not_fatal() {
|
|
// "a" then "b" - b isn't reachable from a, so the second event
|
|
// is silently ignored, not an error.
|
|
let schema = test_schema();
|
|
let events = vec![event(1, "opened"), event(2, "a"), event(3, "b")];
|
|
let agg = replay(&schema, "x", &events).unwrap();
|
|
assert_eq!(agg.state, "a");
|
|
}
|
|
|
|
#[test]
|
|
fn unrecognized_event_type_is_skipped() {
|
|
let schema = test_schema();
|
|
let events = vec![event(1, "opened"), event(2, "bogus"), event(3, "a")];
|
|
let agg = replay(&schema, "x", &events).unwrap();
|
|
assert_eq!(agg.state, "a");
|
|
}
|
|
|
|
#[test]
|
|
fn initial_state_not_declared_is_rejected() {
|
|
let err = parse_aggregates_yaml(
|
|
r#"
|
|
aggregates:
|
|
- bucket: bad
|
|
initial: missing
|
|
states:
|
|
open: { event: opened }
|
|
"#,
|
|
)
|
|
.unwrap_err();
|
|
assert!(err.to_string().contains("initial state"));
|
|
}
|
|
|
|
#[test]
|
|
fn transition_target_not_declared_is_rejected() {
|
|
let err = parse_aggregates_yaml(
|
|
r#"
|
|
aggregates:
|
|
- bucket: bad
|
|
initial: open
|
|
states:
|
|
open: { event: opened }
|
|
transitions:
|
|
open: [nonexistent]
|
|
"#,
|
|
)
|
|
.unwrap_err();
|
|
assert!(err.to_string().contains("undeclared state"));
|
|
}
|
|
|
|
#[test]
|
|
fn duplicate_event_type_is_rejected() {
|
|
let err = parse_aggregates_yaml(
|
|
r#"
|
|
aggregates:
|
|
- bucket: bad
|
|
initial: open
|
|
states:
|
|
open: { event: same }
|
|
other: { event: same }
|
|
"#,
|
|
)
|
|
.unwrap_err();
|
|
assert!(err.to_string().contains("used by both"));
|
|
}
|
|
|
|
proptest::proptest! {
|
|
#[test]
|
|
fn replay_never_panics_and_state_always_valid(
|
|
types in proptest::collection::vec(
|
|
proptest::sample::select(vec!["opened", "a", "b", "bogus"]),
|
|
0..30,
|
|
)
|
|
) {
|
|
let schema = test_schema();
|
|
let events: Vec<Event> = types
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(i, t)| event(i as u64, t))
|
|
.collect();
|
|
let result = replay(&schema, "x", &events);
|
|
if let Some(agg) = result {
|
|
assert!(schema.has_state(&agg.state));
|
|
assert!(agg.sequence <= events.len() as u64);
|
|
}
|
|
}
|
|
}
|
|
}
|