kuilt-raft-test/us.tractat.kuilt.raft.test/MultiNodeRaftSim

MultiNodeRaftSim

class MultiNodeRaftSim(val nodeIds: List<NodeId>, scope: CoroutineScope, nodeScope: CoroutineScope, baseConfig: RaftConfig = MULTI_NODE_SIM_BASE_CONFIG, baseSeed: Long = MULTI_NODE_SIM_SEED, maxPayloadBytes: Int? = null, nodeFactory: (NodeId, RaftTransport, RaftStorage, CoroutineScope) -> RaftNode = defaultNodeFactory(nodeIds, baseConfig, baseSeed))

Multi-node Raft simulation harness for use in tests. See the file-level KDoc for the full determinism contract and setup ceremony.

Parameters

nodeIds

The ordered list of voter NodeIds forming the initial cluster config.

scope

The test's TestScope — used for withTimeout bounds in the await helpers.

nodeScope

Scope for node coroutines — pass TestScope.backgroundScope so the infinite heartbeat/election loops are cancelled at test teardown without kotlinx.coroutines.test.UncompletedCoroutinesError.

baseConfig

Timing parameters for every node. RaftConfig.random is overridden per node (Random(baseSeed + nodeIndex)) to ensure distinct election timeouts and guarantee convergence. Must have RaftConfig.expectVirtualTime = true for tests running under a kotlinx.coroutines.test.TestDispatcher. Defaults to MULTI_NODE_SIM_BASE_CONFIG.

baseSeed

Seed from which per-node Randoms are derived. Change to explore different election orderings; the default MULTI_NODE_SIM_SEED is stable across runs.

maxPayloadBytes

Optional payload cap forwarded to MultiNodeRaftNetwork — forces InstallSnapshot chunking in tests that exercise the snapshot-transfer path.

nodeFactory

Override to wire a custom RaftNode implementation. The default wires CoroutineScope.raftNode with a per-node seeded config. Parameters: (id, transport, storage, childScope).

Constructors

MultiNodeRaftSim

constructor(nodeIds: List<NodeId>, scope: CoroutineScope, nodeScope: CoroutineScope, baseConfig: RaftConfig = MULTI_NODE_SIM_BASE_CONFIG, baseSeed: Long = MULTI_NODE_SIM_SEED, maxPayloadBytes: Int? = null, nodeFactory: (NodeId, RaftTransport, RaftStorage, CoroutineScope) -> RaftNode = defaultNodeFactory(nodeIds, baseConfig, baseSeed))

Properties

network

val network: MultiNodeRaftNetwork

The in-process channel network shared by all nodes. Use to inject partitions or raw messages.

nodeIds

val nodeIds: List<NodeId>

nodes

val nodes: MutableMap<NodeId, RaftNode>

Live node map — entries are removed on crash and re-added on restart.

storages

val storages: Map<NodeId, InMemoryRaftStorage>

Per-node durable storage — each node's log, term, and vote.

Functions

appliedState

fun appliedState(id: NodeId): ByteArray

The applied state for id — ordered concatenation of committed commands, reset on a snapshot install. Two nodes with equal applied state have applied the same sequence of commands.

awaitCommit

suspend fun awaitCommit(index: Long, on: Collection<NodeId> = nodeIds, within: Duration = DEFAULT_AWAIT)

Suspend until index is committed on every node in on; fail fast with a state dump otherwise.

awaitLeader

suspend fun awaitLeader(within: Duration = DEFAULT_AWAIT): RaftNode

Suspend until some node is RaftRole.Leader; return it, or fail fast with a state dump.

suspend fun awaitLeader(among: Set<NodeId>, within: Duration = DEFAULT_AWAIT): RaftNode

Suspend until a node whose id is in among is RaftRole.Leader. Use after a partition when a stale leader from the minority partition may transiently report leader status — scope the await to the surviving voters.

awaitRole

suspend fun awaitRole(id: NodeId, role: RaftRole, within: Duration = DEFAULT_AWAIT)

Suspend until node id holds role; fail fast with a state dump otherwise.

awaitTrue

suspend fun awaitTrue(what: String, within: Duration = DEFAULT_AWAIT, cond: () -> Boolean)

Suspend until cond holds (polled each virtual ms); fail fast with a state dump on timeout.

checkInvariants

suspend fun checkInvariants()

Assert election safety (at most one leader at a time) and state-machine safety (no two nodes committed different commands at the same log index up to the minimum known commit).

crash

fun crash(id: NodeId)

Cancel node id's coroutine scope, simulating a crash. Use restart to bring it back up.

dropLink

fun dropLink(from: NodeId, to: NodeId)

Drop messages from from to to (unidirectional). Restore with heal.

dumpState

suspend fun dumpState(reason: String): String

Render a per-node diagnostic snapshot — roles, terms, commit indices, log ranges, and an election-event histogram (Timeout / BecomeLeader / BecomeFollower) that makes leadership thrash and term inflation visible at a glance. Used as the body of the AssertionError thrown by the bounded await helpers on non-convergence, and callable directly from a failing test assertion.

followers

fun followers(): List<RaftNode>

All nodes currently reporting RaftRole.Follower.

heal

fun heal()

Restore all links cleared by partition or dropLink.

leader

fun leader(): RaftNode?

The current leader, or null if none is known.

partition

fun partition(a: Set<NodeId>, b: Set<NodeId>)

Partition nodes into two groups — messages in both directions between any node in a and any node in b are silently dropped. Restore with heal.

partitionOff

fun partitionOff(id: NodeId)

Isolate id from every other node — the offline-follower scenario. Restore with heal.

proposeOnLeader

suspend fun proposeOnLeader(command: ByteArray, among: Set<NodeId>? = null, within: Duration = DEFAULT_AWAIT): LogEntry

Propose against the current leader, re-acquiring and retrying on NotLeaderException. Returns the committed us.tractat.kuilt.raft.LogEntry. Restricts to among if provided.

restart

fun restart(id: NodeId)

Restart a previously crashed node using its durable storages entry.

settle

suspend fun settle()

Let pending work at the current virtual instant drain without advancing the clock. Under StandardTestDispatcher's FIFO scheduling, yielding hands the single test thread back so already-scheduled coroutines (e.g. a freshly launched collector) run at this instant. Call after launching a collector you need to observe the next emission.