SwimDriver

SwimDriver: runtime I/O glue for the pure Swim state machine (P2 L3).

Owns the RNG and period timer so the transport layer just calls step(state, now, events) each tick. The driver picks probe targets and indirect helpers randomly, advances the Swim FSM, and returns the Actions the transport must perform (send ping, send ping-req, gossip). The transport reads frames, decodes them into Events, and calls step again.

Wire protocol (Msgpack array, first element is the tag): 0 SwimPing(sender_id) 1 SwimPingAck(sender_id) 2 SwimPingReq(requester_id, target_id) 3 SwimGossip(node_id, status_int, incarnation) status_int: 0=Alive, 1=Suspect, 2=Dead 3 SwimGossipLoad(node_id, status_int, incarnation, [load_ints]) 5-element frame: the 5th element is a Msgpack array of NodeLoad ints (same tag 3, distinguished by element count)

Period lifecycle (period_ms ≈ 1000–5000 ms is typical):

step() at period boundary: end prior probe → expire suspects →

   refute if suspected → begin new period (pick random target).
2. step() mid-period: if ack_timeout has elapsed and probe is unacked,
   escalate to k-way indirect ping-req.
3. Transport delivers PingAck events when acks arrive; GossipFrame events
   when gossip frames are decoded; PeerDown events on socket close.

Anti-entropy (P5a): anti_entropy_peers(s, now) returns live peers when the 30s timer fires. The transport calls this each tick and sends REGISTRY_SYNC_REQ to peers. bump_anti_entropy(s, now) resets the timer after firing.

Load gossip (P4): peer_loads tracks the last NodeLoad heard from each live peer, stamped with its receipt time on this node's clock (never the sender's). dispatch_all_with_load embeds the local NodeLoad in outgoing Gossip frames.

Types

typeStateState = {#

typeEventEvent =#

Events delivered from the transport layer each tick.

typeWireMsgWireMsg =#

Wire messages.

Functions

fnaction_framesaction_frames(action : Swim.Action, my_id : String, peer_ids : List(String)) : List((String, List(Int)))#

The frames one Swim.Action turns into, as (destination node_id, payload) pairs, for a transport that writes through its own per-peer writers instead of dispatch's direct fd writes. A ping or ping-req yields one pair; a Gossip yields one full-member MEMBER_GOSSIP frame per id in [peer_ids] (no pairs when [peer_ids] is empty).

fnanti_entropy_peersanti_entropy_peers(s : State, now : Int) : List(String)#

The Alive peers other than ourselves (the anti-entropy targets) when the anti-entropy timer has fired at [now], else Nil. The transport calls this each tick and sends a registry sync request to each returned peer, then calls bump_anti_entropy.

fnbump_anti_entropybump_anti_entropy(s : State, now : Int) : State#

Re-arm the anti-entropy timer for 30 s after [now]; call it right after anti_entropy_peers returned a non-empty list.

fndecode_member_gossipdecode_member_gossip(bytes : List(Int)) : Result(Member, String)#

Decode a MEMBER_GOSSIP frame. Returns Err for invalid MessagePack, a frame with another tag, a malformed member record, or a bad status int.

fndecode_members_syncdecode_members_sync(bytes : List(Int)) : Result(List(Member), String)#

Decode a MEMBERS_SYNC frame into its members, in frame order. A malformed record is skipped, not fatal, so the rest of the view still merges. Returns Err only for invalid MessagePack or a frame that is not [14, [records...]].

fndecode_msgdecode_msg(bytes : List(Int)) : Result(WireMsg, String)#

Decode bytes produced by encode_msg. Returns Err for invalid MessagePack, a non-array, an unknown tag, the wrong element count or types for a tag, or a status int outside 0..2. In a tag-3 load frame, non-integer load elements are silently dropped before ClusterLoad.from_ints.

fndispatchdispatch(action : Swim.Action, peers : PeerRegistry.Registry, my_id : String) : ()#

Execute one Swim.Action by writing the wire frame straight to the peer's fd from [peers]: a ping or ping-req to its one destination, a Gossip to every registered peer as a legacy tag-3 frame (status and incarnation only; the member's name, addr and creation are not sent). Destinations missing from [peers] and send errors are silently ignored.

fndispatch_alldispatch_all(actions : List(Swim.Action), peers : PeerRegistry.Registry, my_id : String) : ()#

Execute each action in order with dispatch.

fndispatch_all_with_loaddispatch_all_with_load(actions : List(Swim.Action), peers : PeerRegistry.Registry,#

Execute each action in order with dispatch_with_load.

fndispatch_with_loaddispatch_with_load(action : Swim.Action, peers : PeerRegistry.Registry,#

Like dispatch, but a Gossip is sent as a tag-3 load frame that piggybacks [my_load].

fnencode_member_gossipencode_member_gossip(m : Member) : List(Int)#

Encode one member with its origin fields as a MEMBER_GOSSIP frame: [13, node_id, name, addr, creation, status_int, incarnation].

fnencode_members_syncencode_members_sync(ms : List(Member)) : List(Int)#

Encode a whole member view as a MEMBERS_SYNC frame: [14, [[node_id, name, addr, creation, status_int, incarnation], ...]]. Exchanged both ways on every new connection (push-pull join and heal).

fnencode_msgencode_msg(msg : WireMsg) : List(Int)#

Encode a wire message as a MessagePack array whose first element is its tag (0 ping, 1 ack, 2 ping-req, 3 gossip; SwimGossipLoad is tag 3 with a fifth element holding the ClusterLoad.to_ints of its load).

fnfresh_peer_loadsfresh_peer_loads(s : State, now : Int) : List(NodeLoad)#

Every peer load sample received at most 10 s before [now], in unspecified order.

fnmakemake(me : String, incarnation : Int, members : Members, swim_cfg : Swim.Config,#

A driver for node [me] wrapping a fresh Swim.make(me, incarnation, members, swim_cfg), with an RNG seeded from [seed] and a protocol period of [period_ms]. The period timer and the anti-entropy timer both start at 0, so the first step begins a period and the first anti_entropy_peers fires.

fnmsg_to_eventmsg_to_event(msg : WireMsg) : Option(Event)#

The driver Event for a decoded wire message: an ack becomes PingAck, a tag-3 gossip becomes GossipFrame/GossipWithLoad with an origin-less member (creation 0, name = node id, addr ""). Returns None for SwimPing and SwimPingReq, which the transport answers directly.

fnpeer_loadpeer_load(s : State, node_id : String, now : Int) : Option(NodeLoad)#

The last load sample heard from [node_id], or None if none was heard or it is more than 10 s old at [now] (receipt time on this node's clock).

fnstepstep(s : State, now : Int, events : List(Event)) : (State, List(Swim.Action))#

Advance the driver one tick at [now] (Unix ms), folding in the transport [events] received since the last call. Returns the updated state and the actions to execute, in order: the actions the events produced (a PeerDown gossips a Dead verdict, a ConnLost gossips a Suspect), then:

at a period boundary (now >= period_end): end the previous probe (may

suspect its target), expire overdue suspects to Dead, refute if we are
suspected, then ping a randomly chosen Alive peer (no ping when there is
none) and roll the period timer to `now + period_ms`;

mid-period: escalate the probe to indirect ping-reqs through up to

`indirect_k` randomly sampled helpers once the ack timeout has passed.

GossipWithLoad samples are stored in the peer-load table restamped with [now]. Deterministic for a given seed and event sequence.

Examples

let d = SwimDriver.make("n1", 0, view, Swim.config(500, 5000, 3), 1000, 42)
match SwimDriver.step(d, now, Nil) do      -- first call begins a period
  (d2, acts) -> SwimDriver.dispatch_all(acts, peers, "n1")
end

fntag_member_gossiptag_member_gossip() : Int do 13 end#

The wire tag of a full-member MEMBER_GOSSIP frame (13).

fntag_members_synctag_members_sync() : Int do 14 end#

The wire tag of a MEMBERS_SYNC frame carrying a whole member view (14).