///|
/// Built-in event categories for deterministic model traces.
pub enum EventKind {
Message
Task
Timer
StateTransition
ExternalCall
}
///|
pub fn EventKind::label(self : EventKind) -> String {
match self {
Message => "message"
Task => "task"
Timer => "timer"
StateTransition => "state_transition"
ExternalCall => "external_call"
}
}
///|
pub fn message_event_kind() -> EventKind {
Message
}
///|
pub fn task_event_kind() -> EventKind {
Task
}
///|
pub fn timer_event_kind() -> EventKind {
Timer
}
///|
pub fn state_transition_event_kind() -> EventKind {
StateTransition
}
///|
pub fn external_call_event_kind() -> EventKind {
ExternalCall
}
///|
pub(all) struct EventRecord {
id : Int
kind : EventKind
tick : Int
priority : Int
correlation_id : String
source : String
target : String
label : String
parent_id : Int
payload : String
dropped : Bool
failed : Bool
}
///|
pub fn event_record(
kind : EventKind,
tick : Int,
label : String,
correlation_id? : String = "",
source? : String = "",
target? : String = "",
parent_id? : Int = 0,
payload? : String = "",
priority? : Int = 0,
dropped? : Bool = false,
failed? : Bool = false,
) -> EventRecord {
{
id: 0,
kind,
tick,
priority,
correlation_id,
source,
target,
label,
parent_id,
payload,
dropped,
failed,
}
}
///|
pub(all) struct EventMutationPolicy {
seed : UInt64
max_delay : Int
delay_percent : Int
drop_percent : Int
duplicate_percent : Int
reorder_same_tick : Bool
failure_percent : Int
}
///|
pub fn event_mutation_policy(
seed? : UInt64 = 1UL,
max_delay? : Int = 0,
delay_percent? : Int = 0,
drop_percent? : Int = 0,
duplicate_percent? : Int = 0,
reorder_same_tick? : Bool = false,
failure_percent? : Int = 0,
) -> EventMutationPolicy {
{
seed,
max_delay: if max_delay < 0 {
0
} else {
max_delay
},
delay_percent: event_percent(delay_percent),
drop_percent: event_percent(drop_percent),
duplicate_percent: event_percent(duplicate_percent),
reorder_same_tick,
failure_percent: event_percent(failure_percent),
}
}
///|
pub(all) struct EventStreamSnapshot {
next_id : Int
events : Array[EventRecord]
}
///|
pub(all) struct EventReplayResult {
seed : UInt64
digest : UInt64
events : Array[EventRecord]
scheduled : Int
dropped : Int
duplicated : Int
failed : Int
}
///|
pub(all) struct EventFailureCase {
seed : UInt64
rule : String
digest : UInt64
events : Array[EventRecord]
policy : EventMutationPolicy
source_events : Array[EventRecord]
}
///|
pub(all) struct EventStream {
mut next_id : Int
mut events : Array[EventRecord]
}
///|
pub fn EventStream::new() -> EventStream {
{ next_id: 1, events: [] }
}
///|
pub fn EventStream::append(
self : EventStream,
event : EventRecord,
) -> EventRecord {
let id = if event.id <= 0 { self.next_id } else { event.id }
if id >= self.next_id {
self.next_id = id + 1
}
let stored = {
id,
kind: event.kind,
tick: event.tick,
priority: event.priority,
correlation_id: event.correlation_id,
source: event.source,
target: event.target,
label: event.label,
parent_id: event.parent_id,
payload: event.payload,
dropped: event.dropped,
failed: event.failed,
}
self.events.push(stored)
stored
}
///|
pub fn EventStream::record(
self : EventStream,
kind : EventKind,
tick : Int,
label : String,
correlation_id? : String = "",
source? : String = "",
target? : String = "",
parent_id? : Int = 0,
payload? : String = "",
priority? : Int = 0,
dropped? : Bool = false,
failed? : Bool = false,
) -> EventRecord {
self.append(
event_record(
kind,
tick,
label,
correlation_id~,
source~,
target~,
parent_id~,
payload~,
priority~,
dropped~,
failed~,
),
)
}
///|
pub fn EventStream::events(self : EventStream) -> Array[EventRecord] {
self.events.copy()
}
///|
pub fn EventStream::ordered(self : EventStream) -> Array[EventRecord] {
let ordered = self.events.copy()
ordered.sort_by(fn(left, right) {
if left.tick != right.tick {
left.tick.compare(right.tick)
} else if left.priority != right.priority {
left.priority.compare(right.priority)
} else {
left.id.compare(right.id)
}
})
ordered
}
///|
pub fn EventStream::snapshot(self : EventStream) -> EventStreamSnapshot {
{ next_id: self.next_id, events: self.events.copy() }
}
///|
pub fn EventStream::restore(
self : EventStream,
snapshot : EventStreamSnapshot,
) -> Unit {
self.next_id = snapshot.next_id
self.events = snapshot.events.copy()
}
///|
pub fn EventStream::digest(self : EventStream) -> UInt64 {
event_records_digest(self.ordered())
}
///|
pub fn EventStream::invariants(self : EventStream) -> InvariantReport {
event_records_invariants(self.ordered())
}
///|
pub fn EventStream::replay(
self : EventStream,
policy : EventMutationPolicy,
) -> EventReplayResult {
let rng = Rng::new(policy.seed)
let replayed = EventStream::new()
let mut dropped = 0
let mut duplicated = 0
let mut failed = 0
for event in self.ordered() {
let delay = if policy.max_delay > 0 && event_roll(rng, policy.delay_percent) {
rng.next_int(policy.max_delay + 1)
} else {
0
}
let is_dropped = event.dropped || event_roll(rng, policy.drop_percent)
let is_failed = event.failed || event_roll(rng, policy.failure_percent)
let priority = if policy.reorder_same_tick {
-event.priority - event.id
} else {
event.priority
}
let replay_event = {
id: event.id,
kind: event.kind,
tick: event.tick + delay,
priority,
correlation_id: event.correlation_id,
source: event.source,
target: event.target,
label: event.label,
parent_id: event.parent_id,
payload: event.payload,
dropped: is_dropped,
failed: is_failed,
}
ignore(replayed.append(replay_event))
if is_dropped {
dropped += 1
}
if is_failed {
failed += 1
}
if event_roll(rng, policy.duplicate_percent) {
ignore(
replayed.record(
event.kind,
event.tick + delay,
event.label + ".duplicate",
correlation_id=event.correlation_id,
source=event.source,
target=event.target,
parent_id=event.id,
payload=event.payload,
priority~,
dropped=is_dropped,
failed=is_failed,
),
)
duplicated += 1
}
}
let events = replayed.ordered()
{
seed: policy.seed,
digest: event_records_digest(events),
scheduled: events.length(),
events,
dropped,
duplicated,
failed,
}
}
///|
pub fn EventReplayResult::invariants(
self : EventReplayResult,
) -> InvariantReport {
event_records_invariants(self.events)
}
///|
pub fn EventReplayResult::matches_digest(
self : EventReplayResult,
other : EventReplayResult,
) -> Bool {
self.digest == other.digest
}
///|
pub fn event_failure_case(
rule : String,
source : EventStream,
policy : EventMutationPolicy,
) -> EventFailureCase {
let replay = source.replay(policy)
{
seed: replay.seed,
rule,
digest: replay.digest,
events: replay.events,
policy,
source_events: source.events(),
}
}
///|
pub fn EventFailureCase::replay(self : EventFailureCase) -> EventReplayResult {
let stream = EventStream::new()
for event in self.source_events {
ignore(stream.append(event))
}
stream.replay(self.policy)
}
///|
fn event_percent(value : Int) -> Int {
if value < 0 {
0
} else if value > 100 {
100
} else {
value
}
}
///|
fn event_roll(rng : Rng, percent : Int) -> Bool {
percent > 0 && rng.next_int(100) < percent
}
///|
fn event_records_digest(events : Array[EventRecord]) -> UInt64 {
let entries : Array[TraceEntry] = []
for event in events {
entries.push(
trace_entry(
event.tick,
event.id,
event.kind.label(),
event.correlation_id +
"|" +
event.source +
"|" +
event.target +
"|" +
event.label +
"|" +
event.parent_id.to_string() +
"|" +
event.payload +
"|" +
event.dropped.to_string() +
"|" +
event.failed.to_string(),
),
)
}
trace_digest(entries)
}
///|
fn event_records_invariants(events : Array[EventRecord]) -> InvariantReport {
let report = InvariantReport::new("event-stream")
let ids : Array[Int] = []
let mut unique_ids = true
let mut parent_exists = true
let mut causal_tick_order = true
for event in events {
for known_id in ids {
if known_id == event.id {
unique_ids = false
}
}
if event.parent_id > 0 {
let mut found_parent = false
for parent in events {
if parent.id == event.parent_id {
found_parent = true
if parent.tick > event.tick {
causal_tick_order = false
}
}
}
if !found_parent {
parent_exists = false
}
}
ids.push(event.id)
}
ignore(report.add(invariant_check("event_id_unique", unique_ids)))
ignore(report.add(invariant_check("causal_parent_exists", parent_exists)))
ignore(report.add(invariant_check("causal_tick_order", causal_tick_order)))
report
}