///| The state of one simulated replica. The simulator is intentionally
///|
/// single-threaded and deterministic: callers control time with `advance_to`.
pub(all) struct ReplicaState {
id : String
clock : HlcClock
vector : VersionVector
online : Bool
} derive(Debug)
///|
/// A message that has been sent but not necessarily delivered.
pub(all) struct PendingMessage {
source : String
target : String
payload : String
sent_at : Int
deliver_at : Int
stamp : HlcTimestamp
context : VersionVector
} derive(Debug)
///|
/// The kind of a trace event emitted by the deterministic simulator.
pub(all) enum TraceKind {
Local(label~ : String)
Sent(target~ : String, payload~ : String)
Received(source~ : String, payload~ : String)
AvailabilityChanged(online~ : Bool)
} derive(Eq, Debug)
///| A causally annotated event. `context` is the version vector after the
///|
/// event, while `stamp` is the HLC value used for stable total ordering.
pub(all) struct TraceEvent {
sequence : Int
time : Int
replica : String
stamp : HlcTimestamp
context : VersionVector
kind : TraceKind
} derive(Debug)
///| A deterministic in-memory transport and replica model. It is a test and
///|
/// teaching tool, not a production network stack.
pub(all) struct Simulator {
mut time : Int
mut replicas : Array[ReplicaState]
mut pending : Array[PendingMessage]
mut trace : Array[TraceEvent]
} derive(Debug)
///|
/// Create a network of replica IDs. IDs must be nonempty and unique.
pub fn Simulator::new(
replica_ids : Array[String],
start_time : Int,
) -> Result[Simulator, SimulatorError] {
if start_time < 0 {
return Err(NegativeSimulationTime(start_time))
}
let replicas : Array[ReplicaState] = []
for id in replica_ids {
if id.length() == 0 {
return Err(EmptyReplicaId)
}
for replica in replicas {
if replica.id == id {
return Err(DuplicateReplica(id))
}
}
match HlcClock::new(id, start_time) {
Ok(clock) =>
replicas.push({ id, clock, vector: VersionVector::new(), online: true })
Err(_) => return Err(NegativeSimulationTime(start_time))
}
}
Ok({ time: start_time, replicas, pending: [], trace: [] })
}
///|
/// Return the simulator's caller-controlled physical time.
pub fn Simulator::time(self : Simulator) -> Int {
self.time
}
///|
/// Return a snapshot of replica state for display or assertions.
pub fn Simulator::replicas(self : Simulator) -> Array[ReplicaState] {
self.replicas
}
///| Return queued messages which are waiting for their target to become online
///|
/// or for the controlled clock to reach their delivery time.
pub fn Simulator::pending(self : Simulator) -> Array[PendingMessage] {
self.pending
}
///|
/// Return the append-only event trace.
pub fn Simulator::trace(self : Simulator) -> Array[TraceEvent] {
self.trace
}
///| Emit one local event. Offline replicas reject events rather than silently
///|
/// accepting writes which could make a partition test misleading.
pub fn Simulator::local_event(
self : Simulator,
replica_id : String,
label : String,
) -> Result[TraceEvent, SimulatorError] {
let index = match self.replica_index(replica_id) {
Ok(value) => value
Err(error) => return Err(error)
}
let replica = self.replicas[index]
if !replica.online {
return Err(ReplicaOffline(replica_id))
}
let stamp = replica.clock.tick(self.time)
let context = match replica.vector.increment(replica.id) {
Ok(vector) => vector
Err(_) => return Err(EmptyReplicaId)
}
self.replicas[index] = {
id: replica.id,
clock: replica.clock,
vector: context,
online: replica.online,
}
Ok(self.record(replica.id, stamp, context, Local(label~)))
}
///| Send a causally annotated message. Sending is a local event; delivery is
///|
/// delayed until `advance_to` reaches `deliver_at`.
pub fn Simulator::send(
self : Simulator,
source : String,
target : String,
payload : String,
delay : Int,
) -> Result[TraceEvent, SimulatorError] {
if delay < 0 {
return Err(NegativeDelay(delay))
}
let source_index = match self.replica_index(source) {
Ok(value) => value
Err(error) => return Err(error)
}
match self.replica_index(target) {
Ok(_) => ()
Err(error) => return Err(error)
}
let replica = self.replicas[source_index]
if !replica.online {
return Err(ReplicaOffline(source))
}
let stamp = replica.clock.tick(self.time)
let context = match replica.vector.increment(replica.id) {
Ok(vector) => vector
Err(_) => return Err(EmptyReplicaId)
}
self.replicas[source_index] = {
id: replica.id,
clock: replica.clock,
vector: context,
online: replica.online,
}
self.pending.push({
source: replica.id,
target,
payload,
sent_at: self.time,
deliver_at: self.time + delay,
stamp,
context,
})
Ok(self.record(replica.id, stamp, context, Sent(target~, payload~)))
}
///| Move simulated time forward and deliver every due message whose target is
///| online. Due messages for an offline target remain queued, modelling a
///|
/// recoverable network partition without inventing a separate transport API.
pub fn Simulator::advance_to(
self : Simulator,
next_time : Int,
) -> Result[Array[TraceEvent], SimulatorError] {
if next_time < self.time {
return Err(TimeWentBackwards(current=self.time, requested=next_time))
}
self.time = next_time
let delivered : Array[TraceEvent] = []
let remaining : Array[PendingMessage] = []
for message in self.pending {
let target_index = match self.replica_index(message.target) {
Ok(value) => value
Err(error) => return Err(error)
}
let target = self.replicas[target_index]
if message.deliver_at <= self.time && target.online {
let stamp = target.clock.receive(message.stamp, self.time)
let merged = target.vector.merge(message.context)
let context = match merged.increment(target.id) {
Ok(vector) => vector
Err(_) => return Err(EmptyReplicaId)
}
self.replicas[target_index] = {
id: target.id,
clock: target.clock,
vector: context,
online: target.online,
}
let event = self.record(
target.id,
stamp,
context,
Received(source=message.source, payload=message.payload),
)
delivered.push(event)
} else {
remaining.push(message)
}
}
self.pending = remaining
Ok(delivered)
}
///| Change whether a replica accepts local events and delivered messages. The
///|
/// transition itself is traced so a replay explains why a message was held.
pub fn Simulator::set_online(
self : Simulator,
replica_id : String,
online : Bool,
) -> Result[TraceEvent, SimulatorError] {
let index = match self.replica_index(replica_id) {
Ok(value) => value
Err(error) => return Err(error)
}
let replica = self.replicas[index]
let stamp = replica.clock.tick(self.time)
let context = match replica.vector.increment(replica.id) {
Ok(vector) => vector
Err(_) => return Err(EmptyReplicaId)
}
self.replicas[index] = {
id: replica.id,
clock: replica.clock,
vector: context,
online,
}
Ok(self.record(replica.id, stamp, context, AvailabilityChanged(online~)))
}
///|
/// Look up a replica's current causal vector.
pub fn Simulator::vector_of(
self : Simulator,
replica_id : String,
) -> Result[VersionVector, SimulatorError] {
let index = match self.replica_index(replica_id) {
Ok(value) => value
Err(error) => return Err(error)
}
Ok(self.replicas[index].vector)
}
///|
/// Find a replica index once, keeping public APIs descriptive on bad IDs.
fn Simulator::replica_index(
self : Simulator,
replica_id : String,
) -> Result[Int, SimulatorError] {
for index in 0.. TraceEvent {
let event = {
sequence: self.trace.length() + 1,
time: self.time,
replica,
stamp,
context,
kind,
}
self.trace.push(event)
event
}
///|
/// Errors exposed by simulator operations.
pub(all) enum SimulatorError {
EmptyReplicaId
DuplicateReplica(String)
UnknownReplica(String)
ReplicaOffline(String)
NegativeSimulationTime(Int)
NegativeDelay(Int)
TimeWentBackwards(current~ : Int, requested~ : Int)
} derive(Eq, Debug)