Pure Stage: Actor Model for deterministic scheduling

Design goals:

• no side effects, only explicit effects (which does include internal state changes)
• easy to use, including the ability to hold references across effects
• executable on concurrent thread pool or using a deterministic simulator
• fully back-pressured
• ability for biased reading from inputs (which is necessary to avoid deadlocks with back-pressure)
• wiring code should be nicely readable

Design elements

We will need to model state machines, processing network nodes, and wiring between those nodes.

State Machines

Writing state machines in Rust can be done explicitly using an enum that implements a trait for computing state × input → state × output. One advantage of this approach is that infrastructure like a (virtual) clock can be passed into the transition function as well. The main disadvantage is that additional states are needed for modelling (biased) input operations, complicating the state machine code.

Another syntactic means to this end is an async fn, which gets translated by the compiler into a state machine like the one above, albeit with a limited transition function: the Future::poll() method. When using this approach, effects like input and output need to be offered as Futures; the programmer may also await other Futures which perform untracked effects. Such abuse can only be rejected using runtime checks because the return type of an .await point cannot be constrained.

The downside of the second approach is that the internal state of a stage would be inaccessible, it is wrapped in the opaque Future generated by the compiler. One way to fix this is to compromise: offer all effects apart from receiving input as Futures and thus have the programmer write an explicit state machine that only needs to switch for inputs from upstream, not results from other effects. Since logging state progression is a rather important debugging tool, we are going for this variant.

Processing Nodes

The main purpose of an API for declaring processing nodes is to obtain type-level information on the connectivity expected by this node, i.e. the input message type, the typed outputs, and the internally handled state (for inspection from the outside during tests). Whether the state machine inside is implemented explicitly or using async fn doesn't matter at this level.

Wiring

While Rust can in principle model type-state to track what has already been wired, this is inconvenient in practice because it requires shadowing and thus restricts the code a programmer can write. Therefore, a compromise would be to establish that all processing nodes are declared first, followed by the declaration of the wiring. The wiring function ensures that message types do match, but it won't prevent connecting the same output to multiple inputs or multiple outputs to the same input; in fact, this freedom is quite desirable.

Future Work

• add deterministic shuffling of effect scheduling in the Simulation StageGraph implementation
• output an execution log for debugging that notes all effects and state changes (not yet done for Tokio)
• add stage error handling (currently a stage just stops working)

Thoughts on World State

A stage's state may well refer to some external real world resource like a database or a network connection; these cannot be serialized nor deserialized. Leaving these resources outside of the stage means manipulating them only via ExternalEffects, which keeps the code and the responsibilities cleanly separated: when starting the simulation, effect handlers for these external effects need to be installed that match the world state needed for the current simulation run. During a replay this means recreating the state of databases or (simulated) network connections such as to match the application state when the trace was recorded. When starting a StageGraph not at the beginning of time, it means opening latest stored databases and re-establishing persistent network connections. The ability to do this depends on recording the expected world state in snapshots (where snapshots are taken at some defined schedule or cadence).

The alternative is to move the resources into the stage state, which means that serialization records a brief declaration and deserialization recreates the resource based on that declaration (e.g. a database path and snapshot name). One important caveat is that stage states are deserialized independently for each stage, meaning that for example a database connection is maintained once per process in shared global mutable state or that each stage has a separate database connection. While the latter may work for some database drivers, it is not an option for resources that are intended to be shared between stages. A possible solution to this problem would be to add a per-StageGraph resource registry that deserialization code accesses using some kind of typed key objects. This means that deserialization needs to be parameterized, which is not foreseen in serde and thus implies using ThreadLocal or similar hacks.

This reasoning leads me to the conclusion that moving resources out of the stages and manipulating them through external effects only is the more robust and more flexible approach: we control the API for evaluating external effects and can thus include access to a resource registry, which can then contain values that are either per-StageGraph or even shared between a selection of StageGraphs.

Notes on TraceBuffer and Serialization

Storing all effects and responses requires serialization (because trait objects cannot implement Clone), which in turn requires some gymnastics due to the incompatibility of serde's Serialize and Deserialize traits with trait objects. While typetag can solve the serialization issue (via erased_serde::Serialize), deserialization will always require the full and concrete target type to be known plus matching deserialization code. The solution typetag provides therefore cannot support generic types, which closes the door on any kind of type parameter occurring within messages, states, and effects in the system. This is clearly a restriction that weighs too heavily, thus a different solution is required.

The current design shifts all deserialization to places where the concrete target type can be named and where thus the compiler can summon a Deserialize instance. This unfortunately requires some trace entries to be deserialized multiple times, with information for the simulation machinery being done in the generic parts and then the specific application data type (for messages or effect responses) to be deserialized from the generic cbor4ii::core::Value on the application side of the airlock; let us give thanks to the universe for schemaless deserialization in this context.